Date stored external events; cutover waits only on review queue and control plane

src/agent_comms/external_transcript.py

@@ -212,19 +212,8 @@ def unjudged_query(cls, thread, todo):
         judged = tuple(verdict.tool_call_id for verdict in todo.verdicts)
         placeholders = lambda values: ",".join("?" for _ in values)
         since = todo.order.at.recorded_at("An order")
-        # SQL NULL is the original undated fact, decoded through FieldCodec
-        # as UnrecordedEventTime when a row is read. No time is substituted.
-        table = cls.declared_name
-        selected = f'u.thread=? AND u.kind=? AND u.tool IN ({placeholders(tools)})'
-        # Two ranges keep timestamp in the index search, rather than scanning
-        # every historical edit to evaluate an OR for each row.
-        where = (
-            f'e.rowid IN (SELECT u.rowid FROM "{table}" u WHERE {selected} AND u.timestamp>=? '
-            f'UNION ALL SELECT u.rowid FROM "{table}" u WHERE {selected} '
-            'AND u.timestamp IS NULL)'
-        )
-        scope = (thread, ToolStartTranscript.declared_name, *tools)
-        parameters = (*scope, since, *scope)
+        where = f'e.thread=? AND e.kind=? AND e.tool IN ({placeholders(tools)}) AND e.timestamp>=?'
+        parameters = (thread, ToolStartTranscript.declared_name, *tools, since)
         if judged:
             where += f' AND e.call_id NOT IN ({placeholders(judged)})'
             parameters += judged

src/agent_comms/plan_phases.py

@@ -95,7 +95,7 @@ class ControlPlanePhase(PlanPhase):
 
 
 class CutoverPhase(PlanPhase):
-    after = (ReviewQueuePhase, StoryPhase, CommanderPagePhase, ControlPlanePhase)
+    after = (ReviewQueuePhase, ControlPlanePhase)
 
 
 class SkillPhase(PlanPhase):

tools/cutover/carry_external_event_times.py

@@ -0,0 +1,109 @@
+"""Date the external events saved before #2011 made every transcript event carry its time.
+
+Those events are Claude tool calls. Replaying each Claude Code transcript through the
+bridge's own reader yields the same events with their recorded times; each undated row
+takes its event's time. An event no transcript dates stops the carry, since readers no
+longer accept undated events.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from collections.abc import Callable
+from contextlib import closing
+from pathlib import Path
+
+from pydantic import ValidationError
+
+from agent_comms.cli_bridges.claude_protocol import (
+    CLAUDE_RECORD,
+    AssistantRecord,
+    ToolResult,
+    ToolUse,
+    UserRecord,
+)
+from agent_comms.coordination_database import CoordinationStore
+from agent_comms.external_transcript import ExternalEventRow, ExternalTranscript
+from agent_comms.transcript_events import ToolEndTranscript, ToolStartTranscript
+
+from .carry import Carried, StoreCarry
+
+TRANSCRIPTS = Path.home() / ".claude" / "projects"
+CALL = re.compile(r'"(?:id|tool_use_id)":"(toolu_\w+)"')
+# Claude's own block types, and which saved event each one dates.
+BLOCKS: tuple[tuple[type[ToolUse | ToolResult], str, Callable[..., str]], ...] = (
+    (ToolUse, ToolStartTranscript.declared_name, lambda block: block.id),
+    (ToolResult, ToolEndTranscript.declared_name, lambda block: block.tool_use_id),
+)
+
+
+def recorded_times(wanted: set[str]) -> dict[tuple[str, str], float]:
+    """(event kind, call id) to the time its transcript line records, for the wanted calls.
+    The bridge's reader replays only a recent window, so every line is decoded here with
+    the bridge's record decoder."""
+    times: dict[tuple[str, str], float] = {}
+    for path in TRANSCRIPTS.glob("**/*.jsonl"):
+        with path.open(errors="replace") as lines:
+            for position, line in enumerate(lines):
+                if "toolu_" not in line or not wanted.intersection(CALL.findall(line)):
+                    continue
+                try:
+                    record = CLAUDE_RECORD.validate_python(
+                        {**json.loads(line), "position": position}
+                    )
+                except ValidationError:
+                    continue  # snapshots and queue lines quote call ids but are no timed record
+                if not isinstance(record, AssistantRecord | UserRecord):
+                    continue
+                for block in record.message.content:
+                    for decoded, kind, call_of in BLOCKS:
+                        if isinstance(block, decoded) and call_of(block) in wanted:
+                            times.setdefault((kind, call_of(block)), record.timestamp)
+    return times
+
+
+class ExternalEventTimes(StoreCarry):
+    store = ExternalTranscript
+
+    @classmethod
+    def carry(cls, root: Path, *, apply: bool) -> Carried:
+        path = cls.store.path_in(root)
+        if not path.exists():
+            return Carried(0)
+        table = ExternalEventRow.declared_name
+        with closing(CoordinationStore.writer(path)) as db:
+            db.execute("BEGIN IMMEDIATE")
+            undated = db.execute(
+                f'SELECT DISTINCT kind, call_id FROM "{table}" WHERE timestamp IS NULL'
+            ).fetchall()
+            times = recorded_times({call for _, call in undated})
+            missing = [event for event in undated if tuple(event) not in times]
+            if missing:
+                db.rollback()
+                raise ValueError(
+                    f"{len(missing)} saved events have no transcript time, e.g. {missing[:3]}"
+                )
+            # One pass over the table: per-event updates would each scan it,
+            # since no index leads with call_id.
+            db.execute(
+                "CREATE TEMP TABLE dated"
+                " (kind TEXT, call_id TEXT, at REAL, PRIMARY KEY (kind, call_id))"
+            )
+            db.executemany(
+                "INSERT INTO dated VALUES (?, ?, ?)",
+                ((kind, call, at) for (kind, call), at in times.items()),
+            )
+            db.execute(
+                f"UPDATE \"{table}\" AS t SET event=json_set(event, '$.timestamp',"
+                " (SELECT at FROM dated d WHERE d.kind=t.kind AND d.call_id=t.call_id))"
+                " WHERE timestamp IS NULL"
+            )
+            db.commit() if apply else db.rollback()
+        return Carried(
+            len(undated), {"events": len(undated)}, (path.name,) if apply and undated else ()
+        )
+
+
+if __name__ == "__main__":
+    ExternalEventTimes.main()

tools/upgrade_bus.py

@@ -256,6 +256,9 @@ def rehearsal_home(scratch: Path, root: Path, copy: Path) -> Path:
             record["root"] = str(copy)
         (home / route).write_text(json.dumps(record))
         (home / route).chmod(0o600)
+    # Claude Code's transcripts are read-only input that carries date from;
+    # the copy reads the real ones.
+    (home / ".claude").symlink_to(Path.home() / ".claude")
     return home