diff --git a/src/agent_comms/external_transcript.py b/src/agent_comms/external_transcript.py index d2eb3a25b0..d254b81736 100644 --- a/src/agent_comms/external_transcript.py +++ b/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 diff --git a/src/agent_comms/plan_phases.py b/src/agent_comms/plan_phases.py index 4fb954a3b9..2a78e0159f 100644 --- a/src/agent_comms/plan_phases.py +++ b/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): diff --git a/tools/cutover/carry_external_event_times.py b/tools/cutover/carry_external_event_times.py new file mode 100644 index 0000000000..93787efa27 --- /dev/null +++ b/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() diff --git a/tools/upgrade_bus.py b/tools/upgrade_bus.py index 62369f88d1..ebab0fbfbb 100644 --- a/tools/upgrade_bus.py +++ b/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