From e73e2615c05b0e7aa0860836e5e3785bd988c218 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:52:40 -0400 Subject: [PATCH 1/7] P265: preserve source-prefix clocks and require recorded times through EventTime --- src/agent_comms/cli_bridges/codex_protocol.py | 2 +- src/agent_comms/cli_protocol.py | 4 +- src/agent_comms/recorded_time.py | 34 ++++++++------ src/agent_comms/runtime_updates.py | 7 +-- tests/test_claude_tool_preview.py | 4 +- tests/test_claude_turn.py | 21 +++++---- tests/test_codex_earlier_pages.py | 2 +- tests/test_codex_items.py | 2 +- tests/test_transcript_event_time.py | 46 +++++++++++++++++-- 9 files changed, 88 insertions(+), 34 deletions(-) diff --git a/src/agent_comms/cli_bridges/codex_protocol.py b/src/agent_comms/cli_bridges/codex_protocol.py index 9523eccefd..d11883533e 100644 --- a/src/agent_comms/cli_bridges/codex_protocol.py +++ b/src/agent_comms/cli_bridges/codex_protocol.py @@ -741,7 +741,7 @@ def publication(self) -> RuntimeUpdate: if self.completedAtMs is None: raise ValueError(f'Item {self.item.id} has not finished') event_time = self.started_at.continued_from(self.completedAtMs / 1000) - return self.item.completed(self.turnId, self.item.events(event_time.transcript_timestamp())) + return self.item.completed(self.turnId, self.item.events(event_time.recorded_at('A Codex item'))) class ItemsPage(WireModel): diff --git a/src/agent_comms/cli_protocol.py b/src/agent_comms/cli_protocol.py index 93aa3650e3..86881d80fc 100644 --- a/src/agent_comms/cli_protocol.py +++ b/src/agent_comms/cli_protocol.py @@ -15,7 +15,7 @@ from .maintenance_barrier import MaintenanceHeldError from .messages import Message from .field_codec import TimestampText -from .recorded_time import MissingTranscriptTimestampError +from .recorded_time import UnrecordedTimeError from .external_transcript import RequestedEarlierPage if TYPE_CHECKING: @@ -43,7 +43,7 @@ class TimestampedWire(WireModel): @classmethod def require_source_time(cls, value): if isinstance(value, dict) and value.get('timestamp') is None: - raise MissingTranscriptTimestampError(f'{cls.__name__} has no recorded timestamp') + raise UnrecordedTimeError(f'{cls.__name__} has no recorded timestamp') return value @field_validator('timestamp', mode='before') diff --git a/src/agent_comms/recorded_time.py b/src/agent_comms/recorded_time.py index 35bb91357d..97d52c5e26 100644 --- a/src/agent_comms/recorded_time.py +++ b/src/agent_comms/recorded_time.py @@ -4,15 +4,21 @@ import time from abc import abstractmethod -from dataclasses import dataclass +from dataclasses import dataclass, replace from math import isfinite - +from typing import TYPE_CHECKING, TypeVar from .declared_family import DeclaredFamily from .field_codec import FieldCodec, FieldRepresentation -class MissingTranscriptTimestampError(ValueError): - """An external source cannot publish an event without its recorded time.""" +if TYPE_CHECKING: + from .transcript_events import TranscriptEvent + +Event = TypeVar("Event", bound="TranscriptEvent") + + +class UnrecordedTimeError(ValueError): + """An operation requires a source-recorded time, but none was supplied.""" class GeneratedTimestamp(FieldRepresentation): @@ -67,11 +73,11 @@ def clock(self) -> str: return "-" def require_recorded(self, label: str) -> None: - raise ValueError(f"{label} requires a recorded time.") + self.recorded_at(label) def recorded_at(self, label: str) -> float: - """The recorded instant; an unrecorded time refuses like ``require_recorded``.""" - raise ValueError(f"{label} requires a recorded time.") + """The recorded instant; callers requiring a time cannot invent one.""" + raise UnrecordedTimeError(f"{label} requires a recorded time.") def lateness(self, deadline: EventTime) -> Lateness: """How late this delivery time is against ``deadline``; either unrecorded refuses.""" @@ -83,8 +89,8 @@ def require_deadline(self) -> None: def require_delivery(self) -> None: self.require_recorded("A delivery") - def transcript_timestamp(self) -> float: - raise MissingTranscriptTimestampError("A transcript event requires its source's recorded time") + @abstractmethod + def stamp(self, event: Event) -> Event: ... @abstractmethod def continued_from(self, original_at: float): ... @@ -92,6 +98,9 @@ def continued_from(self, original_at: float): ... @dataclass(frozen=True) class UnrecordedEventTime(EventTime): + def stamp(self, event: Event) -> Event: + return event + def continued_from(self, original_at): return RecordedEventTime(original_at) @@ -104,9 +113,6 @@ def __post_init__(self): if type(self.at) not in (int, float) or not isfinite(self.at): raise ValueError("A recorded event time must be a finite Unix timestamp.") - def require_recorded(self, label: str) -> None: - pass - def since(self, timestamp: float) -> bool: return self.at >= timestamp @@ -119,8 +125,8 @@ def clock(self) -> str: def recorded_at(self, label: str) -> float: return self.at - def transcript_timestamp(self) -> float: - return self.at + def stamp(self, event: Event) -> Event: + return replace(event, timestamp=self.at) def continued_from(self, original_at): return self diff --git a/src/agent_comms/runtime_updates.py b/src/agent_comms/runtime_updates.py index b1b9f6bba6..07390707b4 100644 --- a/src/agent_comms/runtime_updates.py +++ b/src/agent_comms/runtime_updates.py @@ -49,8 +49,9 @@ def append(self, text: str, section: int) -> None: raise ValueError("Received a text delta for a non-text item") def finish(self, update: CompletedItem) -> tuple[ClosedDraft, FinishedItem]: - """Close with the completed record's own already decoded event times.""" - return ClosedDraft(self.started_at), FinishedItem(update.turn_id, update.item_id, update.events) + """Keep the observed source prefix's clock; completions are already dated.""" + events = tuple(self.started_at.stamp(event) for event in update.events) + return ClosedDraft(self.started_at), FinishedItem(update.turn_id, update.item_id, events) class NoPrefixDraft(DraftAccumulator): @@ -95,7 +96,7 @@ def events(self) -> tuple[TranscriptEvent, ...]: ) return ( (self.event_type(text, position=self.position, - timestamp=self.started_at.transcript_timestamp()),) if text else () + timestamp=self.started_at.recorded_at("A transcript event")),) if text else () ) def snapshot(self, turn_id: str, item_id: str) -> tuple[DraftItem, ...]: diff --git a/tests/test_claude_tool_preview.py b/tests/test_claude_tool_preview.py index 118983aab1..0bf1ffc0be 100644 --- a/tests/test_claude_tool_preview.py +++ b/tests/test_claude_tool_preview.py @@ -24,13 +24,15 @@ def test_claude_tool_names_arguments_and_diff_reach_shared_events(name, argument record=TimedRecord(position=0,uuid='record',timestamp='2026-10-17T00:00:00+00:00') tool=ToolUse(id='call',name=name,input=arguments) active,=tool.updates(record,session) - call=session.tools['call'] + call,source=session.tools['call'] + assert source is record assert call.name==native and call.arguments==expected and call.title==title assert active.detail == title result=ToolResult(tool_use_id='call',content=[{'type':'text','text':'done'}]) completed,_=result.updates(record,session) started,ended=completed.events assert started.raw_input==expected and started.title==title + assert started.timestamp == ended.timestamp == record.timestamp if patch is None: assert ended.diff is None else: diff --git a/tests/test_claude_turn.py b/tests/test_claude_turn.py index d09ad40c36..6cc2c60ab7 100644 --- a/tests/test_claude_turn.py +++ b/tests/test_claude_turn.py @@ -21,10 +21,10 @@ def test_unknown_turn_uses_the_reader_phase_for_every_item(question, phase, expectation): session = ClaudeSession(phase=phase()) record = TimedRecord(uuid='record', timestamp='2026-10-18T00:00:00Z', position=17) - session.started(ToolCall('call', 'read', {'path': 'a.py'})) + session.started(ToolCall('call', 'read', {'path': 'a.py'}), record) actions = { - 'finished': lambda: session.finished('item', (AssistantTranscript('done', position=17),)), - 'result': lambda: session.tool_result('call', 'done', True, record.position), + 'finished': lambda: session.finished('item', (AssistantTranscript('done', position=17, timestamp=record.timestamp),)), + 'result': lambda: session.tool_result('call', 'done', True, record), 'thinking': lambda: RedactedThinking().updates(record, session), 'tool': lambda: ToolUse(id='next', name='Read', input={'file_path': 'a.py'}).updates(record, session), } @@ -34,10 +34,15 @@ def test_unknown_turn_uses_the_reader_phase_for_every_item(question, phase, expe def test_prompt_identifies_finished_and_active_items_with_the_same_turn(): session = ClaudeSession(phase=Live()) - UserRecord(uuid='prompt', timestamp='2026-10-18T00:00:00Z', position=17, - promptId='turn-id', message={'content': []}).updates(session) + record = UserRecord(uuid='prompt', timestamp='2026-10-18T00:00:00Z', position=17, + promptId='turn-id', message={'content': []}) + record.updates(session) assert session.turn == KnownTurn('turn-id') - session.started(ToolCall('call', 'read', {'path': 'a.py'})) - completed, active = session.tool_result('call', 'done', True, 18) - assert completed == FinishedItem('turn-id', 'call', ToolCall('call', 'read', {'path': 'a.py'}).finished('done', True, 18)) + call = ToolCall('call', 'read', {'path': 'a.py'}) + session.started(call, record) + result = TimedRecord(uuid='result', timestamp='2026-10-18T00:00:01Z', position=18) + completed, active = session.tool_result('call', 'done', True, result) + assert completed == FinishedItem('turn-id', 'call', ( + call.started(record.position, record.timestamp), + call.ended('done', True, result.position, result.timestamp))) assert active == ActiveItem('turn-id', 'call', ActivityState.WORKING, '') diff --git a/tests/test_codex_earlier_pages.py b/tests/test_codex_earlier_pages.py index 35e0f47538..4d8658267e 100644 --- a/tests/test_codex_earlier_pages.py +++ b/tests/test_codex_earlier_pages.py @@ -170,7 +170,7 @@ def test_stopped_carry_preserves_external_items(tmp_path): root = tmp_path / 'root' root.mkdir() path = ExternalTranscript.path_in(root) - saved = json.dumps(FieldCodec.encode((AssistantTranscript('preserved', position='i'),))) + saved = json.dumps(FieldCodec.encode((AssistantTranscript('preserved', position='i', timestamp=1.0),))) # This historical conversion predates the separate event table. with closing(CoordinationStore.writer(path)) as connection: connection.execute('CREATE TABLE external_item (thread TEXT NOT NULL, turn_id TEXT NOT NULL, ' diff --git a/tests/test_codex_items.py b/tests/test_codex_items.py index 14ab5793ed..f23fb237f5 100644 --- a/tests/test_codex_items.py +++ b/tests/test_codex_items.py @@ -118,7 +118,7 @@ def test_coalescing_preserves_completion_and_unrelated_updates(): first = DraftItem("t", "a", TextDraft.from_parts(AssistantTranscript, ("H",), UnrecordedEventTime(), "a")) newer = DraftItem("t", "a", TextDraft.from_parts(AssistantTranscript, ("Hello",), UnrecordedEventTime(), "a")) other = DraftItem("t", "b", TextDraft.from_parts(AssistantTranscript, ("Other",), UnrecordedEventTime(), "b")) - final = FinishedItem("t", "a", (AssistantTranscript("Hello!", position="a"),)) + final = FinishedItem("t", "a", (AssistantTranscript("Hello!", position="a", timestamp=1.0),)) end = TurnEnded() assert RuntimeUpdate.coalesced([first, other, newer]) == [other, newer] assert RuntimeUpdate.coalesced([first, other, newer, final, end]) == [other, final, end] diff --git a/tests/test_transcript_event_time.py b/tests/test_transcript_event_time.py index 49a042ffb1..fcf03b4db8 100644 --- a/tests/test_transcript_event_time.py +++ b/tests/test_transcript_event_time.py @@ -56,10 +56,10 @@ def test_original_record_clock_survives_split_and_routed_projection(tmp_path, st def test_new_native_entry_inherits_original_clock_projection(): class ExternalNoteEntry(NativeEntry): - def _events(self, context): - return [AssistantTranscript("a new external note", position=context.position)] + def _events(self, context, timestamp): + return [AssistantTranscript("a new external note", position=context.position, timestamp=timestamp)] - # Adding a content declaration does not add another clock propagation site. + # The inherited entry owner decodes the source clock before constructing any content. stamp = "2026-09-28T23:12:33.123Z" expected = datetime.fromisoformat(stamp).timestamp() note = ExternalNoteEntry(timestamp=RecordedTimestamp(stamp)) @@ -153,3 +153,43 @@ async def serve(reader, writer): print( f"actual native clocks={expected}; ACP snapshot + rebuild exact; native history/proof unchanged" ) + + +@pytest.mark.parametrize('source', ('claude', 'codex')) +@pytest.mark.parametrize('stamp', (None, 'missing')) +def test_external_source_without_a_time_refuses_before_event_construction(source, stamp): + from pydantic import ValidationError + from agent_comms.cli_bridges.claude_protocol import TimedRecord + from agent_comms.cli_bridges.codex_rollout import ROLLOUT_RECORD + from agent_comms.recorded_time import UnrecordedTimeError + + record = ({'uuid': 'record', 'position': 0} if source == 'claude' else + {'type': 'response_item', 'payload': {'type': 'message', 'role': 'assistant', 'content': []}}) + if stamp is None: + record['timestamp'] = None + decode = TimedRecord.model_validate if source == 'claude' else ROLLOUT_RECORD.validate_python + with pytest.raises(ValidationError) as refused: + decode(record) + assert any(isinstance(error.get('ctx', {}).get('error'), UnrecordedTimeError) + for error in refused.value.errors()) + + +def test_codex_rollout_projects_the_enclosing_record_clock(): + from agent_comms.cli_bridges.codex_rollout import ROLLOUT_RECORD, RolloutState + + record = ROLLOUT_RECORD.validate_python({ + 'type': 'response_item', 'timestamp': '2026-10-19T18:00:00Z', + 'payload': {'type': 'message', 'role': 'assistant', 'id': 'item', + 'content': [{'text': 'saved source text'}]}, + }) + update, = record.updates(RolloutState(), 17) + assert update.events[0].timestamp == datetime.fromisoformat('2026-10-19T18:00:00Z').timestamp() + + +def test_malformed_shell_words_remain_visible_to_the_install_rule(): + from agent_comms.commander.send_guard import InstallCommand, simple_commands + + command = 'uv pip install package "unterminated' + words, = simple_commands(command) + assert words == tuple(command.split()) + assert InstallCommand.runs(words) From e95aeb9aa32d994c6e695a3c42483d34914d9110 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:24:49 -0400 Subject: [PATCH 2/7] Date the external events saved before event time became required #2011 made every transcript event carry its time but left 45,458 saved Claude tool events undated, so every rotate crashed decoding them. The carry takes each event's time from its Claude Code transcript (all 45,458 found, 6 s) and refuses if any stays undated; the unjudged query loses its undated branch. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/external_transcript.py | 15 +---- tools/cutover/carry_external_event_times.py | 73 +++++++++++++++++++++ 2 files changed, 75 insertions(+), 13 deletions(-) create mode 100644 tools/cutover/carry_external_event_times.py 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/tools/cutover/carry_external_event_times.py b/tools/cutover/carry_external_event_times.py new file mode 100644 index 0000000000..6355896a72 --- /dev/null +++ b/tools/cutover/carry_external_event_times.py @@ -0,0 +1,73 @@ +"""Date the external events saved before #2011 made every transcript event carry its time. + +Those events are Claude tool calls; their Claude Code transcripts record when each +tool_use and tool_result happened. Each undated event takes that time; an event the +transcripts cannot date stops the carry, since readers no longer accept undated events. +""" +from __future__ import annotations + +import json +import re +from contextlib import closing +from datetime import datetime +from pathlib import Path + +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+)"') +# Which transcript content item records each saved event kind. +RECORDED_BY = {"tool_use": (ToolStartTranscript.declared_name, "id"), + "tool_result": (ToolEndTranscript.declared_name, "tool_use_id")} + + +def recorded_times(wanted: set[str]) -> dict[tuple[str, str], float]: + """(kind, call id) to the time its transcript line records, for the wanted calls.""" + times: dict[tuple[str, str], float] = {} + for path in TRANSCRIPTS.glob("**/*.jsonl"): + with path.open(errors="replace") as lines: + for line in lines: + if "toolu_" not in line or not wanted.intersection(CALL.findall(line)): + continue + record = json.loads(line) + at = datetime.fromisoformat(record["timestamp"]).timestamp() + for item in record["message"]["content"]: + kind, key = RECORDED_BY.get(item.get("type"), (None, None)) + if kind and item[key] in wanted: + times.setdefault((kind, item[key]), at) + 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]}") + for kind, call in undated: + db.execute(f'UPDATE "{table}" SET event=json_set(event, \'$.timestamp\', ?) ' + 'WHERE timestamp IS NULL AND kind=? AND call_id=?', + (times[kind, call], kind, call)) + 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() From db384b26609f1277b9dfd6410634a0d6df9dc9be Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:39:25 -0400 Subject: [PATCH 3/7] Carry reads tool blocks through Claude's decoders and dates all rows in one update Co-Authored-By: Claude Opus 5.5 --- tools/cutover/carry_external_event_times.py | 66 ++++++++++++++------- 1 file changed, 45 insertions(+), 21 deletions(-) diff --git a/tools/cutover/carry_external_event_times.py b/tools/cutover/carry_external_event_times.py index 6355896a72..647c258bff 100644 --- a/tools/cutover/carry_external_event_times.py +++ b/tools/cutover/carry_external_event_times.py @@ -1,17 +1,21 @@ """Date the external events saved before #2011 made every transcript event carry its time. -Those events are Claude tool calls; their Claude Code transcripts record when each -tool_use and tool_result happened. Each undated event takes that time; an event the -transcripts cannot date stops the carry, since readers no longer accept undated events. +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 +from contextlib import closing +from pathlib import Path + import json import re -from contextlib import closing from datetime import datetime -from pathlib import Path +from agent_comms.cli_bridges.claude_protocol import ToolResult, ToolUse from agent_comms.coordination_database import CoordinationStore from agent_comms.external_transcript import ExternalEventRow, ExternalTranscript from agent_comms.transcript_events import ToolEndTranscript, ToolStartTranscript @@ -20,13 +24,16 @@ TRANSCRIPTS = Path.home() / ".claude" / "projects" CALL = re.compile(r'"(?:id|tool_use_id)":"(toolu_\w+)"') -# Which transcript content item records each saved event kind. -RECORDED_BY = {"tool_use": (ToolStartTranscript.declared_name, "id"), - "tool_result": (ToolEndTranscript.declared_name, "tool_use_id")} +# Claude's own block decoders: which saved event each block dates. +BLOCKS = { + ToolUse.wire_tag: (ToolUse, ToolStartTranscript, lambda block: block.id), + ToolResult.wire_tag: (ToolResult, ToolEndTranscript, lambda block: block.tool_use_id), +} def recorded_times(wanted: set[str]) -> dict[tuple[str, str], float]: - """(kind, call id) to the time its transcript line records, for the wanted calls.""" + """(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 the full file is read here.""" times: dict[tuple[str, str], float] = {} for path in TRANSCRIPTS.glob("**/*.jsonl"): with path.open(errors="replace") as lines: @@ -34,11 +41,16 @@ def recorded_times(wanted: set[str]) -> dict[tuple[str, str], float]: if "toolu_" not in line or not wanted.intersection(CALL.findall(line)): continue record = json.loads(line) + content = record.get("message", {}).get("content") + if "timestamp" not in record or not isinstance(content, list): + continue # snapshots and queue lines quote ids but time nothing at = datetime.fromisoformat(record["timestamp"]).timestamp() - for item in record["message"]["content"]: - kind, key = RECORDED_BY.get(item.get("type"), (None, None)) - if kind and item[key] in wanted: - times.setdefault((kind, item[key]), at) + for item in content: + block, event, call_of = BLOCKS.get(item.get("type"), (None, None, None)) + if block is not None: + call = call_of(block.model_validate(item)) + if call in wanted: + times.setdefault((event.declared_name, call), at) return times @@ -54,19 +66,31 @@ def carry(cls, root: Path, *, apply: bool) -> Carried: 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() + 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]}") - for kind, call in undated: - db.execute(f'UPDATE "{table}" SET event=json_set(event, \'$.timestamp\', ?) ' - 'WHERE timestamp IS NULL AND kind=? AND call_id=?', - (times[kind, call], kind, call)) + 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}\" SET event=json_set(event, '$.timestamp', (SELECT at FROM dated d " + f'WHERE d.kind="{table}".kind AND d.call_id="{table}".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 ()) + return Carried( + len(undated), {"events": len(undated)}, (path.name,) if apply and undated else () + ) if __name__ == "__main__": From bfa9807a5eb6bd5c0b71556801d81d6928783234 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:49:05 -0400 Subject: [PATCH 4/7] Wrap the carry's SQL and look up block decoders by membership Co-Authored-By: Claude Opus 5.5 --- tools/cutover/carry_external_event_times.py | 32 ++++++++++++--------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/tools/cutover/carry_external_event_times.py b/tools/cutover/carry_external_event_times.py index 647c258bff..d18bcc2e99 100644 --- a/tools/cutover/carry_external_event_times.py +++ b/tools/cutover/carry_external_event_times.py @@ -8,12 +8,12 @@ from __future__ import annotations -from contextlib import closing -from pathlib import Path - import json import re +from collections.abc import Callable +from contextlib import closing from datetime import datetime +from pathlib import Path from agent_comms.cli_bridges.claude_protocol import ToolResult, ToolUse from agent_comms.coordination_database import CoordinationStore @@ -25,9 +25,11 @@ TRANSCRIPTS = Path.home() / ".claude" / "projects" CALL = re.compile(r'"(?:id|tool_use_id)":"(toolu_\w+)"') # Claude's own block decoders: which saved event each block dates. -BLOCKS = { - ToolUse.wire_tag: (ToolUse, ToolStartTranscript, lambda block: block.id), - ToolResult.wire_tag: (ToolResult, ToolEndTranscript, lambda block: block.tool_use_id), +BLOCKS: dict[str, tuple[tuple[type[ToolUse | ToolResult], str, Callable[..., str]], ...]] = { + ToolUse.wire_tag: ((ToolUse, ToolStartTranscript.declared_name, lambda block: block.id),), + ToolResult.wire_tag: ( + (ToolResult, ToolEndTranscript.declared_name, lambda block: block.tool_use_id), + ), } @@ -45,12 +47,11 @@ def recorded_times(wanted: set[str]) -> dict[tuple[str, str], float]: if "timestamp" not in record or not isinstance(content, list): continue # snapshots and queue lines quote ids but time nothing at = datetime.fromisoformat(record["timestamp"]).timestamp() - for item in content: - block, event, call_of = BLOCKS.get(item.get("type"), (None, None, None)) - if block is not None: + for item in (item for item in content if item["type"] in BLOCKS): + for block, kind, call_of in BLOCKS[item["type"]]: call = call_of(block.model_validate(item)) if call in wanted: - times.setdefault((event.declared_name, call), at) + times.setdefault((kind, call), at) return times @@ -75,17 +76,20 @@ def carry(cls, root: Path, *, apply: bool) -> Carried: 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. + # 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))" + "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}\" SET event=json_set(event, '$.timestamp', (SELECT at FROM dated d " - f'WHERE d.kind="{table}".kind AND d.call_id="{table}".call_id)) WHERE timestamp IS NULL' + 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( From 4534a64e7aa5cf0301684ff53220186759a54b97 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:53:18 -0400 Subject: [PATCH 5/7] Decode each transcript line with the bridge's record decoder Co-Authored-By: Claude Opus 5.5 --- tools/cutover/carry_external_event_times.py | 52 ++++++++++++--------- 1 file changed, 30 insertions(+), 22 deletions(-) diff --git a/tools/cutover/carry_external_event_times.py b/tools/cutover/carry_external_event_times.py index d18bcc2e99..93787efa27 100644 --- a/tools/cutover/carry_external_event_times.py +++ b/tools/cutover/carry_external_event_times.py @@ -12,10 +12,17 @@ import re from collections.abc import Callable from contextlib import closing -from datetime import datetime from pathlib import Path -from agent_comms.cli_bridges.claude_protocol import ToolResult, ToolUse +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 @@ -24,34 +31,35 @@ TRANSCRIPTS = Path.home() / ".claude" / "projects" CALL = re.compile(r'"(?:id|tool_use_id)":"(toolu_\w+)"') -# Claude's own block decoders: which saved event each block dates. -BLOCKS: dict[str, tuple[tuple[type[ToolUse | ToolResult], str, Callable[..., str]], ...]] = { - ToolUse.wire_tag: ((ToolUse, ToolStartTranscript.declared_name, lambda block: block.id),), - ToolResult.wire_tag: ( - (ToolResult, ToolEndTranscript.declared_name, lambda block: block.tool_use_id), - ), -} +# 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 the full file is read here.""" + 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 line in lines: + for position, line in enumerate(lines): if "toolu_" not in line or not wanted.intersection(CALL.findall(line)): continue - record = json.loads(line) - content = record.get("message", {}).get("content") - if "timestamp" not in record or not isinstance(content, list): - continue # snapshots and queue lines quote ids but time nothing - at = datetime.fromisoformat(record["timestamp"]).timestamp() - for item in (item for item in content if item["type"] in BLOCKS): - for block, kind, call_of in BLOCKS[item["type"]]: - call = call_of(block.model_validate(item)) - if call in wanted: - times.setdefault((kind, call), at) + 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 @@ -87,7 +95,7 @@ def carry(cls, root: Path, *, apply: bool) -> Carried: ((kind, call, at) for (kind, call), at in times.items()), ) db.execute( - f'UPDATE "{table}" AS t SET event=json_set(event, \'$.timestamp\',' + 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" ) From 4038452f6fabb71f01b6046dd1b5fe2667cc6e68 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 20:17:50 -0400 Subject: [PATCH 6/7] P265: retain malformed shell input after integrating the source-backed carry --- src/agent_comms/commander/send_guard.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/src/agent_comms/commander/send_guard.py b/src/agent_comms/commander/send_guard.py index 642e93b2e8..d302f042d4 100644 --- a/src/agent_comms/commander/send_guard.py +++ b/src/agent_comms/commander/send_guard.py @@ -35,11 +35,14 @@ def simple_commands(command: str) -> tuple[Words, ...]: lexer = shlex.shlex(command, posix=True, punctuation_chars=True) lexer.whitespace_split = True commands: list[list[str]] = [[]] - for word in lexer: - if word and set(word) <= set(';&|()'): - commands.append([]) - else: - commands[-1].append(word) + try: + for word in lexer: + if word and set(word) <= set(';&|()'): + commands.append([]) + else: + commands[-1].append(word) + except ValueError: + return (tuple(command.split()),) return tuple(tuple(words) for words in commands if words) From 47a5a1e3fc7810a95820d7bdd998e705b92e85e9 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 20:20:09 -0400 Subject: [PATCH 7/7] P265: verify the source-backed carry on a snapshot of the actual event store --- tests/verify_external_event_times.py | 39 ++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 tests/verify_external_event_times.py diff --git a/tests/verify_external_event_times.py b/tests/verify_external_event_times.py new file mode 100644 index 0000000000..e1d80b326a --- /dev/null +++ b/tests/verify_external_event_times.py @@ -0,0 +1,39 @@ +"""Run the source-backed carry on a SQLite snapshot, preserving the live store.""" +from __future__ import annotations + +import json +import sys +from contextlib import closing +from pathlib import Path + +from agent_comms.commander.order import COMMS +from agent_comms.coordination_database import CoordinationStore +from agent_comms.external_transcript import ExternalEventRow, ExternalTranscript +from agent_comms.field_codec import FieldCodec +from agent_comms.transcript_events import TranscriptEvent +from cutover.carry_external_event_times import ExternalEventTimes + + +root = Path(sys.argv[1]) +root.mkdir(parents=True, exist_ok=False) +source = ExternalTranscript.path_in(COMMS.root) +copy = root / source.name +with closing(CoordinationStore.read_snapshot(source)) as reader: + with closing(CoordinationStore.writer(copy)) as writer: + reader.backup(writer) +with CoordinationStore.reading(copy) as connection: + before, = connection.execute( + f'SELECT count(*) FROM "{ExternalEventRow.declared_name}" WHERE timestamp IS NULL' + ).fetchone() +print('copied undated rows:', before, flush=True) +result = ExternalEventTimes.carry(root, apply=True) +print(json.dumps(result.report('carry_external_event_times', True)), flush=True) +assert ExternalEventTimes.carry(root, apply=True).changed == 0 +count = 0 +with CoordinationStore.reading(copy) as connection: + for encoded, in connection.execute(f'SELECT event FROM "{ExternalEventRow.declared_name}"'): + event = FieldCodec.decode(TranscriptEvent, FieldCodec.parse(encoded)) + assert isinstance(event.timestamp, float) + count += 1 +print('all copied events decoded with recorded float timestamps:', count) +print('source-backed carry idempotent; live store opened read-only')