From b6c20f595a78462bdd65a4dc5335f9de1c425532 Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:14:24 +0800
Subject: [PATCH 1/3] refactor(events): fold Todo replay through bounded typed
 plans

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../control_plane/effect_runtime_handlers.ts  |   2 +
 .../control_plane/goals/state_event_replay.ts | 175 ++++++++++++++++
 loopx/event_sourced_state.py                  | 190 +++++++++---------
 3 files changed, 271 insertions(+), 96 deletions(-)
 create mode 100644 loopx/control_plane/goals/state_event_replay.ts

diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 14477e78a3..730c3b9c8a 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -1,3 +1,4 @@
+import {planStateEventReplay} from "./goals/state_event_replay.ts";
 import {projectTodoSummary} from "./todos/summary_projection.ts";
 import {admitAutomationStart, confirmAutomationStart, manageAutomationCadence, projectCadenceSchedule} from "./quota/automation_cadence.ts";
 import {deliverShadowEntry} from "./coordination/shadow_entry_delivery.ts";
@@ -438,6 +439,7 @@ export function createEffectRuntimeHandlers(
       (params) => interpretTurnJournal(turnJournalInspectionRequest(params)),
     ],
     ["turn_journal.write", commitTurnJournal],
+    ["goal.state_event.plan_replay", planStateEventReplay],
     ["todo.completion_fence.evaluate", evaluateTodoCompletionFence],
     ["todo.completion_state.normalize", normalizeTodoCompletionValue],
     ["todo.completion_state.require_metadata", requireTodoCompletionMetadataValue],
