diff --git a/docs/architecture/task-execution-event-storage.md b/docs/architecture/task-execution-event-storage.md index 6a9afa441..0a10c3dab 100644 --- a/docs/architecture/task-execution-event-storage.md +++ b/docs/architecture/task-execution-event-storage.md @@ -1,58 +1,62 @@ -# Execution-event storage: migration stage 3.1 - -This stage adds storage primitives only. No Web, channel, runner, checkpoint, -or historical-reader path writes or reads these events in production yet. - -## Version boundary - -`tasks.conversation_storage_version` defaults to `1` (legacy), both in the -ORM and in SQL. A database CHECK pins it to `1`: the event-backed runtime is -not available in this release. Activating another version requires a later -migration **and** the complete writer/reader routing from stages 3.2–3.3. -This is a conversation-storage version, not an agent runtime version. - -Existing rows and inserts from older applications retain legacy behavior. -The migration adds columns with inline CHECKs, avoiding a SQLite tasks-table -rebuild and its inbound foreign-key hazards. PostgreSQL still takes an ALTER -TABLE lock; rollout needs a short lock window, not a zero-lock assumption. -The new indexes apply only to the new, initially empty event table. - -## Event envelope - -`task_execution_events` records one fact with: - -- A generated stable `event_id` and a task-local `sequence`. -- An explicit nonempty `scope_id` (`root` or a stable child execution scope). -- Optional `run_id`, `turn_id`, `assistant_message_id`, and `tool_attempt_id`. - An assistant tool-call batch and each tool attempt have distinct identities. -- A nonempty producer-supplied `idempotency_key`, unique within task and scope. - Keys for per-run facts must include the run/attempt identity; user acceptance - keys can instead identify the durable input command across delivery retries. -- A `kind`, positive `payload_version`, JSON payload, and occurrence timestamp. - -The envelope can carry message/attachment data, full tool results, input -application facts, recovery state and references. Event-specific schemas and -runtime identity producers belong to stage 3.2; this store does not invent -missing run state or infer event kinds from existing Trace rows. No history -backfill is performed here. - -Payloads use the existing JSON sanitizer before persistence, including the -PostgreSQL JSONB code-point policy. They are not clipped or summarized. -This normalization is not an authorization or client-disclosure policy. - -## Transaction contract - -`append_task_execution_event_no_commit` stages a fact in the caller's Session; -the caller commits or rolls back alongside its business state. It first -locks the task with an UPDATE, then checks idempotency and allocates the next -`conversation_event_sequence` in the same transaction. Task `updated_at` is -preserved. No writer can commit a later task sequence past a pending append; -a rollback rolls back the sequence allocation as well as the event. - -A retry with the same key and normalized fact returns the original event ID, -sequence and first occurrence timestamp. Reusing a key for different content -or correlation raises `ExecutionEventConflict`. Callers must finish their -transaction, including on errors; this helper never commits or rolls it back. +# Execution-event storage and writers (stages 3.1–3.2) + +`task_execution_events` is the fact source for explicitly created version-two +test tasks. Production task creation still defaults to version `1`; Web, +channels and existing tasks retain legacy routing. There is no public switch +or automatic conversion of existing conversations. Production activation waits +for the event readers in stage 3.3 and the rollout in stage 3.4. + +## Commit boundaries + +| Boundary | Durable facts | Transaction / compatibility | +| --- | --- | --- | +| Web and channel input, live-message claim | `input_accepted` with original text, attachments and turn identity | Existing acceptance transaction; chat row derived from the event | +| Delivery transition | `input_delivery_changed` | Existing monotonic delivery update, same transaction | +| Control command / interaction | `command_accepted`, `interaction_requested`, `control_state_changed` | Existing permission checks, command identity, state-version and lease fences | +| Runner/runtime and delegated execution | Runtime events, tool start/result/error, `recovery_state` | Strict writer before observers; legacy Trace/checkpoint rows derived in the same transaction | +| First input application | `input_applied`, referring to the proving recovery event | Same transaction as the complete recovery state; acceptance alone does not imply application | +| Outbound messages and Web streams | Message or stream envelope, including protocol identity | Commit before WebSocket broadcast; protocol shape stays unchanged | +| Normal, resumed and channel settlement; orchestrator error settlement | `assistant_message`, `execution_settled` | Existing fenced result/lease transaction; rollback also rolls back transcript and events | + +The runtime bridge consumes events at their producer boundary, before Trace +observers. It does not import historical Trace rows into the fact log. The +ordinary database Trace callback becomes a no-op on this path, retaining its +checkpoint read interface only for the transition. Console, WebSocket and +exporter failures cannot invalidate a committed fact. A failed fact commit +raises `ExecutionEventPersistenceError` and stops execution before broadcast. + +Recovery events contain the complete existing execution snapshot: context, +messages, adopted summaries, pattern state, planning state and pending work. +Their payloads are not clipped to Trace display limits. Runtime LLM payloads +are committed before the observer-side normalization. The JSONB sanitizer +still applies at the storage boundary; unserializable facts fail rather than +becoming a successful write of a diagnostic placeholder. + +## Identity and ordering + +The envelope has a generated `event_id`, task-local `sequence`, explicit +`scope_id` (`root` or the stable delegated execution ID), and optional run, +turn, assistant-batch and tool-attempt identities. Append requires a producer +idempotency key unique within task and scope. Reusing it for a different fact +raises `ExecutionEventConflict`; replay preserves the first ID and timestamp. + +For version-two ReAct execution, each adopted tool batch and each call attempt +receive separate IDs before the `after_llm` recovery state is committed. Those +IDs survive restoring pending calls. Provider call IDs and DAG step IDs are +not treated as globally unique attempt IDs. Tool facts retain full results. +A previously started attempt is blocked from blind execution: until event +recovery reconciles its result, an uncertain external side effect must not be +repeated. This conservative block is intentional during the writer-only stage. + +Appends and pre-append attempt decisions use the same task-row UPDATE lock, +including on SQLite. This serializes sequence allocation, replay and rollback. +Delivery writers acquire the task lock before the message update to preserve +lock order. Runtime and outbound writes reject a replaced bound lease. + +The temporary chat projection has a nullable, unique `execution_event_id`. +Legacy rows keep NULL; replaying one canonical message produces one compatible +chat row. Final assistant message keys include run and state-version identity. +No table independently chooses authoritative message content on the new path. `load_task_execution_events` requires task and scope, and reads by sequence with a page size of 1–100 (default 100). Out-of-range sizes raise `ValueError` @@ -61,18 +65,28 @@ connection cannot see uncommitted events, while read-your-writes in the same Session is intentional. Authorization remains the future calling service's responsibility. Neither helper is an externally exposed endpoint. -Only the append interface is provided; there is no edit or pruning API. -Task deletion cascades at the database foreign key. Stage 3.2 must integrate -the full application lifecycle before production events are enabled. +## Migration and rollback -## Validation +The new migration permits storage versions `1` and `2`, keeping SQL and ORM +creation defaults at `1`. On SQLite it replaces the version column, whose old +CHECK guarantees all pre-migration values are `1`; it does not rebuild `tasks` +or trigger inbound cascade deletes. PostgreSQL replaces the CHECK. Existing +rows, attachments and public protocol fields remain unchanged. + +The legacy readers still consume derived chat / Trace / checkpoint records in +this stage. Removing those readers and implementing event-based reconstruction +belongs to 3.3; version-two tasks must remain test-only until then. There is no +mixed-history fallback or legacy backfill in this change. -The migration tests cover SQLite and PostgreSQL upgrade/downgrade, old SQL -inserts, retained chat rows and foreign keys, create_all parity, and offline -SQL generation. Store tests cover uncommitted visibility, rollback, concurrent -commit ordering and replay, conflicting identities, scope pagination, JSON -payload fidelity and database constraints. +Code rollback for legacy production tasks can retain the additive schema. +Schema downgrade refuses when version-two tasks exist, so it cannot silently +reclassify authoritative event history as legacy data. + +## Validation -Code rollback may retain the additive schema. Alembic downgrade deletes the -new event table and counter: it is suitable for this unwired stage, not a way -to revert an event-backed task after later stages have been activated. +SQLite and PostgreSQL tests cover migration with inbound foreign keys, +unchanged defaults, downgrade refusal, atomic acceptance/settlement, complete +recovery payloads, actual ReAct batch identity, failure-before-broadcast, +observer failures after commit, ownership fences and uncertain-attempt replay. +Existing runner, checkpoint, channel, command, interaction and Trace tests +exercise the unchanged legacy path. diff --git a/src/xagent/core/agent/checkpoint.py b/src/xagent/core/agent/checkpoint.py index d47e90621..1e5958524 100644 --- a/src/xagent/core/agent/checkpoint.py +++ b/src/xagent/core/agent/checkpoint.py @@ -134,6 +134,10 @@ class TraceCheckpointStore: tracer: Any require_persisted: bool = True + @property + def records_execution_events(self) -> bool: + return getattr(self.tracer, "records_execution_events", False) is True + async def checkpoint(self, **payload: Any) -> str | None: return await self.save(payload) diff --git a/src/xagent/core/agent/pattern/dag/dag.py b/src/xagent/core/agent/pattern/dag/dag.py index d2c1a0d7d..bcfbabab3 100644 --- a/src/xagent/core/agent/pattern/dag/dag.py +++ b/src/xagent/core/agent/pattern/dag/dag.py @@ -31,6 +31,7 @@ PatternRuntime, prepare_llm_for_context, ) +from ...trace import ExecutionEventPersistenceError from ..base import AgentPattern, PatternResult, RequiredToolCallError from ..final_answer_stream import FinalAnswerStreamSession, ToolCallStringFieldStreamer from ..react import ReActPattern, ReActReasoningMode @@ -524,6 +525,8 @@ async def _run( raise except RequiredToolCallError: raise + except ExecutionEventPersistenceError: + raise except Exception as exc: # noqa: BLE001 return await self._fail( context=context, @@ -567,6 +570,8 @@ async def _run( raise except RequiredToolCallError: raise + except ExecutionEventPersistenceError: + raise except Exception as exc: # noqa: BLE001 return await self._fail( context=context, @@ -1044,6 +1049,8 @@ async def _execute_step_impl( ) except ExecutionInterrupted: raise + except ExecutionEventPersistenceError: + raise except Exception as exc: step.status = "failed" step.error = str(exc) @@ -1372,6 +1379,8 @@ async def _handle_completed_plan( if interrupted is not None: return interrupted raise + except ExecutionEventPersistenceError: + raise except Exception as exc: # noqa: BLE001 return await self._fail( context=context, @@ -1450,6 +1459,8 @@ async def _handle_completed_plan( raise except RequiredToolCallError: raise + except ExecutionEventPersistenceError: + raise except Exception as exc: # noqa: BLE001 return await self._fail( context=context, diff --git a/src/xagent/core/agent/pattern/react/react.py b/src/xagent/core/agent/pattern/react/react.py index 1e9180c1e..5bb654edf 100644 --- a/src/xagent/core/agent/pattern/react/react.py +++ b/src/xagent/core/agent/pattern/react/react.py @@ -61,6 +61,7 @@ from datetime import timezone from enum import Enum from typing import Any, cast +from uuid import uuid4 from ....file_ref import ( WORKSPACE_OUTPUT_FILES_TOOL_NAME, @@ -103,6 +104,7 @@ prepare_llm_for_context, resolved_llm_metadata, ) +from ...trace import ExecutionEventPersistenceError from ..base import AgentPattern, PatternResult, truncate_prompt_preview from ..final_answer_stream import ReActFinalAnswerStreamer @@ -926,6 +928,11 @@ async def _run_tool_calling_loop( assistant_content = normalized.get("content") tool_calls = normalized.get("tool_calls", []) + if getattr(runtime.tracer, "records_execution_events", False) is True: + batch_id = str(uuid4()) + for tool_call in tool_calls: + tool_call["assistant_message_id"] = batch_id + tool_call["tool_attempt_id"] = str(uuid4()) if assistant_content is not None or normalized.get("tool_calls"): # A tool-protocol error response never carries tool_calls (see # tool_protocol_error_response), so this guard never mistakes @@ -3629,6 +3636,8 @@ async def _execute_tool_safely( ) recorded_terminal = True raise + except ExecutionEventPersistenceError: + raise except Exception as exc: # noqa: BLE001 error_result = { "success": False, diff --git a/src/xagent/core/agent/runner.py b/src/xagent/core/agent/runner.py index 394580ef3..3637e7fb8 100644 --- a/src/xagent/core/agent/runner.py +++ b/src/xagent/core/agent/runner.py @@ -23,6 +23,7 @@ from .language import reset_output_language_to_request_context from .result import extract_assistant_message from .runtime import ExecutionInterrupted, PatternRuntime, load_pattern_checkpoint +from .trace import ExecutionEventPersistenceError logger = logging.getLogger(__name__) @@ -320,6 +321,10 @@ async def run( result=normalized, ) return normalized + except ExecutionEventPersistenceError: + # A storage failure cannot authorize another pattern to + # repeat potentially completed external work. + raise except Exception as exc: # noqa: BLE001 teardown_status = "failed" logger.exception( diff --git a/src/xagent/core/agent/runtime.py b/src/xagent/core/agent/runtime.py index 8ca102c07..25baab9f0 100644 --- a/src/xagent/core/agent/runtime.py +++ b/src/xagent/core/agent/runtime.py @@ -10,6 +10,7 @@ from ...config import get_compact_threshold_ratio from ..agent.trace import ( + ExecutionEventPersistenceError, TraceAction, TraceCategory, TraceEventType, @@ -894,6 +895,11 @@ async def on_tool_start(self, *, tool_call: dict[str, Any]) -> None: "tool_name": tool_call.get("name"), "tool_params": tool_call.get("args", {}), "tool_call_id": tool_call.get("id"), + **{ + key: tool_call[key] + for key in ("assistant_message_id", "tool_attempt_id") + if key in tool_call + }, } assistant_content = tool_call.get("assistant_content") if isinstance(assistant_content, str) and assistant_content.strip(): @@ -927,6 +933,11 @@ async def on_tool_end(self, *, tool_call: dict[str, Any], result: Any) -> None: "tool_name": tool_call.get("name"), "tool_params": tool_call.get("args", {}), "tool_call_id": tool_call.get("id"), + **{ + key: tool_call[key] + for key in ("assistant_message_id", "tool_attempt_id") + if key in tool_call + }, "result": result, "success": False, "status": WAITING_FOR_USER_STATUS, @@ -964,6 +975,11 @@ async def on_tool_end(self, *, tool_call: dict[str, Any], result: Any) -> None: "tool_name": tool_call.get("name"), "tool_params": tool_call.get("args", {}), "tool_call_id": tool_call.get("id"), + **{ + key: tool_call[key] + for key in ("assistant_message_id", "tool_attempt_id") + if key in tool_call + }, "result": result, "success": True, } @@ -994,6 +1010,11 @@ async def on_tool_error( "error_message": str(error), "tool_name": tool_call.get("name"), "tool_call_id": tool_call.get("id"), + **{ + key: tool_call[key] + for key in ("assistant_message_id", "tool_attempt_id") + if key in tool_call + }, } if result is not None: data["result"] = result @@ -1031,6 +1052,11 @@ async def on_tool_cancelled( "tool_name": tool_call.get("name"), "tool_params": tool_call.get("args", {}), "tool_call_id": tool_call.get("id"), + **{ + key: tool_call[key] + for key in ("assistant_message_id", "tool_attempt_id") + if key in tool_call + }, "success": False, "interrupted": True, "interrupt_reason": cancellation_reason, @@ -1632,7 +1658,11 @@ async def _emit_trace_event( # truncates bulky content (messages, response, tool_calls, ...). # Non-LLM categories (TOOL / DAG / REACT / COMPACT / GENERAL) # pass through unchanged. - if data and getattr(event_type, "category", None) == TraceCategory.LLM: + if ( + data + and getattr(event_type, "category", None) == TraceCategory.LLM + and getattr(self.tracer, "records_execution_events", False) is not True + ): data = normalize_llm_trace_payload(data) try: await self._maybe_await( @@ -1643,6 +1673,8 @@ async def _emit_trace_event( data=data or {}, ) ) + except ExecutionEventPersistenceError: + raise except Exception: # UI trace events are best-effort; checkpoint persistence remains strict. return diff --git a/src/xagent/core/agent/trace.py b/src/xagent/core/agent/trace.py index 28c150daf..7774e5ad8 100644 --- a/src/xagent/core/agent/trace.py +++ b/src/xagent/core/agent/trace.py @@ -6,7 +6,7 @@ from abc import ABC, abstractmethod from datetime import datetime, timezone from enum import Enum -from typing import Any, Callable, Dict, List, Optional, Set, cast +from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, cast from uuid import uuid4 from ..utils.security import redact_sensitive_text @@ -1223,13 +1223,22 @@ async def _handle_system_event(self, event: TraceEvent) -> None: logger.debug(f"[DB] System event: {event.event_type.value}") +class ExecutionEventPersistenceError(RuntimeError): + """A conversation fact was not durably committed; execution must stop.""" + + class Tracer: """Main tracing class that manages trace events and handlers.""" def __init__(self) -> None: self.handlers: List[TraceHandler] = [] + self.event_writer: Callable[[TraceEvent], Awaitable[None]] | None = None # No default handlers - let users add their own + @property + def records_execution_events(self) -> bool: + return self.event_writer is not None + def add_handler(self, handler: TraceHandler) -> None: """Add a trace handler.""" self.handlers.append(handler) @@ -1268,6 +1277,15 @@ async def trace_event( require_persisted=require_persisted, ) + # Commit facts before any observer sees them. Observer failures do not + # invalidate an already committed fact. Legacy tracers have no writer. + if self.event_writer is not None: + await self.event_writer(event) + require_persisted = False + event.require_persisted = False + if event.event_type.category == TraceCategory.LLM: + event.data = normalize_llm_trace_payload(event.data) + # Notify all handlers logger.info( f"Notifying {len(self.handlers)} handlers for event {event_type.value}" diff --git a/src/xagent/core/tools/adapters/vibe/agent_tool.py b/src/xagent/core/tools/adapters/vibe/agent_tool.py index fa3f24549..6067e8d4a 100644 --- a/src/xagent/core/tools/adapters/vibe/agent_tool.py +++ b/src/xagent/core/tools/adapters/vibe/agent_tool.py @@ -5,7 +5,7 @@ import logging from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any, Mapping, Optional, Type +from typing import TYPE_CHECKING, Any, Mapping, Optional, Type, cast from uuid import uuid4 from pydantic import BaseModel, Field, field_validator @@ -22,6 +22,7 @@ # (which also keeps it patchable in tests). from .....web.tools.config import WebToolConfig from ....agent.result import NO_OUTPUT_PLACEHOLDER, NO_RESPONSE_PLACEHOLDER +from ....agent.trace import ExecutionEventPersistenceError from ....agent.voice_policy import apply_output_voice from ....task_runtime import FILE_OPERATION_ACCESS_VERSION_KEY from ....tracing import create_agent_tracer @@ -59,12 +60,12 @@ def __init__( build_id: str, metadata: Mapping[str, Any], ) -> None: - from .....web.api.trace_handlers import DatabaseTraceHandler + from .....web.tracing import task_database_handler self.task_id = task_id self.build_id = build_id self.metadata = dict(metadata) - self._handler = DatabaseTraceHandler(task_id, build_id=build_id) + self._handler = task_database_handler(task_id, build_id=build_id) async def handle_event(self, event: Any) -> None: original_data = event.data @@ -75,6 +76,14 @@ async def handle_event(self, event: Any) -> None: finally: event.data = original_data + async def commit_event(self, event: Any) -> None: + original_data = event.data + event.data = {**(original_data or {}), **self.metadata} + try: + await cast(Any, self._handler).commit_event(event) + finally: + event.data = original_data + async def load_latest_checkpoint( self, execution_id: str ) -> Optional[dict[str, Any]]: @@ -1971,7 +1980,7 @@ def _create_child_execution_tracer( ) ) - return create_agent_tracer( + tracer = create_agent_tracer( handlers=handlers, task_id=execution_task_id, user_id=self._user_id, @@ -1981,6 +1990,10 @@ def _create_child_execution_tracer( metadata=metadata, ) + if parent_db_task_id is not None and handlers[0]._handler.authoritative: + tracer.event_writer = handlers[0].commit_event + return tracer + def _resolve_delegated_output_path(self, workspace: Any, raw_path: str) -> Path: raw = raw_path.strip() path = Path(raw) @@ -2401,6 +2414,8 @@ async def execute_delegated_runtime() -> tuple[dict[str, Any], Any]: file_outputs=file_outputs, ).model_dump(exclude_none=True) + except ExecutionEventPersistenceError: + raise except Exception as e: error_msg = f"Error executing agent {self._agent_id}: {str(e)}" logger.error(error_msg, exc_info=True) diff --git a/src/xagent/migrations/versions/20260905_enable_task_execution_event_writers.py b/src/xagent/migrations/versions/20260905_enable_task_execution_event_writers.py new file mode 100644 index 000000000..84ea589ba --- /dev/null +++ b/src/xagent/migrations/versions/20260905_enable_task_execution_event_writers.py @@ -0,0 +1,78 @@ +"""Allow explicitly created event-backed test tasks; keep legacy defaults.""" + +import sqlalchemy as sa +from alembic import op + +revision = "20260905_execution_event_writers" +down_revision = "20260905_task_execution_events" +branch_labels = None +depends_on = None + + +def _replace_check(expression: str) -> None: + if op.get_bind().dialect.name == "sqlite": + # This column is not referenced by any FK/index. Replacing the column + # avoids rebuilding tasks and triggering inbound ON DELETE CASCADE. + # Upgrade starts with only version 1; downgrade verifies that below. + op.drop_column("tasks", "conversation_storage_version") + op.execute( + "ALTER TABLE tasks ADD COLUMN conversation_storage_version INTEGER " + "DEFAULT 1 NOT NULL CONSTRAINT ck_tasks_conversation_storage_version " + f"CHECK ({expression})" + ) + else: + op.drop_constraint( + "ck_tasks_conversation_storage_version", "tasks", type_="check" + ) + op.create_check_constraint( + "ck_tasks_conversation_storage_version", "tasks", expression + ) + + +def upgrade() -> None: + offline = op.get_context().as_sql + inspector = None if offline else sa.inspect(op.get_bind()) + if offline or inspector.has_table("task_chat_messages"): + if offline or "execution_event_id" not in { + c["name"] for c in inspector.get_columns("task_chat_messages") + }: + op.add_column( + "task_chat_messages", + sa.Column("execution_event_id", sa.String(36), nullable=True), + ) + op.create_index( + "ix_task_chat_messages_execution_event_id", + "task_chat_messages", + ["execution_event_id"], + unique=True, + ) + if not op.get_context().as_sql: + inspector = sa.inspect(op.get_bind()) + if not inspector.has_table("tasks"): + return + check = next( + c + for c in inspector.get_check_constraints("tasks") + if c["name"] == "ck_tasks_conversation_storage_version" + ) + if "IN" in check["sqltext"].upper() or "ANY" in check["sqltext"].upper(): + return # create_all already installed this schema + _replace_check("conversation_storage_version IN (1, 2)") + + +def downgrade() -> None: + if op.get_context().as_sql: + raise RuntimeError("Event-writer downgrade requires an online version check") + inspector = sa.inspect(op.get_bind()) + has_tasks = inspector.has_table("tasks") + if has_tasks and op.get_bind().scalar( + sa.text("SELECT count(*) FROM tasks WHERE conversation_storage_version <> 1") + ): + raise RuntimeError("Cannot downgrade while event-backed tasks exist") + if inspector.has_table("task_chat_messages"): + op.drop_index( + "ix_task_chat_messages_execution_event_id", table_name="task_chat_messages" + ) + op.drop_column("task_chat_messages", "execution_event_id") + if has_tasks: + _replace_check("conversation_storage_version = 1") diff --git a/src/xagent/web/api/trace_handlers.py b/src/xagent/web/api/trace_handlers.py index 9239af69e..68210cc67 100644 --- a/src/xagent/web/api/trace_handlers.py +++ b/src/xagent/web/api/trace_handlers.py @@ -191,6 +191,7 @@ def __init__(self, task_id: int, build_id: Optional[str] = None): super().__init__() self.task_id = task_id self.build_id = build_id + self.authoritative = False async def _handle_task_event(self, event: CoreTraceEvent) -> None: """Handle task-level events for database storage.""" @@ -981,7 +982,108 @@ def _save_trace_event(self, db: Session, event: CoreTraceEvent) -> None: timestamp = _convert_float_to_datetime(event.timestamp) # Serialize data to ensure JSON compatibility - data = self._serialize_data_for_json(event.data or {}) + data = ( + (event.data or {}) + if self.authoritative + else self._serialize_data_for_json(event.data or {}) + ) + if self.authoritative: + from ..services.task_execution_event_store import ( + lock_task_execution_events_no_commit, + ) + from ..services.task_execution_event_writer import append_fact_no_commit + + lock_task_execution_events_no_commit(db, self.task_id) + lease = current_task_lease() + if lease is not None: + owned = ( + db.query(Task.id) + .filter( + Task.id == self.task_id, + Task.runner_id == lease.runner_id, + Task.run_id == lease.run_id, + Task.status == TaskStatus.RUNNING, + ) + .with_for_update() + .first() + ) + if owned is None: + raise RuntimeError( + "Execution event producer lost its task lease" + ) + is_state = data.get("checkpoint_type") in READABLE_CHECKPOINT_TYPES + attempt = data.get("tool_attempt_id") + if attempt: + from uuid import NAMESPACE_URL, uuid5 + + from ..models.task_execution_event import TaskExecutionEvent + + event.id = str( + uuid5( + NAMESPACE_URL, + f"task:{self.task_id}:{self.build_id or 'root'}:{attempt}:{event_type_str}", + ) + ) + if ( + event_type_str == "tool_execution_start" + and db.query(TaskExecutionEvent.id) + .filter( + TaskExecutionEvent.task_id == self.task_id, + TaskExecutionEvent.scope_id == (self.build_id or "root"), + TaskExecutionEvent.tool_attempt_id == attempt, + ) + .first() + is not None + ): + # Until event-based recovery reconciles the attempt, do + # not re-execute a possibly completed external effect. + raise RuntimeError( + "Tool attempt already started; result reconciliation required" + ) + key = ( + f"tool:{attempt}:{event_type_str}" + if attempt + else f"runtime:{event.id}" + ) + fact = append_fact_no_commit( + db, + task_id=self.task_id, + scope_id=self.build_id or "root", + run_id=lease.run_id if lease is not None else None, + turn_id=data.get("turn_id"), + assistant_message_id=data.get("assistant_message_id"), + tool_attempt_id=attempt, + key=key, + kind="recovery_state" if is_state else event_type_str, + payload={ + "data": data, + "step_id": event.step_id, + "protocol_event_id": str(event.id), + "event_type": event_type_str, + "parent_event_id": str(event.parent_id) + if event.parent_id + else None, + }, + occurred_at=timestamp, + ) + data = fact.payload["data"] + if is_state: + from ..services.task_execution_event_writer import ( + stage_applied_inputs_no_commit, + ) + + stage_applied_inputs_no_commit(db, fact) + if ( + db.query(DatabaseTraceEvent.id) + .filter( + DatabaseTraceEvent.task_id == self.task_id, + DatabaseTraceEvent.event_id == str(event.id), + ) + .first() + is not None + ): + db.commit() + return if event_type_str in { "tool_execution_start", "tool_execution_end", @@ -994,6 +1096,8 @@ def _save_trace_event(self, db: Session, event: CoreTraceEvent) -> None: data.get("turn_id") if isinstance(data, dict) else None, self.task_id, ) + if self.authoritative: + db.commit() return if ( event_type_str == "system_update_general" diff --git a/src/xagent/web/api/websocket.py b/src/xagent/web/api/websocket.py index 5a017d6a8..b70cd8b6e 100644 --- a/src/xagent/web/api/websocket.py +++ b/src/xagent/web/api/websocket.py @@ -645,7 +645,14 @@ def _terminal_task_error_payload( ) -> dict[str, Any] | None: SessionLocal = get_session_local() db = SessionLocal() + canonical = False try: + from ..services.task_execution_event_writer import ( + stage_result_fact_no_commit, + uses_execution_events, + ) + + canonical = uses_execution_events(db, task_id) failed_control_state = TaskControlState.FAILED.value current_version = func.coalesce(Task.state_version, 0) statement = ( @@ -718,10 +725,15 @@ def _terminal_task_error_payload( message_type=TASK_FAILURE_MESSAGE_TYPE, ) except Exception: + if task.conversation_storage_version == 2: + raise logger.warning( "Failed to persist terminal error chat message", exc_info=True, ) + if canonical: + db.refresh(task) + stage_result_fact_no_commit(db, task, {"error": message}) db.commit() return _task_error_payload( db, @@ -729,8 +741,14 @@ def _terminal_task_error_payload( CLIENT_SAFE_TASK_FAILURE, event_type=event_type, ) - except Exception: + except Exception as exc: db.rollback() + if canonical: + from ...core.agent.trace import ExecutionEventPersistenceError + + raise ExecutionEventPersistenceError( + "Terminal failure event commit failed" + ) from exc logger.warning("Failed to persist terminal task error", exc_info=True) return { "type": event_type, @@ -1009,7 +1027,7 @@ def create_final_answer_stream_event( data: Dict[str, Any], timestamp: Optional[Any] = None, ) -> Dict[str, Any]: - """Create non-persistent final-answer UI stream events.""" + """Create final-answer UI stream envelopes without changing their protocol.""" payload = dict(data) payload.pop("type", None) @@ -1037,17 +1055,22 @@ def _stream_timestamp(timestamp: Optional[Any] = None) -> float: return float(timestamp) -def _persist_agent_outbound_event(task_id: int, event: Dict[str, Any]) -> None: +def _persist_agent_outbound_event( + task_id: int, event: Dict[str, Any], *, authoritative: bool = False +) -> None: """Persist agent outbound events and durable waiting prompts.""" from ..models.task import Task as DatabaseTask from ..models.task import TraceEvent as DatabaseTraceEvent - from ..services.chat_history_service import persist_assistant_message + from ..services.chat_history_service import persist_assistant_message_no_commit + task: DatabaseTask | None db_gen = get_db() db = next(db_gen) try: event_data = event.get("data") + if authoritative and event_data is None: + event_data = dict(event) data: Dict[str, Any] = cast( Dict[str, Any], event_data if isinstance(event_data, dict) else {} ) @@ -1073,6 +1096,34 @@ def _persist_agent_outbound_event(task_id: int, event: Dict[str, Any]) -> None: parent_event_id=None, data=data, ) + if authoritative: + from ..services.task_execution_event_writer import append_fact_no_commit + from ..services.task_lease_service import current_task_lease + + task = ( + db.query(DatabaseTask) + .filter(DatabaseTask.id == task_id) + .with_for_update() + .one() + ) + lease = current_task_lease() + if lease is not None and ( + task.run_id != lease.run_id or task.runner_id != lease.runner_id + ): + raise RuntimeError("Outbound event producer lost its task lease") + fact = append_fact_no_commit( + db, + task_id=task_id, + kind=str(trace_event.event_type), + key=f"outbound:{trace_event.event_id}", + run_id=cast(str | None, task.run_id), + payload={"data": data, "protocol_event_id": trace_event.event_id}, + occurred_at=event_time, + ) + setattr(trace_event, "data", fact.payload["data"]) + if trace_event.event_type.startswith("final_answer_"): + db.commit() + return db.add(trace_event) if bool(data.get("expect_response")): @@ -1087,7 +1138,7 @@ def _persist_agent_outbound_event(task_id: int, event: Dict[str, Any]) -> None: and isinstance(metadata.get("interactions"), list) else None ) - persist_assistant_message( + persist_assistant_message_no_commit( db, task_id=task_id, user_id=task_user_id, @@ -1097,8 +1148,14 @@ def _persist_agent_outbound_event(task_id: int, event: Dict[str, Any]) -> None: ) db.commit() - except Exception: + except Exception as exc: db.rollback() + if authoritative: + from ...core.agent.trace import ExecutionEventPersistenceError + + raise ExecutionEventPersistenceError( + "Outbound event commit failed" + ) from exc logger.exception( "Failed to persist agent outbound message for task %s", task_id ) @@ -1134,7 +1191,7 @@ def _reconcile_streamed_final_answer(task_id: int, content: str) -> str: db.close() -def make_agent_outbound_handler(task_id: int) -> Any: +def make_agent_outbound_handler(task_id: int, *, authoritative: bool = False) -> Any: """Create a web bridge for agent agent-to-user messages.""" async def handle_outbound_message(payload: Dict[str, Any]) -> None: @@ -1154,10 +1211,17 @@ async def handle_outbound_message(payload: Dict[str, Any]) -> None: task_id, str(payload["content"]), ) - await manager.broadcast_to_task( - create_final_answer_stream_event(payload_type, task_id, dict(payload)), - task_id, + final_answer_event = create_final_answer_stream_event( + payload_type, task_id, dict(payload) ) + if authoritative: + await asyncio.to_thread( + _persist_agent_outbound_event, + task_id, + final_answer_event, + authoritative=True, + ) + await manager.broadcast_to_task(final_answer_event, task_id) return if payload.get("visible") is False: @@ -1180,7 +1244,12 @@ async def handle_outbound_message(payload: Dict[str, Any]) -> None: }, event_id=payload.get("event_id"), ) - await asyncio.to_thread(_persist_agent_outbound_event, task_id, event) + if authoritative: + await asyncio.to_thread( + _persist_agent_outbound_event, task_id, event, authoritative=True + ) + else: + await asyncio.to_thread(_persist_agent_outbound_event, task_id, event) await manager.broadcast_to_task(event, task_id) return handle_outbound_message @@ -2431,6 +2500,7 @@ def _finalize_task_execution_result_isolated( ) -> _TaskExecutionFinalization: """Persist one task result in a worker-owned, ownership-fenced session.""" from ..services.chat_history_service import persist_assistant_message_no_commit + from ..services.task_execution_event_writer import stage_result_fact_no_commit if prepared_outputs is None: resolved_output_user_id = task_user_id @@ -2564,6 +2634,7 @@ def _finalize_task_execution_result_isolated( task_updated, task_updated.status, ) + stage_result_fact_no_commit(finalize_db, task_updated, result) finalize_db.commit() metadata_committed = True terminal_state_committed = True @@ -2586,6 +2657,7 @@ def _finalize_task_execution_result_isolated( task_updated, task_updated.status, ) + stage_result_fact_no_commit(finalize_db, task_updated, result) finalize_db.commit() metadata_committed = True terminal_state_committed = True @@ -2653,6 +2725,7 @@ def _finalize_task_execution_result_isolated( ), content_is_reconciled=True, ) + stage_result_fact_no_commit(finalize_db, task_updated, result) finalize_db.commit() metadata_committed = True terminal_state_committed = True @@ -2788,7 +2861,13 @@ async def execute_task_background( ) if hasattr(agent_service, "set_outbound_message_handler"): agent_service.set_outbound_message_handler( - make_agent_outbound_handler(task_id) + make_agent_outbound_handler( + task_id, + authoritative=getattr( + agent_service.tracer, "records_execution_events", False + ) + is True, + ) ) agent_service.set_conversation_history( [dict(message) for message in snapshot.conversation_history], @@ -3185,6 +3264,7 @@ def _finalize_resumed_task( """Persist one fenced resumed result in a single worker transaction.""" from ..models.agent import Agent from ..services.chat_history_service import persist_assistant_message_no_commit + from ..services.task_execution_event_writer import stage_result_fact_no_commit finalized: dict[str, Any] = { "task_title": None, @@ -3316,6 +3396,7 @@ def _finalize_resumed_task( db.rollback() finalized["late_result"] = True return finalized + stage_result_fact_no_commit(db, task, result) db.commit() metadata_committed = True finalized["lease_released"] = True @@ -6639,7 +6720,13 @@ async def finish_existing_delivery( ) if hasattr(agent_service, "set_outbound_message_handler"): agent_service.set_outbound_message_handler( - make_agent_outbound_handler(task_id) + make_agent_outbound_handler( + task_id, + authoritative=getattr( + agent_service.tracer, "records_execution_events", False + ) + is True, + ) ) supports_live_control = getattr( agent_service, "supports_live_control", lambda: False diff --git a/src/xagent/web/models/chat_message.py b/src/xagent/web/models/chat_message.py index c3b2505c0..72e0507bc 100644 --- a/src/xagent/web/models/chat_message.py +++ b/src/xagent/web/models/chat_message.py @@ -42,6 +42,8 @@ class TaskChatMessage(Base): # type: ignore # the original file metadata (file_id, name, size, type) is available for # historical replay without having to re-derive it from the message body. attachments = Column(JSON, nullable=True) + # Transitional projection identity; NULL for legacy transcript rows. + execution_event_id = Column(String(36), nullable=True, unique=True, index=True) created_at = Column(DateTime(timezone=True), server_default=func.now()) task = relationship("Task", back_populates="chat_messages") diff --git a/src/xagent/web/models/task.py b/src/xagent/web/models/task.py index 695a234a3..6cf5d3a67 100644 --- a/src/xagent/web/models/task.py +++ b/src/xagent/web/models/task.py @@ -278,12 +278,12 @@ class Task(Base): # type: ignore user_id = Column(Integer, ForeignKey("users.id"), nullable=False) title = Column(String(200), nullable=False) description = Column(Text) - # Stage 3.1 only supports legacy routing. Widen this pin together with - # the event-backed runtime, never by changing the creation default alone. + # Production creation remains legacy; version 2 is exercised by migration + # tests until the event readers are ready for production routing. conversation_storage_version = Column( Integer, CheckConstraint( - "conversation_storage_version = 1", + "conversation_storage_version IN (1, 2)", name="ck_tasks_conversation_storage_version", ), nullable=False, diff --git a/src/xagent/web/models/task_execution_event.py b/src/xagent/web/models/task_execution_event.py index e8cc7f9df..563a0adb1 100644 --- a/src/xagent/web/models/task_execution_event.py +++ b/src/xagent/web/models/task_execution_event.py @@ -1,4 +1,4 @@ -"""Conversation/execution facts; not wired into runtime producers yet.""" +"""Authoritative conversation/execution facts for version-two tasks.""" from sqlalchemy import ( JSON, diff --git a/src/xagent/web/services/chat_history_service.py b/src/xagent/web/services/chat_history_service.py index 5c3a044c5..e02f3881b 100644 --- a/src/xagent/web/services/chat_history_service.py +++ b/src/xagent/web/services/chat_history_service.py @@ -36,6 +36,10 @@ clear_degradation, register_degradation, ) +from .task_execution_event_writer import ( + stage_chat_message_no_commit, + stage_delivery_fact_no_commit, +) logger = logging.getLogger(__name__) @@ -202,7 +206,7 @@ def claim_user_message_delivery( delivery_status=DELIVERY_PENDING, attachments=attachments, ) - db.add(message) + message = stage_chat_message_no_commit(db, message) try: db.commit() db.refresh(message) @@ -257,7 +261,7 @@ def claim_user_message_delivery_no_commit( delivery_status=DELIVERY_PENDING, attachments=attachments, ) - db.add(message) + message = stage_chat_message_no_commit(db, message) db.flush() return UserMessageDeliveryClaim( message=message, @@ -282,6 +286,12 @@ def mark_user_message_delivery( DELIVERY_FAILED, }: raise ValueError(f"Unknown delivery status: {status}") + from .task_execution_event_store import lock_task_execution_events_no_commit + from .task_execution_event_writer import uses_execution_events + + if uses_execution_events(db, task_id): + # Use the same task-before-message lock order as acceptance/finalization. + lock_task_execution_events_no_commit(db, task_id) query = db.query(TaskChatMessage).filter( TaskChatMessage.task_id == task_id, TaskChatMessage.role == "user", @@ -310,6 +320,9 @@ def mark_user_message_delivery( synchronize_session=False, ) if updated: + stage_delivery_fact_no_commit( + db, task_id=task_id, turn_id=turn_id, status=status + ) return UserMessageDeliveryTransition(status=status, outcome="updated") # A concurrent terminal transition won after the read. Reload the durable @@ -646,7 +659,7 @@ def persist_user_message_no_commit( # "attachments key was set, just empty". attachments=attachments, ) - db.add(message) + message = stage_chat_message_no_commit(db, message) return message @@ -725,7 +738,7 @@ def persist_assistant_message_no_commit( turn_id=turn_id, attachments=None, ) - db.add(message) + message = stage_chat_message_no_commit(db, message) return message @@ -1061,7 +1074,7 @@ def _persist_message( # round-trips as ``[]`` rather than being coerced to ``NULL``. attachments=attachments, ) - db.add(message) + message = stage_chat_message_no_commit(db, message) db.commit() db.refresh(message) return message diff --git a/src/xagent/web/services/external_task_cancel.py b/src/xagent/web/services/external_task_cancel.py index a1263dafb..3fd068af6 100644 --- a/src/xagent/web/services/external_task_cancel.py +++ b/src/xagent/web/services/external_task_cancel.py @@ -425,6 +425,15 @@ def _finalize_external_cancel_sync( ) _persist_interruption_transcript_no_commit(db, updated) _mark_cancelled_turn_delivery_dispatched(db, task_id, turn_id=turn_id) + if updated.conversation_storage_version == 2: + from .task_execution_event_writer import stage_result_fact_no_commit + + db.refresh(updated) + stage_result_fact_no_commit( + db, + updated, + {"status": "cancelled", "error": EXTERNAL_CANCEL_ERROR_MESSAGE}, + ) db.commit() _invalidate_task_cache_after_commit(task_id) diff --git a/src/xagent/web/services/managed_task_lease.py b/src/xagent/web/services/managed_task_lease.py index 34064bd96..4d8de42fb 100644 --- a/src/xagent/web/services/managed_task_lease.py +++ b/src/xagent/web/services/managed_task_lease.py @@ -61,6 +61,7 @@ def finalize_managed_task_lease_result( raise ValueError("Cannot finalize a managed lease with RUNNING status") from .chat_history_service import persist_assistant_message_no_commit + from .task_execution_event_writer import stage_result_fact_no_commit from .task_orchestrator import invalidate_task_cache_best_effort try: @@ -99,6 +100,9 @@ def finalize_managed_task_lease_result( message_type=history_message_type, turn_id=turn_id, ) + stage_result_fact_no_commit( + db, task, dict(execution_result or {"error": error_message}) + ) db.commit() except Exception: db.rollback() diff --git a/src/xagent/web/services/task_command_transport.py b/src/xagent/web/services/task_command_transport.py index eb1c31152..4b666a528 100644 --- a/src/xagent/web/services/task_command_transport.py +++ b/src/xagent/web/services/task_command_transport.py @@ -471,6 +471,26 @@ def stage_task_command( ) db.add(command) db.flush() + from .task_execution_event_writer import ( + append_fact_no_commit, + uses_execution_events, + ) + + if uses_execution_events(db, resolved_task_id): + append_fact_no_commit( + db, + task_id=resolved_task_id, + kind="command_accepted", + key=f"command:{normalized_id}", + run_id=snapshot.run_id, + payload={ + "command_id": normalized_id, + "kind": kind.value, + "payload": payload, + "actor_user_id": actor_user_id, + "target_state_version": int(snapshot.state_version or 0), + }, + ) return StagedTaskCommand( staged_db_id=int(command.id), client_command_id=normalized_id, diff --git a/src/xagent/web/services/task_execution_controller.py b/src/xagent/web/services/task_execution_controller.py index a5667c182..a43dfda82 100644 --- a/src/xagent/web/services/task_execution_controller.py +++ b/src/xagent/web/services/task_execution_controller.py @@ -13,7 +13,7 @@ from contextlib import asynccontextmanager from contextvars import ContextVar from dataclasses import dataclass -from typing import Any, AsyncIterator +from typing import Any, AsyncIterator, cast from uuid import uuid4 from sqlalchemy import func, update @@ -187,7 +187,19 @@ def apply_task_control_transition( f"at state version {expected_state_version}" ) session.refresh(task) - return task_control_snapshot(task) + snapshot = task_control_snapshot(task) + if task.conversation_storage_version == 2: + from .task_execution_event_writer import append_fact_no_commit + + append_fact_no_commit( + session, + task_id=int(task.id), + kind="control_state_changed", + key=f"control:{task.state_version}", + run_id=cast(str | None, task.run_id), + payload=snapshot.as_dict(), + ) + return snapshot # Fallback for detached/transient objects. Persistent task rows use the # atomic UPDATE above so concurrent lifecycle writers cannot reuse a diff --git a/src/xagent/web/services/task_execution_event_store.py b/src/xagent/web/services/task_execution_event_store.py index ac8fe86b9..843818273 100644 --- a/src/xagent/web/services/task_execution_event_store.py +++ b/src/xagent/web/services/task_execution_event_store.py @@ -1,8 +1,4 @@ -"""Unwired event-store primitives for the conversation migration. - -The caller owns the transaction. Runtime integration and event-specific payload -contracts ship in stage 3.2; no existing task producer calls this module. -""" +"""Transactional event-store primitives for the conversation migration.""" from __future__ import annotations @@ -25,6 +21,25 @@ class ExecutionEventConflict(ValueError): """An idempotency key was reused for a different fact.""" +def lock_task_execution_events_no_commit(db: Session, task_id: int) -> int: + """Serialize pre-append decisions using the same lock as event allocation.""" + sequence = db.execute( + update(Task) + .where(Task.id == task_id) + .values( + conversation_event_sequence=Task.conversation_event_sequence, + # Allocating an internal cursor must not reorder the task list. + updated_at=Task.updated_at, + ) + .returning(Task.conversation_event_sequence) + .execution_options(synchronize_session=False) + ).scalar_one_or_none() + if sequence is None: + raise ValueError(f"Task {task_id} does not exist") + + return int(sequence) + + def append_task_execution_event_no_commit( db: Session, *, @@ -62,19 +77,7 @@ def append_task_execution_event_no_commit( "assistant_message_id": assistant_message_id, "tool_attempt_id": tool_attempt_id, } - sequence = db.execute( - update(Task) - .where(Task.id == task_id) - .values( - conversation_event_sequence=Task.conversation_event_sequence, - # Allocating an internal cursor must not reorder the task list. - updated_at=Task.updated_at, - ) - .returning(Task.conversation_event_sequence) - .execution_options(synchronize_session=False) - ).scalar_one_or_none() - if sequence is None: - raise ValueError(f"Task {task_id} does not exist") + sequence = lock_task_execution_events_no_commit(db, task_id) existing = db.scalars( select(TaskExecutionEvent).where( diff --git a/src/xagent/web/services/task_execution_event_writer.py b/src/xagent/web/services/task_execution_event_writer.py new file mode 100644 index 000000000..e860bef3c --- /dev/null +++ b/src/xagent/web/services/task_execution_event_writer.py @@ -0,0 +1,182 @@ +"""Version-two facts and their transitional legacy projections. + +All helpers participate in the caller's transaction. Version one remains the +creation default; there is deliberately no API for switching existing tasks. +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any, cast +from uuid import uuid4 + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from ..models.chat_message import TaskChatMessage +from ..models.task import Task, TaskStatus +from ..models.task_execution_event import TaskExecutionEvent +from .task_execution_event_store import append_task_execution_event_no_commit + + +def uses_execution_events(db: Session, task_id: int) -> bool: + return ( + db.scalar(select(Task.conversation_storage_version).where(Task.id == task_id)) + == 2 + ) + + +def _fact_json_default(value: Any) -> Any: + """Serialize the existing execution-result protocols without a lossy fallback.""" + if callable(getattr(value, "model_dump", None)): + return value.model_dump(mode="json") + if callable(getattr(value, "to_dict", None)): + return value.to_dict() + if isinstance(value, datetime): + return value.isoformat() + raise TypeError(f"Unsupported execution fact value: {type(value).__name__}") + + +def append_fact_no_commit( + db: Session, + *, + task_id: int, + kind: str, + key: str, + payload: dict[str, Any], + scope_id: str = "root", + run_id: str | None = None, + turn_id: str | None = None, + assistant_message_id: str | None = None, + tool_attempt_id: str | None = None, + occurred_at: datetime | None = None, +) -> TaskExecutionEvent: + return append_task_execution_event_no_commit( + db, + task_id=task_id, + scope_id=scope_id, + kind=kind, + idempotency_key=key, + payload=json.loads( + json.dumps(payload, default=_fact_json_default, allow_nan=False) + ), + run_id=run_id, + turn_id=turn_id, + assistant_message_id=assistant_message_id, + tool_attempt_id=tool_attempt_id, + occurred_at=occurred_at or datetime.now(timezone.utc), + ) + + +def stage_chat_message_no_commit( + db: Session, message: TaskChatMessage +) -> TaskChatMessage: + """Commit the message fact first, then materialize its old-protocol row.""" + if not uses_execution_events(db, int(message.task_id)): + db.add(message) + return message + task = cast(Any, db.get(Task, message.task_id)) + turn_id = cast(str | None, message.turn_id) + if message.role == "user" and not turn_id: + turn_id = str(uuid4()) + identity = turn_id if message.role == "user" else str(uuid4()) + if message.role == "assistant" and task.status in { + TaskStatus.COMPLETED, + TaskStatus.FAILED, + TaskStatus.PAUSED, + TaskStatus.WAITING_FOR_USER, + }: + identity = f"assistant:{task.run_id}:{task.state_version}:{message.message_type}:{turn_id}" + payload = { + "user_id": message.user_id, + "role": message.role, + "content": message.content, + "message_type": message.message_type, + "interactions": message.interactions, + "attachments": message.attachments, + "turn_id": turn_id, + "delivery_status": message.delivery_status, + } + event = append_fact_no_commit( + db, + task_id=int(message.task_id), + kind="input_accepted" if message.role == "user" else "assistant_message", + key=f"message:{identity}", + payload=payload, + run_id=cast(str | None, task.run_id), + turn_id=turn_id, + ) + # Reuse the committed envelope, including an empty attachments list and the + # actual accepted turn identity. The old row is only a compatibility reader. + existing = db.scalar( + select(TaskChatMessage).where( + TaskChatMessage.execution_event_id == event.event_id, + ) + ) + if existing is not None: + return existing + message.execution_event_id = event.event_id + for field, value in event.payload.items(): + setattr(message, field, value) + db.add(message) + return message + + +def stage_delivery_fact_no_commit( + db: Session, *, task_id: int, turn_id: str, status: str +) -> None: + if uses_execution_events(db, task_id): + append_fact_no_commit( + db, + task_id=task_id, + kind="input_delivery_changed", + key=f"delivery:{turn_id}:{status}", + turn_id=turn_id, + payload={"status": status}, + ) + + +def stage_result_fact_no_commit( + db: Session, task: Task, result: dict[str, Any] +) -> None: + if task.conversation_storage_version == 2: + append_fact_no_commit( + db, + task_id=int(task.id), + kind="execution_settled", + key=f"result:{task.run_id}:{task.status.value}", + run_id=cast(str | None, task.run_id), + payload={"status": task.status.value, "result": result}, + ) + + +def stage_applied_inputs_no_commit(db: Session, state: TaskExecutionEvent) -> None: + """Record first application alongside the durable state that proves it.""" + snapshot = state.payload["data"]["snapshot"] + context = snapshot.get("context") or {} + for message in context.get("messages", []): + if message.get("role") != "user": + continue + turn_id = (message.get("metadata") or {}).get("turn_id") + if not turn_id: + continue + key = f"input-applied:{turn_id}" + existing = db.scalar( + select(TaskExecutionEvent.id).where( + TaskExecutionEvent.task_id == state.task_id, + TaskExecutionEvent.scope_id == state.scope_id, + TaskExecutionEvent.idempotency_key == key, + ) + ) + if existing is None: + append_fact_no_commit( + db, + task_id=int(state.task_id), + scope_id=str(state.scope_id), + run_id=cast(str | None, state.run_id), + turn_id=turn_id, + kind="input_applied", + key=key, + payload={"recovery_event_id": state.event_id}, + ) diff --git a/src/xagent/web/services/task_interaction_staging.py b/src/xagent/web/services/task_interaction_staging.py index a22a4ff3a..d325592db 100644 --- a/src/xagent/web/services/task_interaction_staging.py +++ b/src/xagent/web/services/task_interaction_staging.py @@ -1146,6 +1146,28 @@ def stage_interaction_request( "request in a different slot" ) else: + from .task_execution_event_writer import ( + append_fact_no_commit, + uses_execution_events, + ) + + if uses_execution_events(db, resolved_task_id): + append_fact_no_commit( + db, + task_id=resolved_task_id, + run_id=run_id, + kind="interaction_requested", + key=f"interaction:{new_row.id}", + payload={ + "interaction_id": int(new_row.id), + "kind": kind, + "protocol_version": protocol_version, + "origin": origin, + "request": request_payload, + "request_idempotency_key": normalized_key, + "expires_at": expires_at.isoformat(), + }, + ) inner.commit() return StagedInteractionRequest( staged_db_id=int(new_row.id), diff --git a/src/xagent/web/services/task_orchestrator.py b/src/xagent/web/services/task_orchestrator.py index e133ead47..aacadbb88 100644 --- a/src/xagent/web/services/task_orchestrator.py +++ b/src/xagent/web/services/task_orchestrator.py @@ -1479,6 +1479,11 @@ def settle_task_lease_isolated( content=client_error_message, message_type=client_message_type, ) + from .task_execution_event_writer import stage_result_fact_no_commit + + stage_result_fact_no_commit( + settle_db, task, {"error": error_message} + ) settle_db.commit() invalidate_task_cache_best_effort(lease.task_id) return True diff --git a/src/xagent/web/tracing.py b/src/xagent/web/tracing.py index 6a5805b36..6ff20762c 100644 --- a/src/xagent/web/tracing.py +++ b/src/xagent/web/tracing.py @@ -2,12 +2,14 @@ from __future__ import annotations +import asyncio from typing import Any, Optional from ..core.agent.checkpoint import READABLE_CHECKPOINT_TYPES from ..core.agent.trace import ( BaseTraceHandler, ConsoleTraceHandler, + ExecutionEventPersistenceError, ) from ..core.agent.trace import TraceEvent as CoreTraceEvent from ..core.agent.trace import ( @@ -52,6 +54,45 @@ async def load_latest_checkpoint( return dict(snapshot) if isinstance(snapshot, dict) else None +class ExecutionEventTraceAdapter(DatabaseTraceHandler): + """Strict fact writer plus the transitional checkpoint reader. + + Normal observer dispatch must never write the same event a second time. + """ + + def __init__(self, task_id: int, build_id: str | None = None) -> None: + super().__init__(task_id, build_id=build_id) + self.authoritative = True + + async def handle_event(self, event: CoreTraceEvent) -> None: + pass + + async def commit_event(self, event: CoreTraceEvent) -> None: + required = event.require_persisted + event.require_persisted = True + try: + await asyncio.to_thread(self._sync_save_to_database, event) + except Exception as exc: + raise ExecutionEventPersistenceError( + "Conversation event commit failed" + ) from exc + finally: + event.require_persisted = required + + +def task_database_handler( + task_id: int, build_id: str | None = None +) -> DatabaseTraceHandler: + from .models.database import get_session_local + from .services.task_execution_event_writer import uses_execution_events + + with get_session_local()() as db: + canonical = uses_execution_events(db, task_id) + if canonical: + return ExecutionEventTraceAdapter(task_id, build_id=build_id) + return DatabaseTraceHandler(task_id, build_id=build_id) + + def create_task_tracer( task_id: int, user: Optional[User] = None, @@ -64,10 +105,11 @@ def create_task_tracer( if user is not None and user.id is not None: resolved_user_id = int(user.id) - return create_agent_tracer( + database_handler = task_database_handler(task_id) + tracer = create_agent_tracer( handlers=[ ConsoleTraceHandler(), - DatabaseTraceHandler(task_id), + database_handler, WebSocketTraceHandler(task_id), ], task_id=str(task_id), @@ -82,6 +124,10 @@ def create_task_tracer( }, ) + if isinstance(database_handler, ExecutionEventTraceAdapter): + tracer.event_writer = database_handler.commit_event + return tracer + def create_ephemeral_tracer( *, diff --git a/tests/core/agent/test_react.py b/tests/core/agent/test_react.py index 85b8d1426..54c5db277 100644 --- a/tests/core/agent/test_react.py +++ b/tests/core/agent/test_react.py @@ -8824,3 +8824,19 @@ async def test_react_summarizes_with_the_main_model_when_no_compact_model() -> N # One routing decision for the whole turn, taken on the conversation. assert len(route_prompts) == 1 assert "Conversation history to compact" not in route_prompts[0] + + +@pytest.mark.asyncio +async def test_nested_fact_failure_is_not_a_retryable_tool_error() -> None: + from xagent.core.agent.trace import ExecutionEventPersistenceError + + class UncertainTool(FakeTool): + async def run_json_async(self, args: dict[str, Any]) -> Any: + raise ExecutionEventPersistenceError("child event commit failed") + + with pytest.raises(ExecutionEventPersistenceError): + await ReActPattern()._execute_tool_safely( + {"id": "child1", "name": "calculator", "args": {"expression": "2+2"}}, + [UncertainTool()], + PatternRuntime(), + ) diff --git a/tests/core/agent/test_runner.py b/tests/core/agent/test_runner.py index 08d912384..3da40daf8 100644 --- a/tests/core/agent/test_runner.py +++ b/tests/core/agent/test_runner.py @@ -2152,3 +2152,23 @@ def test_resume_migration_reaches_a_nested_auto_pattern_child_context() -> None: nested = checkpoint["pattern_state"]["dag_state"]["active_step_contexts"]["step_1"] assert OUTPUT_LANGUAGE_METADATA_KEY not in nested["metadata"] + + +@pytest.mark.asyncio +async def test_runner_does_not_try_another_pattern_after_fact_commit_failure() -> None: + from xagent.core.agent.trace import ExecutionEventPersistenceError + + class UncertainPattern: + async def run(self, **_: Any) -> dict[str, Any]: + raise ExecutionEventPersistenceError("tool result commit failed") + + fallback = FakePattern({"success": True, "output": "must not retry"}) + runner = AgentRunner( + agent=Agent( + name="fact-failure", patterns=[UncertainPattern(), fallback], llm=None + ), + workspace_enabled=False, + ) + with pytest.raises(ExecutionEventPersistenceError): + await runner.run(task="write once", execution_id="uncertain-effect") + assert fallback.calls == [] diff --git a/tests/core/tools/test_create_agent_tool.py b/tests/core/tools/test_create_agent_tool.py index c97d4ad9d..887c9144a 100644 --- a/tests/core/tools/test_create_agent_tool.py +++ b/tests/core/tools/test_create_agent_tool.py @@ -38,6 +38,15 @@ from xagent.web.models.user import User +@pytest.fixture +def legacy_task_selection(monkeypatch): + from xagent.web.api.trace_handlers import DatabaseTraceHandler + + monkeypatch.setattr( + "xagent.web.tracing.task_database_handler", DatabaseTraceHandler + ) + + def _create_session() -> tuple[Session, str, Any]: """Create a temporary database session for testing. @@ -91,7 +100,7 @@ async def handle_event(self, event: Any) -> None: assert event.data is original_data -def test_agent_tool_child_tracer_persists_and_broadcasts() -> None: +def test_agent_tool_child_tracer_persists_and_broadcasts(legacy_task_selection) -> None: tool = AgentTool( agent_id=17, agent_name="Video Generation Agent", @@ -645,6 +654,7 @@ def test_published_agent_builder_propagates_file_operation_policy(self) -> None: @pytest.mark.asyncio async def test_agent_tool_returns_parent_owned_file_refs_for_worker_outputs( self, + legacy_task_selection, ) -> None: db, db_path, SessionLocal = _create_session() try: @@ -857,6 +867,7 @@ async def trace_event( @pytest.mark.asyncio async def test_agent_tool_classified_failure_does_not_register_file_outputs( self, + legacy_task_selection, ) -> None: db, db_path, SessionLocal = _create_session() try: diff --git a/tests/migrations/test_20260905_enable_task_execution_event_writers.py b/tests/migrations/test_20260905_enable_task_execution_event_writers.py new file mode 100644 index 000000000..a46f303fe --- /dev/null +++ b/tests/migrations/test_20260905_enable_task_execution_event_writers.py @@ -0,0 +1,90 @@ +from pathlib import Path + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + +from tests.migrations.test_20260905_add_task_execution_events import ( + MIGRATION as STORAGE, +) +from tests.migrations.test_20260905_add_task_execution_events import ( + engine as engine_fixture, +) +from tests.shared.postgres_disposable import load_migration_module + +engine = engine_fixture + +WRITERS = load_migration_module( + Path(__file__).parents[2] + / "src/xagent/migrations/versions/20260905_enable_task_execution_event_writers.py" +) + + +def run(connection, migration, operation): + with Operations.context(MigrationContext.configure(connection)): + getattr(migration, operation)() + + +def test_upgrade_preserves_tasks_and_inbound_cascade_rows(engine): + with engine.begin() as db: + db.execute(sa.text("CREATE TABLE tasks (id INTEGER PRIMARY KEY)")) + db.execute( + sa.text( + "CREATE TABLE task_chat_messages (id INTEGER PRIMARY KEY, task_id INTEGER REFERENCES tasks(id) ON DELETE CASCADE)" + ) + ) + db.execute(sa.text("INSERT INTO tasks VALUES (1)")) + db.execute(sa.text("INSERT INTO task_chat_messages VALUES (1, 1)")) + run(db, STORAGE, "upgrade") + run(db, WRITERS, "upgrade") + assert db.scalar(sa.text("SELECT count(*) FROM task_chat_messages")) == 1 + assert db.scalar(sa.text("SELECT conversation_storage_version FROM tasks")) == 1 + db.execute( + sa.text("INSERT INTO tasks(id, conversation_storage_version) VALUES (2, 2)") + ) + # create_all parity / idempotent re-entry must not reset canonical tasks. + run(db, WRITERS, "upgrade") + assert ( + db.scalar( + sa.text("SELECT conversation_storage_version FROM tasks WHERE id=2") + ) + == 2 + ) + with pytest.raises(RuntimeError, match="event-backed"): + run(db, WRITERS, "downgrade") + db.execute(sa.text("DELETE FROM tasks WHERE id=2")) + run(db, WRITERS, "downgrade") + assert db.scalar(sa.text("SELECT count(*) FROM task_chat_messages")) == 1 + run(db, WRITERS, "upgrade") + db.execute(sa.text("INSERT INTO tasks(id) VALUES (3)")) + assert ( + db.scalar( + sa.text("SELECT conversation_storage_version FROM tasks WHERE id=3") + ) + == 1 + ) + + +@pytest.mark.parametrize("with_chat_table", [False, True]) +def test_upgrade_downgrade_without_metadata_owned_tasks(engine, with_chat_table): + with engine.begin() as db: + if with_chat_table: + db.execute( + sa.text("CREATE TABLE task_chat_messages (id INTEGER PRIMARY KEY)") + ) + run(db, STORAGE, "upgrade") + run(db, WRITERS, "upgrade") + if with_chat_table: + assert "execution_event_id" in { + column["name"] + for column in sa.inspect(db).get_columns("task_chat_messages") + } + run(db, WRITERS, "downgrade") + run(db, STORAGE, "downgrade") + assert not sa.inspect(db).has_table("tasks") + if with_chat_table: + assert [ + column["name"] + for column in sa.inspect(db).get_columns("task_chat_messages") + ] == ["id"] diff --git a/tests/web/api/client_safe_ast_guard.py b/tests/web/api/client_safe_ast_guard.py index 3c073e707..7d1bfdf6e 100644 --- a/tests/web/api/client_safe_ast_guard.py +++ b/tests/web/api/client_safe_ast_guard.py @@ -154,6 +154,38 @@ def trusted_builder(candidate: ast.expr, result_index: int | None) -> bool: and not _has_local_binding(candidate, candidate.func.id, parents) ) + if isinstance(expr, ast.Name): + # A literal non-error type set narrows this name only when it has no + # stores anywhere in the guarded body (including nested branches). + current = reference + while current in parents: + parent = parents[current] + if isinstance(parent, ast.If) and current in parent.body: + test = parent.test + if ( + isinstance(test, ast.Compare) + and isinstance(test.left, ast.Name) + and test.left.id == expr.id + and len(test.ops) == 1 + and isinstance(test.ops[0], ast.In) + and isinstance(test.comparators[0], (ast.Set, ast.Tuple, ast.List)) + and all( + isinstance(value, ast.Constant) + and isinstance(value.value, str) + and value.value not in ERROR_PAYLOAD_TYPES + for value in test.comparators[0].elts + ) + and not any( + isinstance(node, ast.Name) + and node.id == expr.id + and isinstance(node.ctx, (ast.Store, ast.Del)) + for statement in parent.body + for node in ast.walk(statement) + ) + ): + return True + current = parent + if not isinstance(expr, ast.Name): return isinstance(expr, ast.expr) and trusted_builder(expr, None) @@ -203,7 +235,7 @@ def _dict_variants( if ( isinstance(expr, ast.Call) and isinstance(expr.func, ast.Name) - and expr.func.id == "create_stream_event" + and expr.func.id in {"create_stream_event", "create_final_answer_stream_event"} and expr.func.id in module_helpers and not _has_local_binding(expr, expr.func.id, parents) ): @@ -320,7 +352,7 @@ def _error_payload_messages( return [argument] message = _call_argument(argument, 1, "message") return [message if message is not None else argument] - if helper != "create_stream_event": + if helper not in {"create_stream_event", "create_final_answer_stream_event"}: return [] if not isinstance(argument, (ast.Call, ast.Dict, ast.Name, ast.Await)): return [] @@ -1080,6 +1112,7 @@ def _trusted_module_helpers(tree: ast.Module) -> set[str]: *DICT_ERROR_PAYLOAD_BUILDERS, *NON_ERROR_STREAM_EVENT_BUILDERS, "create_stream_event", + "create_final_answer_stream_event", } overload_bindings = [ (node, alias) diff --git a/tests/web/api/test_websocket_client_safe_errors.py b/tests/web/api/test_websocket_client_safe_errors.py index acdbcea21..011c604ac 100644 --- a/tests/web/api/test_websocket_client_safe_errors.py +++ b/tests/web/api/test_websocket_client_safe_errors.py @@ -3156,3 +3156,25 @@ def test_registry_is_bounded_by_lru_eviction(_clean_origins: None) -> None: websocket_api.manager, "is_connection_registered", return_value=True ): assert origins.resolve("cmd-0", 7) is None # evicted -> safe discard + + +@pytest.mark.parametrize( + ("types", "reassignment", "blocked"), + [ + ('"final_answer_start", "final_answer_error"', "", False), + ('"final_answer_start", "error"', "", True), + ('"final_answer_start"', 'kind = "error"', True), + ], +) +def test_assigned_final_answer_envelope_type_narrowing(types, reassignment, blocked): + source = f""" +def create_final_answer_stream_event(event_type, task_id, data): + return {{"type": event_type, **data}} + +async def send(kind, raw): + if kind in {{{types}}}: + {reassignment or "pass"} + envelope = create_final_answer_stream_event(kind, 1, {{"message": str(raw)}}) + await manager.broadcast_to_task(envelope, 1) +""" + assert bool(_guard_offenders(source)) is blocked diff --git a/tests/web/services/test_task_execution_event_store.py b/tests/web/services/test_task_execution_event_store.py index e45687240..05518372c 100644 --- a/tests/web/services/test_task_execution_event_store.py +++ b/tests/web/services/test_task_execution_event_store.py @@ -76,7 +76,7 @@ def test_defaults_pin_legacy_without_events(engine, task_id): assert task.conversation_storage_version == 1 assert task.conversation_event_sequence == 0 assert load_task_execution_events(db, task_id=task_id, scope_id="root") == [] - task.conversation_storage_version = 2 + task.conversation_storage_version = 3 with pytest.raises(IntegrityError): db.flush() diff --git a/tests/web/services/test_task_execution_event_writer.py b/tests/web/services/test_task_execution_event_writer.py new file mode 100644 index 000000000..fd7c36cdb --- /dev/null +++ b/tests/web/services/test_task_execution_event_writer.py @@ -0,0 +1,474 @@ +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest +import sqlalchemy as sa +from sqlalchemy.orm import sessionmaker + +from tests.web.services.test_task_execution_event_store import engine as engine_fixture +from tests.web.services.test_task_execution_event_store import ( + task_id as task_id_fixture, +) +from xagent.core.agent.checkpoint import TraceCheckpointStore +from xagent.core.agent.runtime import PatternRuntime +from xagent.core.agent.trace import ( + ExecutionEventPersistenceError, + TraceAction, + TraceCategory, + TraceEventType, + TraceScope, +) +from xagent.web.models.chat_message import TaskChatMessage +from xagent.web.models.task import Task, TaskStatus, TraceEvent +from xagent.web.models.task_execution_event import TaskExecutionEvent +from xagent.web.services.chat_history_service import persist_user_message_no_commit +from xagent.web.services.managed_task_lease import finalize_managed_task_lease_result +from xagent.web.services.task_execution_controller import ( + TaskControlState, + apply_task_control_transition, +) +from xagent.web.services.task_lease_service import acquire_task_lease +from xagent.web.tracing import ExecutionEventTraceAdapter, create_task_tracer + +engine = engine_fixture +task_id = task_id_fixture + + +@pytest.fixture +def canonical(engine, task_id, monkeypatch): + factory = sessionmaker(engine) + with factory() as db: + task = db.get(Task, task_id) + task.conversation_storage_version = 2 + db.commit() + monkeypatch.setattr("xagent.web.models.database.get_session_local", lambda: factory) + monkeypatch.setattr( + "xagent.web.api.trace_handlers.get_db", lambda: iter([factory()]) + ) + return factory, task_id + + +def facts(db, task_id): + return list( + db.scalars( + sa.select(TaskExecutionEvent) + .where( + TaskExecutionEvent.task_id == task_id, + ) + .order_by(TaskExecutionEvent.sequence) + ) + ) + + +def test_acceptance_and_compatibility_row_share_transaction(canonical): + factory, task_id = canonical + with factory() as db: + task = db.get(Task, task_id) + user_id = task.user_id + persist_user_message_no_commit( + db, task_id, user_id, "hello", turn_id="t1", attachments=[] + ) + assert [e.kind for e in facts(db, task_id)] == ["input_accepted"] + db.rollback() + assert facts(db, task_id) == [] + assert db.query(TaskChatMessage).count() == 0 + first = persist_user_message_no_commit( + db, task_id, user_id, "hello", turn_id="t1", attachments=[] + ) + db.commit() + second = persist_user_message_no_commit( + db, task_id, user_id, "hello", turn_id="t1", attachments=[] + ) + db.commit() + assert first.id == second.id + assert len(facts(db, task_id)) == 1 + assert second.attachments == [] + + +@pytest.mark.asyncio +async def test_factory_commits_recoverable_state_before_observers(canonical): + factory, task_id = canonical + tracer = create_task_tracer(task_id) + assert tracer.records_execution_events + assert any(isinstance(h, ExecutionEventTraceAdapter) for h in tracer.handlers) + observer = AsyncMock() + tracer.handlers = [observer] + payload = { + "execution_id": "run-root", + "pattern": "ReActPattern", + "label": "after_llm", + "context": { + "messages": [ + {"role": "user", "content": "hello", "metadata": {"turn_id": "t1"}} + ] + }, + "pattern_state": {"adopted_plan": "完整计划" * 10000, "pending_tool_calls": []}, + } + await TraceCheckpointStore(tracer).save(payload) + with factory() as db: + rows = facts(db, task_id) + assert [e.kind for e in rows] == ["recovery_state", "input_applied"] + assert rows[0].payload["data"]["snapshot"] == payload + assert rows[1].payload["recovery_event_id"] == rows[0].event_id + assert db.query(TraceEvent).count() == 1 + observer.handle_event.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_failed_fact_commit_stops_runtime_before_broadcast( + canonical, monkeypatch +): + _, task_id = canonical + tracer = create_task_tracer(task_id) + observer = AsyncMock() + tracer.handlers = [observer] + + def fail(*args, **kwargs): + raise OSError("database unavailable") + + monkeypatch.setattr( + "xagent.web.services.task_execution_event_writer.append_task_execution_event_no_commit", + fail, + ) + runtime = PatternRuntime(tracer=TraceCheckpointStore(tracer)) + with pytest.raises(ExecutionEventPersistenceError): + await runtime.on_tool_start( + tool_call={ + "id": "call1", + "name": "write", + "args": {}, + "tool_attempt_id": "attempt1", + "assistant_message_id": "batch1", + } + ) + observer.handle_event.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_observer_failure_does_not_invalidate_fact(canonical): + factory, task_id = canonical + tracer = create_task_tracer(task_id) + tracer.handlers = [ + AsyncMock(handle_event=AsyncMock(side_effect=OSError("socket closed"))) + ] + await tracer.trace_event( + TraceEventType(TraceScope.TASK, TraceAction.START, TraceCategory.GENERAL), + task_id=str(task_id), + data={"input": "hello"}, + require_persisted=True, + ) + with factory() as db: + assert len(facts(db, task_id)) == 1 + + +@pytest.mark.asyncio +async def test_attempt_result_keeps_batch_identity_and_blocks_blind_replay(canonical): + factory, task_id = canonical + tracer = create_task_tracer(task_id) + tracer.handlers = [] + runtime = PatternRuntime(tracer=TraceCheckpointStore(tracer)) + call = { + "id": "provider-duplicate-id", + "name": "write", + "args": {"value": 1}, + "tool_attempt_id": "attempt1", + "assistant_message_id": "batch1", + } + await runtime.on_tool_start(tool_call=call) + await runtime.on_tool_end( + tool_call=call, result={"success": True, "output": "长结果" * 10000} + ) + with pytest.raises(ExecutionEventPersistenceError): + await runtime.on_tool_start(tool_call=call) + with factory() as db: + rows = facts(db, task_id) + assert len(rows) == 2 + assert {r.assistant_message_id for r in rows} == {"batch1"} + assert {r.tool_attempt_id for r in rows} == {"attempt1"} + assert rows[1].payload["data"]["result"]["output"] == "长结果" * 10000 + + +@pytest.mark.parametrize( + "status", + [ + TaskStatus.COMPLETED, + TaskStatus.FAILED, + TaskStatus.PAUSED, + TaskStatus.WAITING_FOR_USER, + ], +) +def test_channel_outcome_and_transcript_commit_with_lease(canonical, status): + factory, task_id = canonical + with factory() as db: + lease = acquire_task_lease(db, task_id, new_run=True) + assert lease is not None + assert finalize_managed_task_lease_result( + db, + lease, + status=status, + assistant_content="result", + execution_result={"output": "result"}, + ) + rows = facts(db, task_id) + assert rows[-1].kind == "execution_settled" + assert rows[-1].payload["status"] == status.value + assert any(row.kind == "assistant_message" for row in rows) + assert db.get(Task, task_id).runner_id is None + + +def test_control_and_event_rollback_together(canonical): + factory, task_id = canonical + with factory() as db: + task = db.get(Task, task_id) + apply_task_control_transition( + task, TaskControlState.PAUSED, status=TaskStatus.PAUSED + ) + assert facts(db, task_id)[-1].kind == "control_state_changed" + db.rollback() + assert facts(db, task_id) == [] + assert db.get(Task, task_id).status == TaskStatus.PENDING + + +@pytest.mark.asyncio +async def test_real_react_loop_records_batch_before_tool_and_retains_identity( + canonical, +): + from tests.core.agent.test_react import FakeLLM, FakeTool + from xagent.core.agent import ExecutionContext, ReActPattern + + factory, task_id = canonical + tracer = create_task_tracer(task_id) + tracer.handlers = [] + runtime = PatternRuntime(tracer=TraceCheckpointStore(tracer)) + context = ExecutionContext(system_prompt="Calculate") + context.add_user_message("2+2", metadata={"turn_id": "real-turn"}) + tool = FakeTool() + pattern = ReActPattern(max_iterations=3) + result = await pattern.run( + context=context, + tools=[tool], + runtime=runtime, + llm=FakeLLM( + responses=[ + { + "content": "calculate", + "tool_calls": [ + { + "id": "call1", + "function": { + "name": "calculator", + "arguments": '{"expression":"2+2"}', + }, + } + ], + }, + {"content": "4", "done": True}, + ] + ), + ) + assert result["success"] + assert len(tool.calls) == 1 + with factory() as db: + rows = facts(db, task_id) + tools = [row for row in rows if row.kind.startswith("tool_execution_")] + state = next( + row + for row in rows + if row.kind == "recovery_state" + and row.payload["data"]["snapshot"]["label"] == "after_llm" + ) + snapshot = state.payload["data"]["snapshot"] + saved_call = snapshot["pattern_state"]["pending_tool_calls"][0] + assert state.sequence < tools[0].sequence < tools[1].sequence + assert {row.tool_attempt_id for row in tools} == {saved_call["tool_attempt_id"]} + assert {row.assistant_message_id for row in tools} == { + saved_call["assistant_message_id"] + } + restored = ReActPattern() + restored.load_state(snapshot["pattern_state"]) + assert ( + restored.pending_tool_calls[0]["tool_attempt_id"] + == saved_call["tool_attempt_id"] + ) + + +@pytest.mark.asyncio +async def test_outbound_stream_is_committed_before_websocket_and_failure_is_strict( + canonical, monkeypatch +): + from xagent.web.api import websocket + + factory, task_id = canonical + monkeypatch.setattr(websocket, "get_db", lambda: iter([factory()])) + broadcasts = [] + + async def broadcast(event, task_id): + with factory() as db: + assert facts(db, task_id)[-1].payload["data"]["delta"] == "hello" + broadcasts.append(event) + + monkeypatch.setattr(websocket.manager, "broadcast_to_task", broadcast) + handler = websocket.make_agent_outbound_handler(task_id, authoritative=True) + payload = {"type": "final_answer_delta", "delta": "hello", "stream_id": "stream1"} + await handler(payload) + assert len(broadcasts) == 1 + + def fail(*args, **kwargs): + raise OSError("write failure") + + monkeypatch.setattr( + "xagent.web.services.task_execution_event_writer.append_task_execution_event_no_commit", + fail, + ) + with pytest.raises(ExecutionEventPersistenceError): + await handler(payload) + assert len(broadcasts) == 1 + + +def test_compatibility_failure_rolls_back_outcome_and_retains_lease( + canonical, monkeypatch +): + factory, task_id = canonical + with factory() as db: + lease = acquire_task_lease(db, task_id, new_run=True) + + def fail(*args, **kwargs): + raise OSError("event storage unavailable") + + monkeypatch.setattr( + "xagent.web.services.task_execution_event_writer.append_task_execution_event_no_commit", + fail, + ) + with pytest.raises(OSError): + finalize_managed_task_lease_result( + db, lease, status=TaskStatus.COMPLETED, assistant_content="result" + ) + db.expire_all() + task = db.get(Task, task_id) + assert task.status == TaskStatus.RUNNING + assert task.runner_id == lease.runner_id + assert db.query(TaskChatMessage).count() == 0 + assert facts(db, task_id) == [] + + +@pytest.mark.asyncio +async def test_replaced_lease_cannot_append_or_broadcast(canonical): + from xagent.web.services.task_lease_service import bind_task_lease_context + + factory, task_id = canonical + with factory() as db: + lease = acquire_task_lease(db, task_id, new_run=True) + db.get(Task, task_id).run_id = "replacement-run" + db.commit() + tracer = create_task_tracer(task_id) + observer = AsyncMock() + tracer.handlers = [observer] + with bind_task_lease_context(lease): + with pytest.raises(ExecutionEventPersistenceError): + await tracer.trace_event( + TraceEventType( + TraceScope.TASK, TraceAction.START, TraceCategory.GENERAL + ), + task_id=str(task_id), + ) + observer.handle_event.assert_not_awaited() + with factory() as db: + assert facts(db, task_id) == [] + + +def test_assistant_projection_replay_has_one_fact_and_one_row(canonical): + from xagent.web.services.chat_history_service import ( + persist_assistant_message_no_commit, + ) + + factory, task_id = canonical + with factory() as db: + task = db.get(Task, task_id) + task.run_id = "run1" + task.status = TaskStatus.COMPLETED + db.commit() + for _ in range(2): + persist_assistant_message_no_commit( + db, + task_id, + task.user_id, + "done", + message_type="assistant_response", + content_is_reconciled=True, + ) + db.commit() + rows = facts(db, task_id) + assert len(rows) == 1 + assert db.query(TaskChatMessage).one().execution_event_id == rows[0].event_id + + +def test_command_fact_and_inbox_are_one_transaction(canonical): + from xagent.web.models.task_command import TaskExecutionCommand + from xagent.web.services.task_command_transport import ( + TaskCommandKind, + stage_task_command, + ) + + factory, task_id = canonical + with factory() as db: + task = db.get(Task, task_id) + stage_task_command( + db, + task_id=task_id, + actor_user_id=task.user_id, + command_id="answer1", + kind=TaskCommandKind.RESUME, + payload={"response": "yes"}, + ) + assert facts(db, task_id)[-1].payload["payload"] == {"response": "yes"} + db.rollback() + assert facts(db, task_id) == [] + assert db.query(TaskExecutionCommand).count() == 0 + + +def test_pre_runner_failure_is_a_fact_and_failed_commit_is_not_broadcastable( + canonical, monkeypatch +): + from xagent.web.api.websocket import _terminal_task_error_payload + + factory, task_id = canonical + monkeypatch.setattr("xagent.web.api.websocket.get_session_local", lambda: factory) + _terminal_task_error_payload(task_id, "sandbox unavailable") + with factory() as db: + rows = facts(db, task_id) + assert rows[-1].kind == "execution_settled" + assert rows[-1].payload["result"]["error"] == "sandbox unavailable" + assert rows[-1].payload["status"] == TaskStatus.FAILED.value + + def fail(*args, **kwargs): + raise OSError("commit failure") + + monkeypatch.setattr( + "xagent.web.services.task_execution_event_writer.append_task_execution_event_no_commit", + fail, + ) + with pytest.raises(ExecutionEventPersistenceError): + _terminal_task_error_payload(task_id, "sandbox unavailable") + + +def test_settlement_serializes_execution_context_without_losing_state(canonical): + from xagent.core.agent import ExecutionContext + from xagent.web.services.task_execution_event_writer import ( + stage_result_fact_no_commit, + ) + + factory, task_id = canonical + context = ExecutionContext(execution_id="run-context") + context.add_assistant_message("完整回复" * 10000) + with factory() as db: + task = db.get(Task, task_id) + task.status = TaskStatus.COMPLETED + stage_result_fact_no_commit(db, task, {"agent_result": {"context": context}}) + db.commit() + assert ( + facts(db, task_id)[0].payload["result"]["agent_result"]["context"] + == context.to_dict() + ) + with pytest.raises(TypeError, match="Unsupported execution fact"): + stage_result_fact_no_commit(db, task, {"unknown": object()}) diff --git a/tests/web/services/test_task_interaction_service.py b/tests/web/services/test_task_interaction_service.py index 5cc616f09..a1ef7d1a0 100644 --- a/tests/web/services/test_task_interaction_service.py +++ b/tests/web/services/test_task_interaction_service.py @@ -6023,17 +6023,18 @@ def _before_cursor_execute( return [_leading_keyword(s) for s in statements] -def test_statement_sequence_is_unchanged_by_the_receipt_widening( +def test_statement_sequence_includes_execution_event_version_check( _db: Session, _system_call_ctx: dict[str, Any] ) -> None: - """Widening _identity_lookup_stmt's column list (this delivery's own - change) adds no new statement and reorders nothing: a fresh insert - still runs exactly this sequence -- the SQLite-only dummy UPDATE, the + """A fresh insert includes the storage-version query added by event writers. + + It runs exactly this sequence -- the SQLite-only dummy UPDATE, the outer savepoint interaction_handoff opens, the identity SELECT stage_interaction_request's step 3 runs (now widened, same statement), the reclaim UPDATE (step 4), the inner savepoint stage_interaction_request opens for its own INSERT (step 5), the - INSERT itself, and the two savepoints releasing in reverse order. + INSERT itself, the execution-event storage version SELECT, and the two + savepoints releasing in reverse order. Asserted by leading-keyword shape, in order -- not just a count -- against a real, captured sequence, not an estimate.""" @@ -6047,6 +6048,7 @@ def test_statement_sequence_is_unchanged_by_the_receipt_widening( "UPDATE", # reclaim stale/superseded slot, step 4 "SAVEPOINT", # inner savepoint, step 5 "INSERT", # the new row, step 5 + "SELECT", # choose legacy or execution-event persistence "RELEASE SAVEPOINT", # inner savepoint commits "RELEASE SAVEPOINT", # outer savepoint commits ] diff --git a/tests/web/test_agent_manager_reconstruction.py b/tests/web/test_agent_manager_reconstruction.py index 475f28b5a..7c0ff429c 100644 --- a/tests/web/test_agent_manager_reconstruction.py +++ b/tests/web/test_agent_manager_reconstruction.py @@ -24,6 +24,15 @@ ) +@pytest.fixture(autouse=True) +def legacy_task_selection(monkeypatch): + from xagent.web.api.trace_handlers import DatabaseTraceHandler + + monkeypatch.setattr( + "xagent.web.tracing.task_database_handler", DatabaseTraceHandler + ) + + def _build_reconstruction_snapshot( task: Task, user: User, diff --git a/tests/web/test_tracing_factory.py b/tests/web/test_tracing_factory.py index 675140f95..15ed69baa 100644 --- a/tests/web/test_tracing_factory.py +++ b/tests/web/test_tracing_factory.py @@ -4,6 +4,8 @@ from typing import cast +import pytest + from tests.utils.mock_helpers import create_langfuse_mock from xagent.core.agent.trace import TraceEvent, TraceHandler from xagent.core.tracing.langfuse.client import get_langfuse_client @@ -12,6 +14,15 @@ from xagent.web.tracing import create_ephemeral_tracer, create_task_tracer +@pytest.fixture(autouse=True) +def legacy_task_selection(monkeypatch): + from xagent.web.api.trace_handlers import DatabaseTraceHandler + + monkeypatch.setattr( + "xagent.web.tracing.task_database_handler", DatabaseTraceHandler + ) + + class DummyTraceHandler(TraceHandler): async def handle_event(self, event: TraceEvent) -> None: del event