P265: preserve source-prefix clocks with required event times

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):

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

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)
 
 

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/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,15 +89,18 @@ 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): ...
 
 
 @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

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, ...]:

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:

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, '')

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, '

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]

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)

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