diff --git a/loopx/control_plane/goals/state_event_replay.ts b/loopx/control_plane/goals/state_event_replay.ts
new file mode 100644
index 0000000000..3e66c749f8
--- /dev/null
+++ b/loopx/control_plane/goals/state_event_replay.ts
@@ -0,0 +1,175 @@
+/** Replay admission and lifecycle projection for the retained event source.
+ * Content stays in the host codec: ordinals address that exact input batch.
+ * This pure plan grants no append, lease, capture or promotion authority. */
+import type {JsonObject} from "../effect_program.ts";
+import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
+import {assertNever, requireJsonObject, requireNonEmptyString, requireStringLiteral, requireStringArray} from "../runtime_decode.ts";
+import {authorityUnicodeCompare} from "../coordination/authority_store_codec.ts";
+import {normalizeTodoPriority} from "../todos/priority.ts";
+
+const KINDS = ["todo_added", "todo_claimed", "todo_updated", "todo_blocked", "todo_deferred",
+  "todo_completed", "refresh_recorded", "run_recorded", "quota_spent", "evidence_attached",
+  "supervisor_proposed", "supervisor_receipt_recorded"] as const;
+type Kind = typeof KINDS[number];
+type Status = "open" | "blocked" | "deferred" | "done";
+interface Event {
+  ordinal: number; id: string; goal: string; kind: Kind; sequence: number | null;
+  time: string; todo: string | null; role: "user" | "agent" | null;
+  priority: string | null; plannerOrder: number | null; contentChanged: boolean;
+  fields: string[]; binding: string | null; continuation: string | null;
+  removedPolicy: string | null; exclusions: boolean; goalBound: boolean | null;
+}
+interface Todo {
+  todo_id: string; field_sources: Record<string, number>; status: Status; done: boolean;
+  role: "user" | "agent"; priority: string; planner_order: number | null;
+  source_section: string; render_priority: boolean; append_sequence: number | null;
+  binding: string | null; removedPolicy: string | null;
+}
+function nullableInteger(value: unknown, name: string, positive = false): number | null {
+  if (value === null) return null;
+  if (typeof value !== "number" || !Number.isSafeInteger(value) || (positive && value < 1)) {
+    throw new EffectRuntimeRequestError(`${name} must be ${positive ? "a positive" : "a"} safe integer or null`);
+  }
+  return value;
+}
+function nullableText(value: unknown, name: string): string | null {
+  return value === null ? null : requireNonEmptyString(value, name);
+}
+function decode(raw: unknown, index: number, offset: number): Event {
+  const ordinal = index + offset;
+  const r = requireJsonObject(raw, "event replay facts");
+  const kind = requireStringLiteral(r.event_type, KINDS, "event_type");
+  const todo = r.todo_id === null ? null : requireNonEmptyString(r.todo_id, "todo_id");
+  if (["todo_added", "todo_claimed", "todo_updated", "todo_blocked", "todo_deferred", "todo_completed"].includes(kind) && todo === null) throw new EffectRuntimeRequestError(`${kind} requires refs.todo_id`);
+  if (typeof r.content_changed !== "boolean") throw new EffectRuntimeRequestError("content_changed must be boolean");
+  if (typeof r.has_exclusions !== "boolean" || (r.goal_bound !== null && typeof r.goal_bound !== "boolean")) {
+    throw new EffectRuntimeRequestError("event ownership facts require explicit booleans");
+  }
+  return {ordinal, id: requireNonEmptyString(r.event_id, "event_id"), goal: requireNonEmptyString(r.goal_id, "goal_id"),
+    kind, todo, sequence: nullableInteger(r.append_sequence, "append_sequence", true),
+    time: requireNonEmptyString(r.recorded_at, "recorded_at"),
+    role: r.role === null ? null : requireStringLiteral(r.role, ["user", "agent"] as const, "role"),
+    priority: r.priority === null ? null : normalizeTodoPriority(r.priority),
+    plannerOrder: nullableInteger(r.planner_order, "planner_order"), contentChanged: r.content_changed,
+    fields: requireStringArray(r.fields, "event content fields"),
+    binding: nullableText(r.capability_binding_ref, "capability_binding_ref"),
+    continuation: nullableText(r.continuation_policy, "continuation_policy"),
+    removedPolicy: nullableText(r.removed_continuation_policy, "removed_continuation_policy"),
+    exclusions: r.has_exclusions, goalBound: r.goal_bound};
+}
+
+/** Continuation state is ephemeral replay data, never a durable authority token. */
+function decodeTodo(value: unknown): Todo {
+  const r = requireJsonObject(value, "replay continuation");
+  const sources = requireJsonObject(r.field_sources, "content sources");
+  const field_sources: Record<string, number> = {};
+  for (const [key, index] of Object.entries(sources)) {
+    const n = nullableInteger(index, "content source ordinal");
+    if (n === null || n < 0) throw new EffectRuntimeRequestError("invalid content source ordinal");
+    Object.defineProperty(field_sources, key, {value: n, enumerable: true, writable: true, configurable: true});
+  }
+  const role = requireStringLiteral(r.role, ["user", "agent"] as const, "role");
+  const status = requireStringLiteral(r.status, ["open", "done", "blocked", "deferred"] as const, "status");
+  if (typeof r.render_priority !== "boolean") throw new EffectRuntimeRequestError("render_priority must be boolean");
+  const priority = normalizeTodoPriority(r.priority);
+  if (priority === null) throw new EffectRuntimeRequestError("continuation priority is required");
+  return {todo_id: requireNonEmptyString(r.todo_id, "todo_id"), field_sources,
+    status, done: status === "done", role, priority,
+    source_section: role === "user" ? "User Todo / Owner Review Reading Queue" : "Agent Todo",
+    render_priority: r.render_priority, planner_order: nullableInteger(r.planner_order, "planner_order"),
+    append_sequence: nullableInteger(r.append_sequence, "append_sequence", true),
+    binding: nullableText(r.binding, "capability binding"), removedPolicy: nullableText(r.removedPolicy, "removed policy")};
+}
+
+export function planStateEventReplay(value: unknown): JsonObject {
+  const r = requireJsonObject(value, "state event replay request");
+  if (r.schema_version !== "state_event_replay_request_v0" || !Array.isArray(r.events)) {
+    throw new EffectRuntimeRequestError("state event replay requires its schema and complete event facts");
+  }
+  const offset = nullableInteger(r.offset ?? 0, "batch offset");
+  if (offset === null || offset < 0 || offset > Number.MAX_SAFE_INTEGER - r.events.length) {
+    throw new EffectRuntimeRequestError("invalid batch offset");
+  }
+  const events = r.events.map((event, index) => decode(event, index, offset));
+  const ids = new Set<string>();
+  for (const event of events) {
+    if (ids.has(event.id)) throw new EffectRuntimeRequestError("event replay facts must be deduplicated by the source codec");
+    ids.add(event.id);
+  }
+  events.sort((a, b) => (a.sequence ?? 0) - (b.sequence ?? 0) || authorityUnicodeCompare(a.time, b.time) || authorityUnicodeCompare(a.id, b.id));
+  const goal = r.goal_id === null ? events[0]?.goal ?? "" : requireNonEmptyString(r.goal_id, "goal_id");
+  const todos = new Map<string, Todo>();
+  if (r.initial_todos !== undefined && !Array.isArray(r.initial_todos)) {
+    throw new EffectRuntimeRequestError("initial_todos must be an array");
+  }
+  for (const raw of (r.initial_todos ?? []) as unknown[]) {
+    const todo = decodeTodo(raw);
+    if (todos.has(todo.todo_id) || Object.values(todo.field_sources).some(index => index >= offset)) {
+      throw new EffectRuntimeRequestError("continuation does not precede this replay batch");
+    }
+    todos.set(todo.todo_id, todo);
+  }
+  const timeline: number[] = [];
+  for (const event of events) {
+    if (event.goal !== goal) throw new EffectRuntimeRequestError("all events in a projection must share one goal_id");
+    const kind = event.kind;
+    switch (kind) {
+      case "todo_added": {
+        const id = event.todo!; // decode requires identity for every Todo event.
+        if (todos.has(id)) throw new EffectRuntimeRequestError(`todo_id already exists: ${id}; use todo_updated`);
+        const role = event.role ?? "agent";
+        todos.set(id, {todo_id: id, field_sources: Object.fromEntries(event.fields.map(field => [field, event.ordinal])), status: "open", done: false,
+          role, priority: event.priority ?? "P2", planner_order: event.plannerOrder,
+          source_section: role === "user" ? "User Todo / Owner Review Reading Queue" : "Agent Todo",
+          render_priority: event.priority !== null, append_sequence: event.sequence,
+          binding: event.binding, removedPolicy: event.removedPolicy});
+        break;
+      }
+      case "todo_claimed": case "todo_updated": case "todo_blocked": case "todo_deferred": case "todo_completed": {
+        const todo = todos.get(event.todo!);
+        if (!todo) throw new EffectRuntimeRequestError(`${kind} references unknown todo_id: ${event.todo}`);
+        for (const field of event.fields) Object.defineProperty(todo.field_sources, field,
+          {value: event.ordinal, enumerable: true, configurable: true, writable: true});
+        if (!event.fields.includes("last_actor_agent_id")) delete todo.field_sources.last_actor_agent_id;
+        if (kind === "todo_updated") {
+          if (event.binding !== null) {
+            if (todo.binding !== null && todo.binding !== event.binding) {
+              throw new EffectRuntimeRequestError("capability_binding_ref is immutable once set");
+            }
+            todo.binding = event.binding;
+          }
+          if (event.removedPolicy !== null) {
+            delete todo.field_sources.continuation_policy;
+            todo.removedPolicy = event.removedPolicy;
+          } else if (event.continuation !== null && todo.removedPolicy !== null) {
+            if (event.continuation === "independent_handoff" && event.exclusions) {
+              delete todo.field_sources.removed_continuation_policy;
+              todo.removedPolicy = null;
+            } else delete todo.field_sources.continuation_policy;
+          }
+          if (event.fields.includes("bound_agent")) delete todo.field_sources.goal_bound;
+          if (event.goalBound === true) delete todo.field_sources.bound_agent;
+          if (event.role !== null) todo.role = event.role;
+          if (event.priority !== null) todo.priority = event.priority;
+          if (event.priority !== null || event.contentChanged) todo.render_priority = true;
+          todo.source_section = todo.role === "user" ? "User Todo / Owner Review Reading Queue" : "Agent Todo";
+        } else if (kind === "todo_blocked") todo.status = "blocked";
+        else if (kind === "todo_deferred") todo.status = "deferred";
+        else if (kind === "todo_completed") todo.status = "done";
+        todo.done = todo.status === "done";
+        break;
+      }
+      case "refresh_recorded": case "run_recorded": case "quota_spent": case "evidence_attached":
+        timeline.push(event.ordinal); break;
+      case "supervisor_proposed": case "supervisor_receipt_recorded": break;
+      default: assertNever(kind, "unhandled event replay kind");
+    }
+  }
+  const items = [...todos.values()].sort((a, b) => Number(a.role !== "user") - Number(b.role !== "user") ||
+    authorityUnicodeCompare(a.priority, b.priority) || (a.planner_order ?? 9999) - (b.planner_order ?? 9999) ||
+    (a.append_sequence ?? 0) - (b.append_sequence ?? 0));
+  return {schema_version: "state_event_replay_plan_v0", goal_id: goal,
+    event_indices: events.map(e => e.ordinal), timeline_indices: timeline,
+    todos: items.map(item => ({...item, sort_key: [Number(item.role !== "user"), item.priority,
+      item.planner_order ?? 9999, item.append_sequence ?? 0]}))};
+}
diff --git a/loopx/event_sourced_state.py b/loopx/event_sourced_state.py
index e9a84b76bb..e4ef42ba31 100644
--- a/loopx/event_sourced_state.py
+++ b/loopx/event_sourced_state.py
@@ -739,20 +739,11 @@ def event_sort_key(event: dict[str, Any]) -> tuple[int, str, str]:
     )
 
 
-def _todo_from_added_event(event: dict[str, Any]) -> dict[str, Any]:
+def _decode_added_todo_content(event: dict[str, Any]) -> dict[str, Any]:
     payload = event.get("payload") or {}
     refs = event.get("refs") or {}
     text = compact_text(payload.get("text") or payload.get("title"))
-    role = compact_text(payload.get("role") or "agent")
-    source_section = (
-        "User Todo / Owner Review Reading Queue" if role == "user" else "Agent Todo"
-    )
-    todo_id = normalize_todo_id(refs.get("todo_id")) or build_todo_id(
-        role=role,
-        source_section=source_section,
-        index=event.get("append_sequence"),
-        text=text,
-    )
+    todo_id = refs["todo_id"]  # The legacy decoder already requires this identity.
     task_class = normalize_explicit_todo_task_class(payload.get("task_class"))
     action_kind = normalize_todo_action_kind(payload.get("action_kind"))
     task_domain = normalize_todo_task_domain(payload.get("task_domain"))
