feat(persistence): wire execution event writers (stage 3.2)

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.

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)
 

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,

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,

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(

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