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/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) 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/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) 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') 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()