@@ -781,16 +772,8 @@ def _todo_from_added_event(event: dict[str, Any]) -> dict[str, Any]:
     claimed_by = normalize_todo_claimed_by(payload.get("claimed_by"))
     actor_agent_id = normalize_todo_claimed_by(event.get("actor_agent_id"))
     todo: dict[str, Any] = {
-        "schema_version": "todo_item_v0",
-        "todo_id": todo_id,
-        "role": role,
-        "status": TODO_STATUS_OPEN,
-        "done": False,
-        "priority": compact_text(payload.get("priority") or "P2"),
-        "title": text,
-        "text": text if not payload.get("priority") else f"[{compact_text(payload.get('priority'))}] {text}",
-        "source_section": source_section,
-        "planner_order": payload.get("planner_order"),
+        "schema_version": "todo_item_v0", "todo_id": todo_id,
+        "title": text, "text": text, "planner_order": payload.get("planner_order"),
         "append_sequence": event.get("append_sequence"),
         "last_event_id": event.get("event_id"),
         "updated_at": compact_text(payload.get("updated_at")),
@@ -837,22 +820,20 @@ def _todo_from_added_event(event: dict[str, Any]) -> dict[str, Any]:
     return todo
 
 
-def _update_todo_from_event(todo: dict[str, Any], event: dict[str, Any]) -> None:
+def _decode_todo_event_content(event: dict[str, Any]) -> dict[str, Any]:
+    """Normalize historical payload values without reading prior Todo state."""
+    todo: dict[str, Any] = {}
     payload = event.get("payload") or {}
     event_type = event.get("event_type")
     actor_agent_id = normalize_todo_claimed_by(event.get("actor_agent_id"))
     if actor_agent_id:
         todo["last_actor_agent_id"] = actor_agent_id
-    elif event_type not in (REFRESH_RECORDED, RUN_RECORDED, QUOTA_SPENT, EVIDENCE_ATTACHED):
-        todo.pop("last_actor_agent_id", None)
     if event_type == TODO_CLAIMED:
         claimed_by = normalize_todo_claimed_by(payload.get("claimed_by"))
         if claimed_by:
             todo["claimed_by"] = claimed_by
     elif event_type == TODO_UPDATED:
         for key in (
-            "priority",
-            "role",
             "title",
             "task_class",
             "action_kind",
@@ -866,9 +847,6 @@ def _update_todo_from_event(todo: dict[str, Any], event: dict[str, Any]) -> None
             payload.get("capability_binding_ref")
         )
         if capability_binding_ref:
-            existing_binding_ref = todo.get("capability_binding_ref")
-            if existing_binding_ref and existing_binding_ref != capability_binding_ref:
-                raise StateEventError("capability_binding_ref is immutable once set")
             todo["capability_binding_ref"] = capability_binding_ref
         continuation_policy = normalize_todo_continuation_policy(
             payload.get("continuation_policy")
@@ -881,18 +859,9 @@ def _update_todo_from_event(todo: dict[str, Any], event: dict[str, Any]) -> None
             payload.get("excluded_agents")
         )
         if removed_continuation_policy:
-            todo.pop("continuation_policy", None)
             todo["removed_continuation_policy"] = removed_continuation_policy
         elif continuation_policy:
-            explicit_repair = (
-                todo.get("removed_continuation_policy")
-                and continuation_policy == "independent_handoff"
-                and bool(update_excluded_agents)
-            )
-            if not todo.get("removed_continuation_policy") or explicit_repair:
-                todo["continuation_policy"] = continuation_policy
-            if explicit_repair:
-                todo.pop("removed_continuation_policy", None)
+            todo["continuation_policy"] = continuation_policy
         required_write_scopes = normalize_required_write_scopes(
             payload.get("required_write_scopes")
         )
@@ -914,12 +883,9 @@ def _update_todo_from_event(todo: dict[str, Any], event: dict[str, Any]) -> None
         bound_agent = normalize_todo_bound_agent(payload.get("bound_agent"))
         if bound_agent:
             todo["bound_agent"] = bound_agent
-            todo.pop("goal_bound", None)
         goal_bound = normalize_todo_goal_bound(payload.get("goal_bound"))
         if goal_bound is not None:
             todo["goal_bound"] = goal_bound
-            if goal_bound:
-                todo.pop("bound_agent", None)
         global_gate = normalize_todo_global_gate(payload.get("global_gate"))
         if global_gate is not None:
             todo["global_gate"] = global_gate
@@ -931,23 +897,15 @@ def _update_todo_from_event(todo: dict[str, Any], event: dict[str, Any]) -> None
         if payload.get("text") or payload.get("title"):
             title = compact_text(payload.get("text") or payload.get("title"))
             todo["title"] = title
-            priority = compact_text(todo.get("priority") or "")
-            todo["text"] = f"[{priority}] {title}" if priority else title
     elif event_type == TODO_BLOCKED:
-        todo["status"] = "blocked"
-        todo["done"] = False
         if payload.get("reason"):
             todo["reason"] = compact_text(payload["reason"])
     elif event_type == TODO_DEFERRED:
-        todo["status"] = "deferred"
-        todo["done"] = False
         if payload.get("reason"):
             todo["reason"] = compact_text(payload["reason"])
         if payload.get("resume_when"):
             todo["resume_when"] = compact_text(payload["resume_when"])
     elif event_type == TODO_COMPLETED:
-        todo["status"] = TODO_STATUS_DONE
-        todo["done"] = True
         for key in (
             "evidence",
             "reason",
@@ -966,6 +924,7 @@ def _update_todo_from_event(todo: dict[str, Any], event: dict[str, Any]) -> None
             todo["successor_todo_ids"] = successor_todo_ids
     todo["last_event_id"] = event.get("event_id")
     todo["last_append_sequence"] = event.get("append_sequence")
+    return todo
 
 
 def build_state_projection(
@@ -974,61 +933,100 @@ def build_state_projection(
     goal_id: str | None = None,
     generated_at: str | None = None,
 ) -> dict[str, Any]:
-    normalized = _dedupe_events(normalize_state_event(event) for event in events)
-    ordered = sorted(normalized, key=event_sort_key)
-    inferred_goal_id = goal_id or (ordered[0]["goal_id"] if ordered else "")
-    todos: dict[str, dict[str, Any]] = {}
-    timeline: list[dict[str, Any]] = []
-
-    for event in ordered:
-        if inferred_goal_id and event["goal_id"] != inferred_goal_id:
-            raise StateEventError("all events in a projection must share one goal_id")
-        event_type = event["event_type"]
-        todo_id = (event.get("refs") or {}).get("todo_id")
-        if event_type == TODO_ADDED:
-            todo = _todo_from_added_event(event)
-            todos[todo["todo_id"]] = todo
-        elif event_type in TODO_EVENT_TYPES and todo_id:
-            todo = todos.get(todo_id)
-            if todo is None:
-                raise StateEventError(f"{event_type} references unknown todo_id: {todo_id}")
-            _update_todo_from_event(todo, event)
-        elif event_type in {REFRESH_RECORDED, RUN_RECORDED, QUOTA_SPENT, EVIDENCE_ATTACHED}:
-            timeline_entry: dict[str, Any] = {
-                "event_id": event["event_id"],
-                "event_type": event_type,
-                "append_sequence": event.get("append_sequence"),
-                "recorded_at": event.get("recorded_at"),
-                "summary": compact_text((event.get("payload") or {}).get("summary")),
-                "refs": event.get("refs") or {},
-            }
-            if event.get("actor_agent_id"):
-                timeline_entry["actor_agent_id"] = event["actor_agent_id"]
-            timeline.append(timeline_entry)
-
-    todo_items = sorted(
-        todos.values(),
-        key=lambda item: (
-            item.get("role") != "user",
-            str(item.get("priority") or "P9"),
-            int(item.get("planner_order") or 9999),
-            int(item.get("append_sequence") or 0),
-        ),
-    )
-    user_todos = [item for item in todo_items if item.get("role") == "user"]
-    agent_todos = [item for item in todo_items if item.get("role") != "user"]
+    # Python owns legacy event normalization and exact historical checksum bytes.
+    # TS receives bounded semantic facts, never evidence, validation commands or
+    # arbitrary payloads; returned ordinals address this exact normalized batch.
+    from .control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
+
+    normalized = sorted(_dedupe_events(normalize_state_event(event) for event in events), key=event_sort_key)
+    facts = []
+    contents = []
+    for event in normalized:
+        payload = event["payload"]
+        kind = event["event_type"]
+        edits = kind in (TODO_ADDED, TODO_UPDATED)
+        order = payload.get("planner_order") if kind == TODO_ADDED else None
+        if order is not None:
+            if isinstance(order, bool):
+                raise StateEventError("planner_order must be an integer")
+            try:
+                order = int(order)
+            except (ValueError, TypeError, OverflowError) as exc:
+                raise StateEventError("planner_order must be an integer") from exc
+        sequence = event.get("append_sequence")
+        for value in (order, sequence):
+            if value is not None and abs(value) > 2**53 - 1:
+                raise StateEventError("event replay integers must be safe integers")
+        content = (_decode_added_todo_content(event) if kind == TODO_ADDED else
+                   _decode_todo_event_content(event) if kind in TODO_EVENT_TYPES else {})
+        contents.append(content)
+        facts.append({
+            "event_id": event["event_id"], "goal_id": event["goal_id"],
+            "event_type": kind, "append_sequence": sequence,
+            "recorded_at": event["recorded_at"],
+            "todo_id": event["refs"].get("todo_id") if kind in TODO_EVENT_TYPES else None,
+            "role": compact_text(payload.get("role")) or None if edits else None,
+            "priority": compact_text(payload.get("priority")) or None if edits else None,
+            "planner_order": order, "fields": list(content),
+            "capability_binding_ref": content.get("capability_binding_ref"),
+            "continuation_policy": content.get("continuation_policy"),
+            "removed_continuation_policy": content.get("removed_continuation_policy"),
+            "has_exclusions": bool(content.get("excluded_agents")),
+            "goal_bound": content.get("goal_bound"),
+            "content_changed": bool(payload.get("text") or payload.get("title")) if edits else False,
+        })
+    # Only touched Todo continuation rows cross each bounded call. The host
+    # retains content and prior results; TS alone decides all state transitions.
+    rows: dict[str, dict[str, Any]] = {}
+    timeline_indices: list[int] = []
+    inferred_goal = goal_id or (normalized[0]["goal_id"] if normalized else "")
+    for offset in range(0, len(facts), 256):
+        batch = facts[offset:offset + 256]
+        touched = {fact["todo_id"] for fact in batch if fact["todo_id"] is not None}
+        try:
+            plan = effect_runtime_result("goal.state_event.plan_replay", {
+                "schema_version": "state_event_replay_request_v0", "events": batch,
+                "goal_id": inferred_goal, "offset": offset,
+                "initial_todos": [rows[key] for key in touched if key in rows],
+            })
+        except EffectRuntimeRejected as exc:
+            raise StateEventError(str(exc)) from exc
+        if not isinstance(plan, dict) or plan.get("schema_version") != "state_event_replay_plan_v0":
+            raise RuntimeError("invalid typed state event replay plan")
+        for row in plan["todos"]:
+            rows[row["todo_id"]] = row
+        timeline_indices.extend(plan["timeline_indices"])
+    ordered = normalized  # Exact historical checksum order remains in the codec.
+    todo_items = []
+    for row in sorted(rows.values(), key=lambda row: row["sort_key"]):
+        todo = {key: contents[index][key] for key, index in row["field_sources"].items()}
+        for key in ("role", "priority", "status", "done", "source_section"):
+            todo[key] = row[key]
+        if row["render_priority"]:
+            todo["text"] = f"[{row['priority']}] {todo['title']}"
+        todo_items.append(todo)
+    timeline = []
+    for index in timeline_indices:
+        event = normalized[index]
+        entry = {key: event.get(key) for key in (
+            "event_id", "event_type", "append_sequence", "recorded_at", "refs",
+        )}
+        entry["summary"] = compact_text(event["payload"].get("summary"))
+        if event.get("actor_agent_id"):
+            entry["actor_agent_id"] = event["actor_agent_id"]
+        timeline.append(entry)
     last_event = ordered[-1] if ordered else {}
     return {
         "schema_version": STATE_PROJECTION_SCHEMA_VERSION,
-        "goal_id": inferred_goal_id,
+        "goal_id": inferred_goal,
         "generated_at": generated_at or now_utc_iso(),
         "source_event_count": len(ordered),
         "source_checksum": event_stream_checksum(ordered),
         "last_event_id": last_event.get("event_id"),
         "last_append_sequence": last_event.get("append_sequence"),
         "projection_version": STATE_PROJECTION_VERSION,
-        "user_todos": _todo_summary(user_todos, role="user"),
-        "agent_todos": _todo_summary(agent_todos, role="agent"),
+        "user_todos": _todo_summary([item for item in todo_items if item["role"] == "user"], role="user"),
+        "agent_todos": _todo_summary([item for item in todo_items if item["role"] != "user"], role="agent"),
         "timeline": timeline,
     }
 

From b5b2d7e02ddfef7109094e88d9df0306611a8cbb Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:14:24 +0800
Subject: [PATCH 2/3] test(events): qualify replay integrity and reconcile
 cutover boundaries

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 .../2026-09-25-event-replay.md                |  98 ++++++++++++++
 .../2026-09-25-event-replay.zh-CN.md          |  71 ++++++++++
 ...shared-goal-authority-state-provider-v0.md |   2 +
 ...-goal-authority-state-provider-v0.zh-CN.md |   2 +
 .../typescript-control-plane-migration-v0.md  |   2 +
 ...script-control-plane-migration-v0.zh-CN.md |   2 +
 .../test_event_replay_integrity.py            | 121 ++++++++++++++++++
 .../state_event_replay.test.ts                | 110 ++++++++++++++++
 8 files changed, 408 insertions(+)
 create mode 100644 docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.md
 create mode 100644 docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.zh-CN.md
 create mode 100644 tests/control_plane/test_event_replay_integrity.py
 create mode 100644 tests/control_plane_ts/state_event_replay.test.ts

diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.md
new file mode 100644
index 0000000000..b27da4d255
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.md
@@ -0,0 +1,98 @@
+# Event replay integrity and the remaining cutover work
+
+Owner: overall roadmap #4574 R1/R5, shared authority L7/L8 and TS migration T3.
+Baseline: `2e1e63260`, 2026-09-25. PR states below are an inventory at this
+checkpoint, not a promise of acceptance or a continuously updated counter.
+
+## Delivered boundary
+
+The legacy event source had independent Python state rules. A second
+`todo_added` with a different event identity could overwrite a completed Todo;
+a priority-only update left old text, a role update left its old source section,
+and planner order zero sorted as missing. Four independent counterexamples fail
+on the baseline and pass with the typed replay owner.
+
+`goals/state_event_replay.ts` owns ordered replay admission, Todo identity,
+lifecycle, role/priority, binding immutability, exclusive addressing and removed
+continuation-policy repair. Python retains legacy value decoding, content,
+checksums and rendering. The runtime receives compact facts and content-field
+names; the returned field-source ordinals address that same immutable batch.
+Evidence, validation commands and arbitrary payload bodies do not cross this RPC.
+One large evidence regression preserves more than 2 MiB of content with under
+4 KiB of replay facts. The adapter folds at most 256 events per call, carrying
+only the affected Todo continuation rows. A 4,100-event history crosses 17
+calls without losing original fields or duplicate-creation protection. Individual
+facts and continuation rows remain subject to the existing RPC budget; no
+unlimited per-field size is promised. Historic byte/checksum ordering remains
+in the legacy codec; TS returns the final display sort keys.
+
+Exact duplicate event identities retain existing codec replay/conflict behavior.
+Different create events targeting one Todo now reject with an actionable update
+instruction. Invalid role/priority and unsafe sequence/order facts reject instead
+of producing an ambiguous typed projection. Existing stored logs are not
+rewritten. Reverting restores the earlier reader; no storage migration is needed.
+
+A detached real-source rehearsal preserves the complete baseline projection and
+checksum, with 900 backfilled events and 398 Todos. A disposable registry and real
+CLI read back an original record. The original source stays unchanged. Synthetic
+mixed history covers dependencies, validation declarations, claims, deferred and
+completed work, independent review, owner work, attribution and all event kinds.
+Seven alternating warm samples on the same detached input measured median
+9.28 ms before and 66.12 ms after. This is an explicit RPC/type-owner cost, not
+a speedup. No latency budget was increased; wider cold/throughput qualification
+is not claimed. Existing compaction, Markdown-backfill, task-graph and API smokes retain their
+original assertions. File/SQLite caller regression remains real, not in-memory.
+
+This does **not** append or bind event writes to the outbox. The
+`event_log_writer_not_bound` hold remains. #5003 owns atomic append/completion;
+this change does not reproduce its writer or change its persistent event schema.
+No provider default, live promotion, PostgreSQL store or external executor changes.
+The existing source-readback type narrowing also matches the small correction
+already carried by #5012/#5013; it adds no new projection rule.
+
+## Count requirements, open implementations and evidence separately
+
+The older 5–8 / 7–9 estimates mixed units and must not be reused. The reconciled
+inventory in #5006 already identifies source transport as implemented on an open
+branch. It is not unstarted work. The remaining *planned new code deliveries*
+are four including this event-replay slice:
+
+| Planned PR boundary | Exit | After this delivery |
+| --- | --- | --- |
+| Event replay integrity (this PR) | One typed replay owner; corruption counterexamples and real-source parity/readback | Ready for review, not merged |
+| External-effect executor fence | Audit real executor consumers; prove how current execution ownership protects the actual effect interval and uncertain result recovery | Unstarted; downstream idempotency/fencing must be explicit, a pre-call check alone cannot promise this |
+| Event-writer binding + integrated whole-Goal migration/rollback | Reuse #5003/#5006, bind actual writer locks/publication to outbox, mixed-writer crash/replay/drain, canonical consumers and fenced rollback on one exact revision/profile | Unstarted; this replay slice closes a proven reader gap within that boundary, not capture admission |
+| Default/onboarding + bounded Python retirement | Qualified profile selected by new-Goal creation/settings/install and packaged entrypoints; explicit existing-Goal migration and rollback; delete only writers with no remaining legal callers | Depends on preceding acceptance and applicable D1–D3 |
+
+Thus **three planned new implementation boundaries remain after this PR**;
+that is not an assertion that three more merges enable a global default.
+Combining writer binding and migration is a plan to verify; if their integration
+reveals a defect, record the concrete defect and revised boundary here rather
+than keep a floating range. The replay slice is separated because its source
+reader and state rules are independently testable/reversible while #5003's
+writer is still under review. It also fixes shipped behavior immediately.
+
+Existing open work to integrate, not implement again:
+
+- #5006 complete source transport; #5003 atomic event completion; #5011 observer
+  retirement: three source/authority integrations.
+- #4991, #4992, #4995: three quota/lease caller repairs. #4994 merged during
+  this delivery and is now included in the rebased baseline, not the open count.
+- #5005, #5012, #5013: three demonstrated long-horizon recovery fixes. These are
+  relevant R1 reliability work, not three additional storage implementations.
+- #4931 and contributor-owned #4224: SQLite D2 performance/qualification.
+  #4915 changes local filesystem placement, not authority selection; #5010/#5008
+  are release/platform integration work, not unstarted provider implementations.
+
+D2 has a measured 1 MiB receipt/scan failure and outstanding recovery, lag,
+restore/upgrade, runtime/OS and elapsed-soak evidence. A stated soak end date is
+not a verified final result. Its further PR count cannot be inferred from this
+inventory. D1 command/consumer coverage and D3 exact-profile/cohort acceptance
+also remain evidence gates. Do not add them to code PR counts or subtract them
+because an unrelated refactor merged. Maintainer approval is needed for actual
+cohort cutover; this task does not modify an active Goal.
+
+PostgreSQL already implements the provider contract. Deployed authentication,
+tenancy, operations/restore/failover and capacity qualification remain separate
+medium-term outcomes. Local default does not wait for that deployment; provider
+conformance alone is not a production service qualification.
diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.zh-CN.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.zh-CN.md
new file mode 100644
index 0000000000..e66ddff3aa
--- /dev/null
+++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.zh-CN.md
@@ -0,0 +1,71 @@
+# 事件重放完整性与剩余切换工作
+
+归属：总路线 #4574 R1/R5、shared authority L7/L8、TS 迁移 T3。
+基线：`2e1e63260`，2026-09-25。以下是此时的 PR 清单，不是持续更新的计数器。
+
+## 本次交付
+
+旧事件来源仍由 Python 单独解释状态：不同 event_id 的第二次 `todo_added`
+会覆盖已完成任务；只改 priority 留下旧文本；改 role 留下旧来源分区；
+planner_order=0 被当作缺省值。四个独立反例在基线失败，在新实现通过。
+
+`goals/state_event_replay.ts` 统一负责有序重放准入、Todo 身份、生命周期、角色与优先级、
+能力绑定不可变、用户寻址互斥，以及废弃续接策略的显式修复。Python 保留历史值
+解码、正文、校验和与渲染。RPC 只传紧凑事实和内容字段名，TS 返回字段对应的
+原批次下标；证据、验证命令和任意正文不传入 RPC。超过 2 MiB 的证据仍完整
+保留，事实请求小于 4 KiB。每次最多折叠 256 条事件，仅传涉及的 Todo 续接状态；
+4,100 条历史经 17 次调用仍保留原字段及重复创建保护。单个事实/续接行仍受 RPC
+预算约束，不声称单字段无限大。历史字节/校验和顺序保留在旧编解码器，最终
+展示排序键由 TS 给出。
+
+同一 event_id 的重放和冲突判断仍由原编解码合同处理。不同创建事件不能覆盖
+同一个 Todo，改用明确的更新；非法角色、优先级和不安全整数拒绝进入类型化
+投影。旧日志不重写，撤回代码无需数据迁移。
+
+真实来源的隔离演练覆盖 900 条回填事件、398 个 Todo，新旧完整投影及校验和
+一致，隔离注册表下真实 CLI 读回原记录，原来源未改变。混合合成历史覆盖依赖、
+验证声明、领取、延期、完成、独立评审、用户工作、归因和各事件类型。原有 API、
+压缩重放、Markdown 回填和任务图 smoke 保留原断言；File/SQLite 调用者回归
+使用真实实现。同一隔离输入交替运行七组 warm 样本，中位数由 9.28 ms 变为
+66.12 ms。这是跨 runtime 和类型化规则的明确成本，不是性能优化；没有增加
+延迟预算，也不声称完成更广的冷启动/吞吐资格。
+
+本次不写事件 outbox，`event_log_writer_not_bound` 保持。#5003 负责原子追加和
+完成，不重复其 writer，不改事件持久格式、默认 provider、活动 Goal 或外部
+执行器。来源读回处的小型类型收窄与 #5012/#5013 已有修正相同，不增加业务规则。
+
+## 剩余工作如何计数
+
+旧 5–8 / 7–9 混算了代码、在审实现和验收证据，不能继续引用。#5006 已经实现了
+完整来源传输，当前状态是在审，不能重新算成尚未开发。当前规划的新代码交付
+是四个边界，含本次：
+
+| 规划 PR | 可观察验收 | 本次之后 |
+| --- | --- | --- |
+| 本次事件重放完整性 | 单一 TS 重放 owner、反例、真实来源一致性及读回 | 待评审，未合入 |
+| 外部 effect 的执行保护 | 审计实际执行入口，证明执行区间及不确定结果恢复；下游幂等和 fencing 合同明确 | 未开始；调用前检查不能保证整个区间 |
+| 事件 writer 绑定与整 Goal 迁移/回滚 | 集成 #5003/#5006，绑定真实写锁与 outbox，混合写入、崩溃恢复、drain、canonical 读回及有 fence 的回滚 | 未开始；本次关闭其中已证实的 reader 缺陷，不开放捕获准入 |
+| 默认启用与有限 Python 退役 | 新 Goal、设置、安装、打包入口一致选择合格 profile；现有 Goal 显式迁移；删除最后调用者已迁走的 writer | 依赖前面验收及适用 D1–D3 |
+
+所以本次之后是**三个规划中的新实现边界**，不等于再合三个 PR 就能全量切换。
+writer 绑定与整体验收是否仍能同批交付，要用集成证据确认；若发现新问题，记录
+具体缺陷及拆分原因，不能继续保留一个浮动范围。此次 reader 单独交付，是因为
+它可独立验证/回滚并立即修复现有行为，而 #5003 writer 仍在评审。
+
+已有在审工作另列，不重复开发：
+
+- #5006 来源传输、#5003 原子事件完成、#5011 观察路径退役，共三个来源集成。
+- #4991、#4992、#4995，共三个 quota/lease 调用者修复。#4994 已在本次
+  开发期间合入，已包含在更新后的基线，不再计入在审数量。
+- #5005、#5012、#5013，共三个真实长程恢复缺陷；它们是 R1 可靠性工作，不是
+  三套新的存储实现。
+- #4931 与贡献者负责的 #4224 是 SQLite D2 性能/资格。#4915 是目录迁移，
+  不等于 authority 切换；#5010/#5008 是发布及平台集成，不是未实现 provider。
+
+D2 已测出 1 MiB 回执/扫描预算失败，恢复、lag、备份/升级、OS/runtime 和自然
+时间 soak 仍需各自证据。到达预计 soak 截止日不等于验收成功，无法据此精确
+分配后续 PR 数量。D1 调用者覆盖、D3 精确 profile/迁移 cohort 验收也不能混入
+代码数量；不因无关 PR 合入而扣减。活动 Goal 的迁移仍需维护者批准。
+
+PostgreSQL 已有 provider 实现，中期另需部署鉴权、租户边界、恢复/故障转移、
+运维及容量资格。本地默认不等待其部署，conformance 通过也不等于生产服务就绪。
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
index f333cc77e7..27a9c7230c 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.md
@@ -3392,3 +3392,5 @@ beside each entry, and that this appendix exists for the directory it names.
 2026-09-24: [Typed complete-source assembly and remaining delivery packages](ledger/shared-goal-authority-state-provider-v0/2026-09-24-source-capture.md) unify source construction, identity rejection and current-graph membership; L7/D2/D3 and provider defaults remain open.
 
 2026-09-24: [Leased continuation and remaining local-default packages](ledger/shared-goal-authority-state-provider-v0/2026-09-24-leased-continuation.md).
+
+Event replay and the reconciled cutover inventory: [2026-09-25](ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.md).
diff --git a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
index e83eacb8bc..104016d419 100644
--- a/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
+++ b/docs/architecture/rfcs/shared-goal-authority-state-provider-v0.zh-CN.md
@@ -2613,3 +2613,5 @@ onboarding 按当前核对表分别归为代码、在途 PR 和验收活动，
 2026-09-24：[完整源捕获的 TS 组装与剩余交付包](ledger/shared-goal-authority-state-provider-v0/2026-09-24-source-capture.zh-CN.md)统一源构造、身份拒绝和当前图成员规则；不关闭 L7/D2/D3 或启用默认 provider。
 
 2026-09-24: [带租约接力与剩余本地默认交付包](ledger/shared-goal-authority-state-provider-v0/2026-09-24-leased-continuation.zh-CN.md).
+
+事件重放与剩余切换清单见 [2026-09-25](ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.zh-CN.md).
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index d6c3104aee..4fd6300b3b 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -1931,3 +1931,5 @@ This advances T3/L5; it does not replace D2/D3 or flip a provider default.
 2026-09-24: [Typed complete-source assembly and remaining delivery packages](ledger/shared-goal-authority-state-provider-v0/2026-09-24-source-capture.md) unify source construction, identity rejection and current-graph membership; L7/D2/D3 and provider defaults remain open.
 
 2026-09-24: [Leased continuation and remaining local-default packages](ledger/shared-goal-authority-state-provider-v0/2026-09-24-leased-continuation.md).
+
+Event replay and the reconciled cutover inventory: [2026-09-25](ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.md).
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index 53cae59d1f..26caadf357 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -1490,3 +1490,5 @@ TS 摘要批次；Python 保留旧格式解码、公开字段筛选及渲染。
 2026-09-24：[完整源捕获的 TS 组装与剩余交付包](ledger/shared-goal-authority-state-provider-v0/2026-09-24-source-capture.zh-CN.md)统一源构造、身份拒绝和当前图成员规则；不关闭 L7/D2/D3 或启用默认 provider。
 
 2026-09-24: [带租约接力与剩余本地默认交付包](ledger/shared-goal-authority-state-provider-v0/2026-09-24-leased-continuation.zh-CN.md).
+
+事件重放与剩余切换清单见 [2026-09-25](ledger/shared-goal-authority-state-provider-v0/2026-09-25-event-replay.zh-CN.md).
diff --git a/tests/control_plane/test_event_replay_integrity.py b/tests/control_plane/test_event_replay_integrity.py
new file mode 100644
index 0000000000..e3b1b92bb3
--- /dev/null
+++ b/tests/control_plane/test_event_replay_integrity.py
@@ -0,0 +1,121 @@
+"""Event replay must preserve Todo identity and coherent scheduling/display."""
+import pytest
+
+from loopx.event_sourced_state import StateEventError, build_state_projection, make_state_event
+
+
+def event(kind, number, *, todo_id="todo_alpha", **payload):
+    return {**make_state_event(event_id=f"event-{number}", goal_id="example-goal",
+        event_type=kind, refs={"todo_id": todo_id}, payload=payload,
+        recorded_at="2026-09-24T00:00:00Z"), "append_sequence": number}
+
+
+def test_second_creation_cannot_erase_a_completed_commitment():
+    events = [event("todo_added", 1, title="Retain commitment", role="agent"),
+              event("todo_completed", 2, evidence="Verified"),
+              event("todo_added", 3, title="Replacement", role="agent")]
+    with pytest.raises(StateEventError, match="already exists"):
+        build_state_projection(events)
+
+
+def test_priority_only_update_updates_rendered_text():
+    result = build_state_projection([event("todo_added", 1, title="Keep title", priority="P2"),
+                                     event("todo_updated", 2, priority="P0")])
+    todo = result["agent_todos"]["items"][0]
+    assert (todo["priority"], todo["title"], todo["text"]) == ("P0", "Keep title", "[P0] Keep title")
+
+
+def test_role_update_moves_both_summary_and_source_section():
+    result = build_state_projection([event("todo_added", 1, title="Owner decision", role="agent"),
+                                     event("todo_updated", 2, role="user")])
+    assert result["agent_todos"]["total_count"] == 0
+    assert result["user_todos"]["items"][0]["source_section"] == "User Todo / Owner Review Reading Queue"
+
+
+def test_zero_planner_order_is_a_real_order_not_a_missing_value():
+    result = build_state_projection([event("todo_added", 1, todo_id="todo_first", title="First", planner_order=0),
+                                     event("todo_added", 2, todo_id="todo_second", title="Second", planner_order=1)])
+    assert [todo["todo_id"] for todo in result["agent_todos"]["items"]] == ["todo_first", "todo_second"]
+
+
+def test_large_content_remains_complete_outside_the_typed_facts(monkeypatch):
+    import json
+    from loopx.control_plane import effect_runtime
+
+    invoke = effect_runtime.effect_runtime_result
+    request_sizes = []
+
+    def measured(method, params, **kwargs):
+        if method == "goal.state_event.plan_replay":
+            request_sizes.append(len(json.dumps(params).encode()))
+            assert "Retain this complete evidence" not in json.dumps(params)
+        return invoke(method, params, **kwargs)
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", measured)
+    evidence = "Retain this complete evidence. " * 100_000
+    source = [event("todo_added", 1, title="Large result"),
+              event("todo_completed", 2, evidence=evidence)]
+    result = build_state_projection(source)
+    assert result["agent_todos"]["items"][0]["evidence"] == evidence.strip()
+    assert len(request_sizes) == 1 and request_sizes[0] < 4096
+
+
+def test_identical_event_replay_keeps_checksum_and_no_duplicate_todo():
+    from loopx.event_sourced_state import StateEventConflictError, event_stream_checksum
+
+    source = event("todo_added", 1, title="Original", role="agent")
+    result = build_state_projection([source, source])
+    assert result["source_event_count"] == result["agent_todos"]["total_count"] == 1
+    assert result["source_checksum"] == event_stream_checksum([source])
+    with pytest.raises(StateEventConflictError):
+        build_state_projection([source, {**source, "payload": {"title": "Changed"}}])
+
+
+def test_mixed_history_preserves_content_attribution_and_dependency_edges():
+    source = [
+        event("todo_added", 1, title="Source", priority="P1", claimed_by="author",
+              task_class="advancement_task", capability_binding_ref="review:source",
+              validation_command_argv=["python", "-c", "print('ok')"]),
+        event("todo_deferred", 2, reason="Wait for review", resume_when="todo_done:todo_review"),
+        event("todo_added", 3, todo_id="todo_review", title="Review", role="agent",
+              task_class="advancement_task", claimed_by="reviewer"),
+        event("todo_completed", 4, todo_id="todo_review", evidence="Review passed"),
+        event("todo_updated", 5, title="Source revised", priority="P0"),
+        event("todo_completed", 6, evidence="Delivered", successor_todo_ids=["todo_followup"]),
+        event("todo_added", 7, todo_id="todo_followup", title="Follow-up", role="user",
+              goal_bound=True, task_class="user_action"),
+    ]
+    result = build_state_projection(source)
+    parent = next(row for row in result["agent_todos"]["items"] if row["todo_id"] == "todo_alpha")
+    assert parent["text"] == "[P0] Source revised"
+    assert parent["claimed_by"] == "author"
+    assert parent["capability_binding_ref"] == "review:source"
+    assert parent["successor_todo_ids"] == ["todo_followup"]
+    assert parent["validation_command_argv"] == ["python", "-c", "print('ok')"]
+    assert result["agent_todos"]["done_count"] == 2
+    assert result["user_todos"]["open_count"] == 1
+
+
+def test_long_history_folds_across_bounded_calls_without_losing_old_fields(monkeypatch):
+    import json
+    from loopx.control_plane import effect_runtime
+    invoke = effect_runtime.effect_runtime_result
+    sizes = []
+    def measured(method, params, **kwargs):
+        if method == "goal.state_event.plan_replay":
+            sizes.append(len(json.dumps(params).encode()))
+            assert len(params["events"]) <= 256
+        return invoke(method, params, **kwargs)
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", measured)
+    source = [event("todo_added", 1, title="Original", claimed_by="author", capability_binding_ref="test:bound")]
+    source.extend(event("todo_updated", i, title=f"Revision {i}") for i in range(2, 4100))
+    source.append(event("todo_completed", 4100, evidence="Long history verified"))
+    result = build_state_projection(source)
+    todo = result["agent_todos"]["items"][0]
+    assert todo["title"] == "Revision 4099"
+    assert todo["status"] == "done" and todo["claimed_by"] == "author"
+    assert todo["capability_binding_ref"] == "test:bound"
+    assert len(sizes) == 17 and max(sizes) < 256_000
+    # Duplicate identity protection must survive a batch boundary as well.
+    with pytest.raises(StateEventError, match="already exists"):
+        build_state_projection([*source, event("todo_added", 4101, title="Cannot reset history")])
diff --git a/tests/control_plane_ts/state_event_replay.test.ts b/tests/control_plane_ts/state_event_replay.test.ts
new file mode 100644
index 0000000000..5dc6a3f6ce
--- /dev/null
+++ b/tests/control_plane_ts/state_event_replay.test.ts
@@ -0,0 +1,110 @@
+import assert from "node:assert/strict";
+import {test} from "node:test";
+import {planStateEventReplay} from "../../loopx/control_plane/goals/state_event_replay.ts";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+
+function event(kind: string, n: number, fields: JsonObject = {}): JsonObject {
+  return {event_id: `event-${n}`, goal_id: "sample", event_type: kind, append_sequence: n,
+    recorded_at: "2026-09-24T00:00:00Z", todo_id: "todo_alpha", role: null, priority: null,
+    planner_order: null, content_changed: false, fields: ["title"], capability_binding_ref: null,
+    continuation_policy: null, removed_continuation_policy: null, has_exclusions: false,
+    goal_bound: null, ...fields};
+}
+function plan(events: JsonObject[], goal: string | null = null): JsonObject {
+  return planStateEventReplay({schema_version: "state_event_replay_request_v0", goal_id: goal, events});
+}
+function rows(result: JsonObject): JsonObject[] {return result.todos as JsonObject[];}
+
+test("replay orders facts without changing caller arrays; ordinals still address original content", () => {
+  const input = [event("todo_completed", 3), event("todo_added", 1), event("todo_claimed", 2)];
+  const before = structuredClone(input);
+  const result = plan(input);
+  assert.deepEqual(input, before);
+  assert.deepEqual(result.event_indices, [1, 2, 0]);
+  assert.deepEqual(rows(result)[0].field_sources, {title: 0});
+  assert.equal(rows(result)[0].status, "done");
+  assert.equal(rows(result)[0].done, true);
+});
+
+test("different event IDs cannot overwrite an existing Todo even after completion", () => {
+  assert.throws(() => plan([event("todo_added", 1), event("todo_completed", 2), event("todo_added", 3)]), /already exists/);
+});
+
+test("mixed Goal, orphan, duplicate identity and unsupported event kind reject", () => {
+  assert.throws(() => plan([event("todo_added", 1)], "another"), /share one goal/);
+  assert.throws(() => plan([event("todo_completed", 1)]), /unknown todo_id/);
+  assert.throws(() => plan([event("todo_added", 1), event("todo_added", 1)]), /deduplicated/);
+  assert.throws(() => plan([event("todo_reopened", 1)]), /unsupported/);
+  assert.throws(() => plan([event("todo_added", 1, {todo_id: null})]), /requires refs.todo_id/);
+});
+
+test("summary order preserves zero, user lanes and the actual source section", () => {
+  const result = plan([event("todo_added", 1, {planner_order: 0}),
+    event("todo_added", 2, {todo_id: "todo_beta", planner_order: 1}),
+    event("todo_updated", 3, {todo_id: "todo_beta", role: "user", priority: "P0"})]);
+  const [user, agent] = rows(result);
+  assert.equal(user.todo_id, "todo_beta");
+  assert.equal(user.source_section, "User Todo / Owner Review Reading Queue");
+  assert.equal(user.render_priority, true);
+  assert.equal(agent.planner_order, 0);
+  assert.equal(agent.render_priority, false);
+});
+
+test("legacy lexical tie ordering follows Unicode scalar order, not UTF-16 order", () => {
+  const input = [event("refresh_recorded", 1, {append_sequence: null, event_id: "\u{10000}"}),
+    event("run_recorded", 2, {append_sequence: null, event_id: "\ue000"})];
+  assert.deepEqual(plan(input).event_indices, [1, 0]);
+});
+
+test("malformed facts and unsafe integers cannot silently change replay ordering", () => {
+  for (const field of ["append_sequence", "planner_order"]) {
+    for (const value of [true, 1.5, Number.MAX_SAFE_INTEGER + 1]) {
+      assert.throws(() => plan([event("todo_added", 1, {[field]: value})]), /safe integer/);
+    }
+  }
+  assert.throws(() => plan([event("todo_added", 1, {role: "superuser"})]), /unsupported/);
+  assert.throws(() => plan([event("todo_updated", 1, {priority: "P99"})]), /priority/);
+});
+
+test("binding cannot change; owner addressing is exclusive and actor attribution can clear", () => {
+  const added = event("todo_added", 1, {capability_binding_ref: "domain:alpha",
+    fields: ["capability_binding_ref", "goal_bound", "last_actor_agent_id"]});
+  assert.throws(() => plan([added, event("todo_updated", 2, {capability_binding_ref: "domain:beta"})]), /immutable/);
+  const result = rows(plan([added, event("todo_updated", 2, {fields: ["bound_agent"]})]))[0];
+  assert.deepEqual(result.field_sources, {capability_binding_ref: 0, bound_agent: 1});
+  const rebound = rows(plan([added, event("todo_updated", 2, {fields: ["bound_agent"]}),
+    event("todo_updated", 3, {fields: ["goal_bound"], goal_bound: true})]))[0];
+  assert.deepEqual(rebound.field_sources, {capability_binding_ref: 0, goal_bound: 2});
+});
+
+test("removed continuation remains blocked until an explicit independent handoff repair", () => {
+  const added = event("todo_added", 1, {removed_continuation_policy: "author_handoff",
+    fields: ["removed_continuation_policy"]});
+  const denied = event("todo_updated", 2, {continuation_policy: "independent_handoff", fields: ["continuation_policy"]});
+  assert.deepEqual(rows(plan([added, denied]))[0].field_sources, {removed_continuation_policy: 0});
+  const repaired = {...denied, has_exclusions: true, fields: ["continuation_policy", "excluded_agents"]};
+  assert.deepEqual(rows(plan([added, repaired]))[0].field_sources, {continuation_policy: 1, excluded_agents: 1});
+});
+
+test("all historical event kinds have an explicit projection disposition", () => {
+  const input = [event("todo_added", 1), event("todo_claimed", 2), event("todo_updated", 3),
+    event("todo_blocked", 4), event("todo_deferred", 5), event("todo_completed", 6),
+    event("refresh_recorded", 7), event("run_recorded", 8), event("quota_spent", 9),
+    event("evidence_attached", 10), event("supervisor_proposed", 11), event("supervisor_receipt_recorded", 12)];
+  const result = plan(input);
+  assert.deepEqual(result.timeline_indices, [6, 7, 8, 9]);
+  assert.equal(rows(result)[0].status, "done");
+  assert.equal((result.event_indices as number[]).length, 12);
+});
+
+test("continuation is equivalent to one fold and cannot point into the new batch", () => {
+  const first = [event("todo_added", 1, {fields: ["title", "claimed_by"]})];
+  const next = [event("todo_updated", 2, {priority: "P0", fields: ["title"]}), event("todo_completed", 3, {fields: ["evidence"]})];
+  const seed = rows(plan(first));
+  const resumed = planStateEventReplay({schema_version: "state_event_replay_request_v0", goal_id: "sample",
+    offset: 1, initial_todos: seed, events: next});
+  assert.deepEqual(rows(resumed), rows(plan([...first, ...next])));
+  assert.throws(() => planStateEventReplay({schema_version: "state_event_replay_request_v0", goal_id: "sample",
+    offset: 0, initial_todos: seed, events: next}), /does not precede/);
+  assert.deepEqual(seed, rows(plan(first)), "continuation input must remain unchanged");
+});

From ae84cfc8b33d72b457ab69a6eb1ba18b2ded1e6c Mon Sep 17 00:00:00 2001
From: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
Date: Fri, 25 Sep 2026 00:20:24 +0800
Subject: [PATCH 3/3] fix(events): preserve explicit addressing flags during
 replay

Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com>
---
 loopx/control_plane/goals/state_event_replay.ts   | 2 +-
 tests/control_plane_ts/state_event_replay.test.ts | 6 ++++++
 2 files changed, 7 insertions(+), 1 deletion(-)

diff --git a/loopx/control_plane/goals/state_event_replay.ts b/loopx/control_plane/goals/state_event_replay.ts
index 3e66c749f8..d2f61adeb9 100644
--- a/loopx/control_plane/goals/state_event_replay.ts
+++ b/loopx/control_plane/goals/state_event_replay.ts
@@ -147,7 +147,7 @@ export function planStateEventReplay(value: unknown): JsonObject {
               todo.removedPolicy = null;
             } else delete todo.field_sources.continuation_policy;
           }
-          if (event.fields.includes("bound_agent")) delete todo.field_sources.goal_bound;
+          if (event.fields.includes("bound_agent") && event.goalBound === null) delete todo.field_sources.goal_bound;
           if (event.goalBound === true) delete todo.field_sources.bound_agent;
           if (event.role !== null) todo.role = event.role;
           if (event.priority !== null) todo.priority = event.priority;
diff --git a/tests/control_plane_ts/state_event_replay.test.ts b/tests/control_plane_ts/state_event_replay.test.ts
index 5dc6a3f6ce..d9f1649af0 100644
--- a/tests/control_plane_ts/state_event_replay.test.ts
+++ b/tests/control_plane_ts/state_event_replay.test.ts
@@ -75,6 +75,12 @@ test("binding cannot change; owner addressing is exclusive and actor attribution
   const rebound = rows(plan([added, event("todo_updated", 2, {fields: ["bound_agent"]}),
     event("todo_updated", 3, {fields: ["goal_bound"], goal_bound: true})]))[0];
   assert.deepEqual(rebound.field_sources, {capability_binding_ref: 0, goal_bound: 2});
+  for (const flag of [false, true]) {
+    const both = rows(plan([added, event("todo_updated", 2, {fields: ["bound_agent", "goal_bound"], goal_bound: flag})]))[0];
+    assert.deepEqual(both.field_sources, flag
+      ? {capability_binding_ref: 0, goal_bound: 1}
+      : {capability_binding_ref: 0, bound_agent: 1, goal_bound: 1});
+  }
 });
 
 test("removed continuation remains blocked until an explicit independent handoff repair", () => {