From d9c2fc6a7ed49d0ca1a3fa3322b6ae0d4b8dc843 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:26:43 -0400 Subject: [PATCH 1/4] P265: keep malformed shell commands visible to hard-rule checks --- src/agent_comms/commander/send_guard.py | 34 ++++++++++++++++++------- 1 file changed, 25 insertions(+), 9 deletions(-) diff --git a/src/agent_comms/commander/send_guard.py b/src/agent_comms/commander/send_guard.py index 642e93b2e8..ce167661aa 100644 --- a/src/agent_comms/commander/send_guard.py +++ b/src/agent_comms/commander/send_guard.py @@ -30,17 +30,33 @@ Words = tuple[str, ...] -def simple_commands(command: str) -> tuple[Words, ...]: +@dataclass(frozen=True) +class SimpleCommand: + words: Words + + +class ParsedCommand(SimpleCommand): + """The shell lexer identified this simple command's argument boundaries.""" + + +class UnparsedCommand(SimpleCommand): + """Malformed shell input still exposes its whitespace-separated words to rules.""" + + +def simple_commands(command: str) -> tuple[SimpleCommand, ...]: """The one parse of a command line: its simple commands, split at shell operators.""" 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) - return tuple(tuple(words) for words in commands if words) + try: + for word in lexer: + if word and set(word) <= set(';&|()'): + commands.append([]) + else: + commands[-1].append(word) + except ValueError: + return (UnparsedCommand(tuple(command.split())),) + return tuple(ParsedCommand(tuple(words)) for words in commands if words) @dataclass(frozen=True) @@ -77,9 +93,9 @@ def refused(cls, words: Words, session: CommanderSession) -> tuple[str, ...]: .. @classmethod def refusals(cls, command: str, session: CommanderSession) -> tuple[str, ...]: - return tuple(reason for words in simple_commands(command) + return tuple(reason for simple in simple_commands(command) for kind in cls.members_with(cls) if not inspect.isabstract(kind) - for reason in kind.refused(words, session)) + for reason in kind.refused(simple.words, session)) @dataclass(frozen=True) From b80f16c1f43b2d47f49b1d0e394a8e3549b0c1cd Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:29:32 -0400 Subject: [PATCH 2/4] P265: remove unused command-state wrappers --- src/agent_comms/commander/send_guard.py | 23 +++++------------------ 1 file changed, 5 insertions(+), 18 deletions(-) diff --git a/src/agent_comms/commander/send_guard.py b/src/agent_comms/commander/send_guard.py index ce167661aa..d302f042d4 100644 --- a/src/agent_comms/commander/send_guard.py +++ b/src/agent_comms/commander/send_guard.py @@ -30,20 +30,7 @@ Words = tuple[str, ...] -@dataclass(frozen=True) -class SimpleCommand: - words: Words - - -class ParsedCommand(SimpleCommand): - """The shell lexer identified this simple command's argument boundaries.""" - - -class UnparsedCommand(SimpleCommand): - """Malformed shell input still exposes its whitespace-separated words to rules.""" - - -def simple_commands(command: str) -> tuple[SimpleCommand, ...]: +def simple_commands(command: str) -> tuple[Words, ...]: """The one parse of a command line: its simple commands, split at shell operators.""" lexer = shlex.shlex(command, posix=True, punctuation_chars=True) lexer.whitespace_split = True @@ -55,8 +42,8 @@ def simple_commands(command: str) -> tuple[SimpleCommand, ...]: else: commands[-1].append(word) except ValueError: - return (UnparsedCommand(tuple(command.split())),) - return tuple(ParsedCommand(tuple(words)) for words in commands if words) + return (tuple(command.split()),) + return tuple(tuple(words) for words in commands if words) @dataclass(frozen=True) @@ -93,9 +80,9 @@ def refused(cls, words: Words, session: CommanderSession) -> tuple[str, ...]: .. @classmethod def refusals(cls, command: str, session: CommanderSession) -> tuple[str, ...]: - return tuple(reason for simple in simple_commands(command) + return tuple(reason for words in simple_commands(command) for kind in cls.members_with(cls) if not inspect.isabstract(kind) - for reason in kind.refused(simple.words, session)) + for reason in kind.refused(words, session)) @dataclass(frozen=True) From 0fdf45b260d8ca4594cdf9d3ec0a5d42704f3eb1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:34:21 -0400 Subject: [PATCH 3/4] P265: construct transcript events with their source record time --- .../cli_bridges/claude_protocol.py | 29 ++++---- src/agent_comms/cli_bridges/codex_protocol.py | 71 +++++++++---------- src/agent_comms/cli_bridges/codex_rollout.py | 24 +++---- src/agent_comms/cli_bridges/opencode_parts.py | 25 ++++--- src/agent_comms/cli_protocol.py | 22 +++++- src/agent_comms/native_entries.py | 11 ++- src/agent_comms/pi_payloads.py | 42 +++++------ src/agent_comms/recorded_time.py | 21 +++--- src/agent_comms/runtime_updates.py | 8 +-- src/agent_comms/tool_calls.py | 20 +++--- src/agent_comms/transcript_events.py | 18 ++--- src/agent_comms/transcripts.py | 5 +- 12 files changed, 151 insertions(+), 145 deletions(-) diff --git a/src/agent_comms/cli_bridges/claude_protocol.py b/src/agent_comms/cli_bridges/claude_protocol.py index 6b40971ae3..d3a956a797 100644 --- a/src/agent_comms/cli_bridges/claude_protocol.py +++ b/src/agent_comms/cli_bridges/claude_protocol.py @@ -4,7 +4,6 @@ import json from abc import ABC, abstractmethod from dataclasses import dataclass, field -from datetime import datetime from typing import ClassVar, Literal from xml.etree import ElementTree @@ -18,7 +17,7 @@ ) from agent_comms.activity import ActivityState -from agent_comms.cli_protocol import TaggedWire, WireHeader, WireModel +from agent_comms.cli_protocol import TaggedWire, TimestampedWire, WireHeader, WireModel from agent_comms.runtime_updates import ( ActiveItem, FinishedItem, @@ -203,7 +202,7 @@ def active(self, tool_id: str, phase: ClaudeReaderPhase, @dataclass class ClaudeSession: turn: ClaudeTurn = field(default_factory=UnknownTurn) - tools: dict[str, ToolCall] = field(default_factory=dict) + tools: dict[str, tuple[ToolCall, TimedRecord]] = field(default_factory=dict) phase: ClaudeReaderPhase = field(default_factory=CatchUp) project: str = '' @@ -213,30 +212,26 @@ def located(self, cwd: str) -> tuple[RuntimeUpdate, ...]: self.project = cwd return (SessionFacts(project=cwd),) - def started(self, call: ToolCall) -> None: - self.tools[call.id] = call + def started(self, call: ToolCall, record: TimedRecord) -> None: + self.tools[call.id] = call, record - def tool_result(self, tool_id: str, text: str, ok: bool, position: int) -> tuple[RuntimeUpdate, ...]: + def tool_result(self, tool_id: str, text: str, ok: bool, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: try: - call = self.tools.pop(tool_id) + call, started = self.tools.pop(tool_id) except KeyError: return self.phase.missing_tool(tool_id) - return (*self.finished(tool_id, call.finished(text, ok, position)), + return (*self.finished(tool_id, ( + call.started(started.position, started.timestamp), + call.ended(text, ok, record.position, record.timestamp))), *self.turn.active(tool_id, self.phase, ActivityState.WORKING, '')) def finished(self, item: str, events: tuple[TranscriptEvent, ...]) -> tuple[RuntimeUpdate, ...]: return self.turn.finished(item, events, self.phase) -class TimedRecord(WireModel): +class TimedRecord(TimestampedWire): position: int = Field(exclude=True) uuid: str - timestamp: float - - @field_validator('timestamp', mode='before') - @classmethod - def epoch(cls, value: str) -> float: - return datetime.fromisoformat(value).timestamp() class ClaudeHeader(WireHeader): @@ -331,7 +326,7 @@ class ToolUse(AssistantBlock): def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: call = self.call() - session.started(call) + session.started(call, record) return session.turn.active(self.id, session.phase, ActivityState.WORKING, call.title) @@ -546,7 +541,7 @@ class ToolResult(UserBlock, ContentBlocks): is_error: bool = False def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return session.tool_result(self.tool_use_id, '\n'.join(part.result_text() for part in self.content), not self.is_error, record.position) + return session.tool_result(self.tool_use_id, '\n'.join(part.result_text() for part in self.content), not self.is_error, record) class UserMessage(ContentBlocks): diff --git a/src/agent_comms/cli_bridges/codex_protocol.py b/src/agent_comms/cli_bridges/codex_protocol.py index d2baeb3663..9523eccefd 100644 --- a/src/agent_comms/cli_bridges/codex_protocol.py +++ b/src/agent_comms/cli_bridges/codex_protocol.py @@ -293,8 +293,8 @@ class Item(TaggedWire, ABC): doing: ClassVar[str] = '' @abstractmethod - def events(self) -> tuple[TranscriptEvent, ...]: - """The finished item's presentation.""" + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + """The finished item's presentation at its source record's time.""" def completed(self, turn_id: str, events: tuple[TranscriptEvent, ...]) -> RuntimeUpdate: return CompletedItem(turn_id, self.id, events) @@ -312,8 +312,8 @@ class TextBody(WireModel): def accumulator(self, started_at: EventTime) -> TextDraft: return TextDraft.from_parts(AssistantTranscript, (self.text,), started_at, self.id) - def events(self) -> tuple[TranscriptEvent, ...]: - return self.accumulator(UnrecordedEventTime()).events + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return self.accumulator(EventTime.capture(timestamp)).events class AssistantItem(TextBody, Item): @@ -335,8 +335,8 @@ def accumulator(self, started_at: EventTime) -> ParagraphDraft: # Only public summaries; private/raw reasoning is never projected. return ParagraphDraft.from_parts(ThinkingTranscript, self.summary, started_at, self.id) - def events(self) -> tuple[TranscriptEvent, ...]: - return self.accumulator(UnrecordedEventTime()).events + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return self.accumulator(EventTime.capture(timestamp)).events class UserItem(Item): @@ -344,8 +344,8 @@ class UserItem(Item): content: list[CodexUserInput] clientId: str | None = None - def events(self) -> tuple[TranscriptEvent, ...]: - return (UserTranscript('\n'.join(part.display for part in self.content), position=self.id),) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return (UserTranscript('\n'.join(part.display for part in self.content), position=self.id, timestamp=timestamp),) def completed(self, turn_id: str, events: tuple[TranscriptEvent, ...]) -> RuntimeUpdate: # Bus inputs already appear in Core's inbox; the live view does not echo them as new input. @@ -377,8 +377,8 @@ def call(self) -> ToolCall: tool = ToolDeclaration.record(self.tool_name, self.arguments, call_id=self.id) return shown.recorded(tool) if tool else shown - def events(self) -> tuple[TranscriptEvent, ...]: - return self.call.finished(self.output, self.succeeded, self.id) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return self.call.finished(self.output, self.succeeded, self.id, timestamp) @property def activity_detail(self) -> str: @@ -529,14 +529,14 @@ class FunctionOutputItem(Item): name: str output: str - def events(self) -> tuple[TranscriptEvent, ...]: - return (ToolEndTranscript(position=self.id, tool_call_id=self.id, tool_name=self.name, text=self.output),) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return (ToolEndTranscript(position=self.id, timestamp=timestamp, tool_call_id=self.id, tool_name=self.name, text=self.output),) class SendOutputItem(FunctionOutputItem): name: Literal[SEND_TOOL] - def events(self) -> tuple[TranscriptEvent, ...]: + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: # The bridge's send result is already rendered from the bus in Toad. return () @@ -545,14 +545,14 @@ class CompactionItem(Item): wire_tag = 'contextCompaction' doing = 'Compacting context' - def events(self) -> tuple[TranscriptEvent, ...]: - return (NoticeTranscript('Context compacted.', position=self.id),) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return (NoticeTranscript('Context compacted.', position=self.id, timestamp=timestamp),) class HookPromptItem(Item): wire_tag = 'hookPrompt' - def events(self) -> tuple[TranscriptEvent, ...]: + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: # Harness instructions are not assistant output. return () @@ -560,8 +560,8 @@ def events(self) -> tuple[TranscriptEvent, ...]: class ReviewBody(WireModel): review: str - def events(self) -> tuple[TranscriptEvent, ...]: - return (NoticeTranscript(self.review, position=self.id),) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return (NoticeTranscript(self.review, position=self.id, timestamp=timestamp),) class EnteredReviewItem(ReviewBody, Item): @@ -576,8 +576,8 @@ class SleepItem(Item): wire_tag = 'sleep' durationMs: int - def events(self) -> tuple[TranscriptEvent, ...]: - return (NoticeTranscript(f'Waited ({self.durationMs} ms requested).', position=self.id),) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return (NoticeTranscript(f'Waited ({self.durationMs} ms requested).', position=self.id, timestamp=timestamp),) @property def activity_detail(self) -> str: @@ -588,8 +588,8 @@ class ImageViewItem(Item): wire_tag = 'imageView' path: str - def events(self) -> tuple[TranscriptEvent, ...]: - return (NoticeTranscript(f'Image: {self.path}', position=self.id),) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return (NoticeTranscript(f'Image: {self.path}', position=self.id, timestamp=timestamp),) class ImageGenerationItem(Item): @@ -598,9 +598,9 @@ class ImageGenerationItem(Item): result: str savedPath: str | None = None - def events(self) -> tuple[TranscriptEvent, ...]: + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: return (NoticeTranscript(f'Image generation: {self.status}' + - (f' — {self.savedPath}' if self.savedPath else ''), position=self.id),) + (f' — {self.savedPath}' if self.savedPath else ''), position=self.id, timestamp=timestamp),) class WebSearchItem(Item): @@ -609,9 +609,9 @@ class WebSearchItem(Item): action: JsonValue = None results: JsonValue = None - def events(self) -> tuple[TranscriptEvent, ...]: + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: # Search status carries no success flag. Do not invent a successful end. - return (NoticeTranscript(f'Web search: {self.query}', position=self.id),) + return (NoticeTranscript(f'Web search: {self.query}', position=self.id, timestamp=timestamp),) @property def activity_detail(self) -> str: @@ -657,10 +657,10 @@ class FileChange(WireModel): diff: str kind: PatchChange.declared_union() - def events(self, call_id: str, ok: bool) -> tuple[TranscriptEvent, ...]: + def events(self, call_id: str, ok: bool, timestamp: float) -> tuple[TranscriptEvent, ...]: call = ToolCall(call_id, 'edit', {'path': self.path}).recorded(FilePatchTool(path=self.path)) diff = self.kind.patch(self.path, self.diff) - return (call.patched(diff) if diff else call).finished('', ok, call_id) + return (call.patched(diff) if diff else call).finished('', ok, call_id, timestamp) class FileChangeItem(Item): @@ -669,11 +669,11 @@ class FileChangeItem(Item): changes: list[FileChange] status: Literal['inProgress', 'completed', 'failed', 'declined'] - def events(self) -> tuple[TranscriptEvent, ...]: + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: if self.status == 'inProgress': raise ValueError('File change completion still says inProgress') return tuple(event for index, change in enumerate(self.changes) - for event in change.events(f'{self.id}:{index}', self.status == 'completed')) + for event in change.events(f'{self.id}:{index}', self.status == 'completed', timestamp)) class AgentState(WireModel): @@ -688,10 +688,10 @@ class AgentToolItem(Item): receiverThreadIds: list[str] agentsStates: dict[str, AgentState] - def events(self) -> tuple[TranscriptEvent, ...]: + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: return (NoticeTranscript(f'{self.tool}: {self.status}\n' + '\n'.join( f'{name}: {state.status}' + (f' — {state.message}' if state.message else '') - for name, state in self.agentsStates.items()), position=self.id),) + for name, state in self.agentsStates.items()), position=self.id, timestamp=timestamp),) class AgentActivityItem(Item): @@ -700,8 +700,8 @@ class AgentActivityItem(Item): agentThreadId: str kind: str - def events(self) -> tuple[TranscriptEvent, ...]: - return (NoticeTranscript(f'{self.agentPath}: {self.kind}', position=self.id),) + def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: + return (NoticeTranscript(f'{self.agentPath}: {self.kind}', position=self.id, timestamp=timestamp),) # The existing registration owner derives this wire grammar from declarations. @@ -741,8 +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, tuple( - event_time.stamp(event) for event in self.item.events())) + return self.item.completed(self.turnId, self.item.events(event_time.transcript_timestamp())) class ItemsPage(WireModel): diff --git a/src/agent_comms/cli_bridges/codex_rollout.py b/src/agent_comms/cli_bridges/codex_rollout.py index 6de7bc3957..4ae5462930 100644 --- a/src/agent_comms/cli_bridges/codex_rollout.py +++ b/src/agent_comms/cli_bridges/codex_rollout.py @@ -15,7 +15,7 @@ from metaclass_registry import RegistryConfig from pydantic import Field, JsonValue, TypeAdapter, field_validator, model_validator -from agent_comms.cli_protocol import OpenWire, WireHeader, WireModel +from agent_comms.cli_protocol import OpenWire, TimestampedWire, WireHeader, WireModel from agent_comms.tool_declaration import ToolDeclaration from agent_comms.transcript_events import AssistantTranscript, ToolEndTranscript, ToolStartTranscript, UserTranscript from agent_comms.turn_context import CodexRolloutProvenance @@ -50,7 +50,7 @@ class RolloutItem(OpenWire, ABC): registry_dict={}, key_attribute='wire_tag', skip_if_no_key=True) header = RolloutItemHeader - def updates(self, state: RolloutState, position: int) -> tuple[RuntimeUpdate, ...]: + def updates(self, state: RolloutState, position: int, timestamp: float) -> tuple[RuntimeUpdate, ...]: return () def instructions(self) -> tuple[InstructionMessage, ...]: @@ -82,8 +82,8 @@ def text(self) -> str: class UserMessageItem(MessageItem): wire_tag = 'message/user' - def updates(self, state: RolloutState, position: int) -> tuple[RuntimeUpdate, ...]: - return (FinishedItem('', self.id, (UserTranscript(self.text, position=position),)),) + def updates(self, state: RolloutState, position: int, timestamp: float) -> tuple[RuntimeUpdate, ...]: + return (FinishedItem('', self.id, (UserTranscript(self.text, position=position, timestamp=timestamp),)),) def request(self) -> str: return self.text @@ -92,8 +92,8 @@ def request(self) -> str: class AssistantMessageItem(MessageItem): wire_tag = 'message/assistant' - def updates(self, state: RolloutState, position: int) -> tuple[RuntimeUpdate, ...]: - return (FinishedItem('', self.id, (AssistantTranscript(self.text, position=position),)),) + def updates(self, state: RolloutState, position: int, timestamp: float) -> tuple[RuntimeUpdate, ...]: + return (FinishedItem('', self.id, (AssistantTranscript(self.text, position=position, timestamp=timestamp),)),) class InstructionMessage(MessageItem, ABC): @@ -122,10 +122,10 @@ class FunctionCallItem(RolloutItem): def raw_input(self) -> dict[str, JsonValue]: return TypeAdapter(dict[str, JsonValue]).validate_json(self.arguments) - def updates(self, state: RolloutState, position: int) -> tuple[RuntimeUpdate, ...]: + def updates(self, state: RolloutState, position: int, timestamp: float) -> tuple[RuntimeUpdate, ...]: state.tools[self.call_id] = self.name return (FinishedItem('', self.call_id, (ToolStartTranscript( - position=position, tool_call_id=self.call_id, tool_name=self.name, raw_input=self.raw_input, + position=position, timestamp=timestamp, tool_call_id=self.call_id, tool_name=self.name, raw_input=self.raw_input, tool=ToolDeclaration.record(self.name, self.raw_input, call_id=self.call_id)),)),) @@ -144,17 +144,17 @@ class FunctionCallOutputItem(RolloutItem): call_id: str = '' output: JsonValue = '' - def updates(self, state: RolloutState, position: int) -> tuple[RuntimeUpdate, ...]: + def updates(self, state: RolloutState, position: int, timestamp: float) -> tuple[RuntimeUpdate, ...]: text = self.output if isinstance(self.output, str) else json.dumps(self.output, ensure_ascii=False) return (FinishedItem('', self.call_id + ':result', (ToolEndTranscript( - position=position, tool_call_id=self.call_id, tool_name=state.tools.get(self.call_id, ''), text=text),)),) + position=position, timestamp=timestamp, tool_call_id=self.call_id, tool_name=state.tools.get(self.call_id, ''), text=text),)),) class CustomToolCallOutputItem(FunctionCallOutputItem): wire_tag = 'custom_tool_call_output' -class RolloutRecord(OpenWire, ABC): +class RolloutRecord(TimestampedWire, OpenWire, ABC): """One rollout line. Conversation is in the items a record carries; the record adds what it says about the session.""" __registry_config__: ClassVar[RegistryConfig] = RegistryConfig( @@ -167,7 +167,7 @@ def own_updates(self) -> tuple[RuntimeUpdate, ...]: return () def updates(self, state: RolloutState, position: int) -> tuple[RuntimeUpdate, ...]: - return (*self.own_updates(), *(update for item in self.entries() for update in item.updates(state, position))) + return (*self.own_updates(), *(update for item in self.entries() for update in item.updates(state, position, self.timestamp))) def instructions(self) -> tuple[InstructionMessage, ...]: return tuple(instruction for item in self.entries() for instruction in item.instructions()) diff --git a/src/agent_comms/cli_bridges/opencode_parts.py b/src/agent_comms/cli_bridges/opencode_parts.py index 3330ecafbf..03eafd8424 100644 --- a/src/agent_comms/cli_bridges/opencode_parts.py +++ b/src/agent_comms/cli_bridges/opencode_parts.py @@ -70,7 +70,7 @@ def finished(self, part: Part) -> bool: def present(self, part: Part) -> tuple[RuntimeUpdate, ...]: # Wire history owns incoming bus messages; the live view never echoes them as CLI output. - return part.prompted() + return part.prompted(self) class AssistantMessage(MessageInfo): @@ -109,8 +109,8 @@ def finished(self, message: MessageInfo) -> bool: @abstractmethod def present(self, message: AssistantMessage) -> tuple[RuntimeUpdate, ...]: ... - def prompted(self) -> tuple[RuntimeUpdate, ...]: - """This part in a user's message, kept for history only.""" + def prompted(self, message: UserMessage) -> tuple[RuntimeUpdate, ...]: + """This part in a user's recorded message, kept for history only.""" return () def summarized(self) -> tuple[RuntimeUpdate, ...]: @@ -130,9 +130,12 @@ class TextualPart(Part, ABC): activity: ClassVar[ActivityState] detail: ClassVar[str] + def timestamp(self, message: MessageInfo) -> float: + return (self.time.start if self.time is not None else message.time.created) / 1000 + def accumulator(self, message: MessageInfo) -> TextDraft: - started_at = RecordedEventTime((self.time.start if self.time is not None else message.time.created) / 1000) - return TextDraft.from_parts(self.event_type, (self.text,), started_at, self.id) + return TextDraft.from_parts(self.event_type, (self.text,), + RecordedEventTime(self.timestamp(message)), self.id) def finished(self, message: MessageInfo) -> bool: return super().finished(message) or (self.time is not None and self.time.end is not None) @@ -153,8 +156,8 @@ class TextPart(TextualPart): def matches(self, session: str, text: str) -> bool: return self.sessionID == session and self.text == text - def prompted(self) -> tuple[RuntimeUpdate, ...]: - return (HiddenItem((UserTranscript(self.text, position=self.id),)),) + def prompted(self, message: UserMessage) -> tuple[RuntimeUpdate, ...]: + return (HiddenItem((UserTranscript(self.text, position=self.id, timestamp=self.timestamp(message)),)),) def summarized(self) -> tuple[RuntimeUpdate, ...]: return (Summarized(self.text),) @@ -215,8 +218,7 @@ class EndedTool(ToolState, ABC): def result(self) -> str: ... def present(self, part: ToolPart, message: AssistantMessage) -> tuple[RuntimeUpdate, ...]: - event_time = RecordedEventTime(self.time.start / 1000) - events = tuple(event_time.stamp(event) for event in self.call(part).finished(self.result, self.ok, part.id)) + events = self.call(part).finished(self.result, self.ok, part.id, self.time.start / 1000) return (CompletedItem(message.id, part.id, events),) @@ -264,9 +266,10 @@ class FilePart(StructuralPart): filename: str = '' mime: str = '' - def prompted(self) -> tuple[RuntimeUpdate, ...]: + def prompted(self, message: UserMessage) -> tuple[RuntimeUpdate, ...]: return (HistoryNotice('Attachment contents are not imported; source references are retained.'), - HiddenItem((ContextTranscript(f'[Historical attachment: {self.filename or self.mime}]', position=self.id),))) + HiddenItem((ContextTranscript(f'[Historical attachment: {self.filename or self.mime}]', + position=self.id, timestamp=message.time.created / 1000),))) class AgentPart(StructuralPart): diff --git a/src/agent_comms/cli_protocol.py b/src/agent_comms/cli_protocol.py index 9b62b5b419..93aa3650e3 100644 --- a/src/agent_comms/cli_protocol.py +++ b/src/agent_comms/cli_protocol.py @@ -8,12 +8,14 @@ from typing import TYPE_CHECKING, Annotated, AsyncIterator, ClassVar, NoReturn, Union from metaclass_registry import AutoRegisterMeta, RegistryConfig -from pydantic import BaseModel, ConfigDict, Discriminator, InstanceOf, PlainSerializer, PlainValidator, Tag, TypeAdapter, computed_field +from pydantic import BaseModel, ConfigDict, Discriminator, InstanceOf, PlainSerializer, PlainValidator, Tag, TypeAdapter, computed_field, field_validator, model_validator from .coordination_errors import CoordinationReadUnavailable from .declared_family import DeclaredFamily from .maintenance_barrier import MaintenanceHeldError from .messages import Message +from .field_codec import TimestampText +from .recorded_time import MissingTranscriptTimestampError from .external_transcript import RequestedEarlierPage if TYPE_CHECKING: @@ -32,6 +34,24 @@ class WireModel(BaseModel): model_config = ConfigDict(strict=True) +class TimestampedWire(WireModel): + """A source record's original ISO timestamp, decoded before making any events.""" + + timestamp: float + + @model_validator(mode='before') + @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') + return value + + @field_validator('timestamp', mode='before') + @classmethod + def epoch(cls, value: str) -> float: + return TimestampText.from_text(value).at.timestamp() + + class RegisteredWireMeta(AutoRegisterMeta, type(WireModel)): """Compose declaration registration with external-field validation.""" diff --git a/src/agent_comms/native_entries.py b/src/agent_comms/native_entries.py index d28aecd804..dd85871c57 100644 --- a/src/agent_comms/native_entries.py +++ b/src/agent_comms/native_entries.py @@ -69,11 +69,10 @@ def events(self, context: TranscriptProjection) -> list[TranscriptEvent]: Decode it once, before splitting into message/tool/routing events; ACP consumers receive Unix seconds. """ - timestamp = self.timestamp.at.timestamp() - return [replace(event, timestamp=timestamp) for event in self._events(context)] + return self._events(context, self.timestamp.at.timestamp()) - def _events(self, context: TranscriptProjection) -> list[TranscriptEvent]: - return [event for message in self.messages for event in message.transcript_events(context)] + def _events(self, context: TranscriptProjection, timestamp: float) -> list[TranscriptEvent]: + return [event for message in self.messages for event in message.transcript_events(context, timestamp)] @property def original_id(self): @@ -547,9 +546,9 @@ class CompactionEntry(NativeEntry): summary: str = "" - def _events(self, context: TranscriptProjection) -> list[TranscriptEvent]: + def _events(self, context: TranscriptProjection, timestamp: float) -> list[TranscriptEvent]: text = self.summary.strip() - return [NoticeTranscript(f"## Context compacted\n\n{text}", position=context.position)] if text else [] + return [NoticeTranscript(f"## Context compacted\n\n{text}", position=context.position, timestamp=timestamp)] if text else [] @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/pi_payloads.py b/src/agent_comms/pi_payloads.py index 5f7bdbc6f6..0a4f976b3b 100644 --- a/src/agent_comms/pi_payloads.py +++ b/src/agent_comms/pi_payloads.py @@ -277,10 +277,10 @@ def preserve_evidence(self, raw: dict) -> PiContent: """Keep unrepresented native fields opaque, never equivalent to plain text.""" return self if self.to_wire() == raw else UnknownContent(raw) - def user_transcript(self, position: int): + def user_transcript(self, position: int, timestamp: float): return [] - def assistant_transcript(self, position: int): + def assistant_transcript(self, position: int, timestamp: float): return [] wire_tag = "type" @@ -303,15 +303,15 @@ def wire_member(cls, value): @dataclass(frozen=True) class TextContent(PiContent): - def user_transcript(self, position: int): + def user_transcript(self, position: int, timestamp: float): from .transcript_events import UserTranscript - return [UserTranscript(self.text, position=position)] + return [UserTranscript(self.text, position=position, timestamp=timestamp)] - def assistant_transcript(self, position: int): + def assistant_transcript(self, position: int, timestamp: float): from .transcript_events import AssistantTranscript - return [AssistantTranscript(self.text, position=position)] + return [AssistantTranscript(self.text, position=position, timestamp=timestamp)] text: str = field() text_signature: str | None = wire_field("textSignature") @@ -324,10 +324,10 @@ def final_text(self) -> str: @dataclass(frozen=True) class ThinkingContent(PiContent): - def assistant_transcript(self, position: int): + def assistant_transcript(self, position: int, timestamp: float): from .transcript_events import ThinkingTranscript - return [ThinkingTranscript(self.thinking, position=position)] + return [ThinkingTranscript(self.thinking, position=position, timestamp=timestamp)] thinking: str = "" thinking_signature: str | None = wire_field("thinkingSignature") @@ -343,11 +343,11 @@ class ToolCallContent(PiContent, declared_name="toolCall"): def tool_calls(self) -> tuple[ToolCallContent, ...]: return (self,) - def assistant_transcript(self, position: int): + def assistant_transcript(self, position: int, timestamp: float): from .transcript_events import ToolStartTranscript return [ - ToolStartTranscript(position=position, tool_call_id=self.id, tool_name=self.name, raw_input=self.arguments, + ToolStartTranscript(position=position, timestamp=timestamp, tool_call_id=self.id, tool_name=self.name, raw_input=self.arguments, tool=ToolDeclaration.record(self.name, self.arguments, call_id=self.id)) ] @@ -359,10 +359,10 @@ def assistant_transcript(self, position: int): @dataclass(frozen=True) class ImageContent(PiContent): - def user_transcript(self, position: int): + def user_transcript(self, position: int, timestamp: float): from .transcript_events import UserTranscript - return [UserTranscript(f"[Image attachment: {self.mime_type or 'image'}]", position=position)] + return [UserTranscript(f"[Image attachment: {self.mime_type or 'image'}]", position=position, timestamp=timestamp)] data: str mime_type: str = wire_field("mimeType") @@ -540,7 +540,7 @@ def tool_calls(self) -> tuple[ToolCallContent, ...]: def unread_reply(self) -> bool: return False - def transcript_events(self, context): + def transcript_events(self, context, timestamp: float): return [] wire_tag = "role" @@ -676,14 +676,14 @@ def error_notice(self) -> str: def unread_reply(self) -> bool: return bool(self.text.strip() or self.error_notice) - def transcript_events(self, context): + def transcript_events(self, context, timestamp: float): from .transcript_events import AssistantTranscript, NoticeTranscript if notice := self.error_notice: - return [NoticeTranscript(f"[agent error] {notice}", position=context.position)] + return [NoticeTranscript(f"[agent error] {notice}", position=context.position, timestamp=timestamp)] events = [] for part in self.parts: - for event in part.assistant_transcript(context.position): + for event in part.assistant_transcript(context.position, timestamp): if ( isinstance(event, AssistantTranscript) and events @@ -707,7 +707,7 @@ async def apply_start(self, session, event): def observe_start_abort(self, session, event): session.observe_input_during_abort(event) - def transcript_events(self, context): + def transcript_events(self, context, timestamp: float): from .transcript_events import ContextTranscript routing, display = context.routing, context.input_display @@ -722,7 +722,7 @@ def transcript_events(self, context): events = [] if display is not None: if raw_text and raw_text != display.text: - events.append(ContextTranscript(raw_text, position=context.position)) + events.append(ContextTranscript(raw_text, position=context.position, timestamp=timestamp)) if display.text is None: return events parts = ( @@ -730,7 +730,7 @@ def transcript_events(self, context): *(part for part in parts if not isinstance(part, TextContent)), ) for part in parts: - events.extend(part.user_transcript(context.position)) + events.extend(part.user_transcript(context.position, timestamp)) return [replace(event, routing=routing).with_native_input(self.input_id) for event in events] @@ -751,7 +751,7 @@ class ToolResultMessage(ToolDetailsPayload, PiMessage, declared_name="toolResult is_error: bool = wire_field("isError", False) details: NativeToolDetails = field(default_factory=lambda: NoToolDetails()) - def transcript_events(self, context): + def transcript_events(self, context, timestamp: float): from .tool_declaration import ToolDeclaration from .transcript_events import ToolEndTranscript @@ -761,7 +761,7 @@ def transcript_events(self, context): result = ProvidedToolResult(content=self.parts, details=self.details) events = [ ToolEndTranscript( - position=context.position, + position=context.position, timestamp=timestamp, text=output, tool_call_id=self.tool_call_id, tool_name=self.tool_name, diff --git a/src/agent_comms/recorded_time.py b/src/agent_comms/recorded_time.py index a67f23c9a2..35bb91357d 100644 --- a/src/agent_comms/recorded_time.py +++ b/src/agent_comms/recorded_time.py @@ -4,17 +4,15 @@ import time from abc import abstractmethod -from dataclasses import dataclass, replace +from dataclasses import dataclass from math import isfinite -from typing import TYPE_CHECKING, TypeVar + from .declared_family import DeclaredFamily from .field_codec import FieldCodec, FieldRepresentation -if TYPE_CHECKING: - from .transcript_events import TranscriptEvent - -Event = TypeVar("Event", bound="TranscriptEvent") +class MissingTranscriptTimestampError(ValueError): + """An external source cannot publish an event without its recorded time.""" class GeneratedTimestamp(FieldRepresentation): @@ -85,8 +83,8 @@ def require_deadline(self) -> None: def require_delivery(self) -> None: self.require_recorded("A delivery") - @abstractmethod - def stamp(self, event: Event) -> Event: ... + def transcript_timestamp(self) -> float: + raise MissingTranscriptTimestampError("A transcript event requires its source's recorded time") @abstractmethod def continued_from(self, original_at: float): ... @@ -94,9 +92,6 @@ 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) @@ -124,8 +119,8 @@ def clock(self) -> str: def recorded_at(self, label: str) -> float: return self.at - def stamp(self, event: Event) -> Event: - return replace(event, timestamp=self.at) + def transcript_timestamp(self) -> float: + return 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 95b0af385e..b1b9f6bba6 100644 --- a/src/agent_comms/runtime_updates.py +++ b/src/agent_comms/runtime_updates.py @@ -49,9 +49,8 @@ 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 on a full item, preserving provider time when no start was observed.""" - 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) + """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) class NoPrefixDraft(DraftAccumulator): @@ -95,7 +94,8 @@ def events(self) -> tuple[TranscriptEvent, ...]: text for index in sorted(self.sections) if (text := "".join(self.sections[index])) ) return ( - (self.started_at.stamp(self.event_type(text, position=self.position)),) if text else () + (self.event_type(text, position=self.position, + timestamp=self.started_at.transcript_timestamp()),) if text else () ) def snapshot(self, turn_id: str, item_id: str) -> tuple[DraftItem, ...]: diff --git a/src/agent_comms/tool_calls.py b/src/agent_comms/tool_calls.py index 0be436c4c1..f100f5ab16 100644 --- a/src/agent_comms/tool_calls.py +++ b/src/agent_comms/tool_calls.py @@ -35,18 +35,18 @@ class ToolCall: def title(self) -> str: return ToolDeclaration.for_name(self.name).label - def started(self, position: int | str) -> ToolStartTranscript: + def started(self, position: int | str, timestamp: float) -> ToolStartTranscript: return ToolStartTranscript( - position=position, tool_call_id=self.id, tool_name=self.name, raw_input=self.arguments + position=position, timestamp=timestamp, tool_call_id=self.id, tool_name=self.name, raw_input=self.arguments ) - def ended(self, text: str, ok: bool, position: int | str) -> ToolEndTranscript: + def ended(self, text: str, ok: bool, position: int | str, timestamp: float) -> ToolEndTranscript: return ToolEndTranscript( - position=position, tool_call_id=self.id, tool_name=self.name, text=text, ok=ok + position=position, timestamp=timestamp, tool_call_id=self.id, tool_name=self.name, text=text, ok=ok ) - def finished(self, text: str, ok: bool, position: int | str) -> tuple[TranscriptEvent, ...]: - return self.started(position), self.ended(text, ok, position) + def finished(self, text: str, ok: bool, position: int | str, timestamp: float) -> tuple[TranscriptEvent, ...]: + return self.started(position, timestamp), self.ended(text, ok, position, timestamp) def recorded(self, tool: ToolDeclaration) -> DecodedToolCall: return DecodedToolCall(self.id, self.name, self.arguments, tool) @@ -62,8 +62,8 @@ class DecodedToolCall(ToolCall): def title(self) -> str: return ToolDeclaration.caption(self.name, self.tool) - def started(self, position: int | str) -> ToolStartTranscript: - return replace(super().started(position), tool=self.tool) + def started(self, position: int | str, timestamp: float) -> ToolStartTranscript: + return replace(super().started(position, timestamp), tool=self.tool) def patched(self, diff: ToolDiff) -> PatchedToolCall: return PatchedToolCall(self.id, self.name, self.arguments, self.tool, diff) @@ -75,8 +75,8 @@ class PatchedToolCall(DecodedToolCall): diff: ToolDiff - def ended(self, text: str, ok: bool, position: int | str) -> ToolEndTranscript: - return replace(super().ended(text, ok, position), diff=self.diff) + def ended(self, text: str, ok: bool, position: int | str, timestamp: float) -> ToolEndTranscript: + return replace(super().ended(text, ok, position, timestamp), diff=self.diff) class CliToolArguments(WireModel): diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index 2f342e73fe..79fd3dbeb0 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -77,9 +77,9 @@ def pi_message(self, timestamp: int) -> dict[str, object]: class TranscriptEvent(EventMerge, DeclaredFamily, affix="Transcript"): category: ClassVar[type[MessageCategory]] position: int | str + timestamp: float routing = None source = None - timestamp = None def as_context(self) -> TranscriptEvent: """Non-text native evidence retains its own presentation declaration.""" @@ -107,13 +107,8 @@ def record_call(self, calls: dict[str, OrderCall], since: float) -> None: def edit_calls(self, thread: str, problem: str, since: float) -> tuple[tool_declaration.EditCall, ...]: return () - @property - def dated(self) -> tuple[float, ...]: - return () if self.timestamp is None else (self.timestamp,) - def before(self, at: float) -> bool: - """Recorded before ``at``; an undated event is never before anything.""" - return any(time < at for time in self.dated) + return self.timestamp < at @dataclass(frozen=True) @@ -136,7 +131,6 @@ def native_inputs(self) -> frozenset[str]: class RecordedTranscript(TranscriptEvent): routing: TurnRouting | None = None source: MessageReference | None = None - timestamp: float | None = None class TextTranscript(TranscriptEvent): @@ -207,6 +201,7 @@ def category(self) -> type[MessageCategory]: message: Message reader: str position: ClassVar[int] + timestamp: ClassVar[float] @property def position(self) -> int: @@ -365,7 +360,7 @@ def title(self) -> str: return tool_declaration.ToolDeclaration.caption(self.tool_name, self.tool) def record_call(self, calls: dict[str, OrderCall], since: float) -> None: - """A call recorded before ``since`` is not work since then; an undated one is its turn's.""" + """A call recorded before ``since`` is not work since then.""" if not self.before(since): digest = tool_declaration.ToolDeclaration.call_digest( self.tool_name, self.tool, self.raw_input) @@ -441,6 +436,5 @@ class TurnDigest: events: tuple[TranscriptEvent, ...] def before(self, at: float) -> bool: - """The turn's last dated event precedes ``at``; an undated turn is never before.""" - dated = [event for event in self.events if event.dated] - return bool(dated) and dated[-1].before(at) + """The turn's last recorded event precedes ``at``; an empty turn is never before.""" + return bool(self.events) and self.events[-1].before(at) diff --git a/src/agent_comms/transcripts.py b/src/agent_comms/transcripts.py index 58f3eed7a9..fe08de2595 100644 --- a/src/agent_comms/transcripts.py +++ b/src/agent_comms/transcripts.py @@ -563,8 +563,9 @@ def _fork_start_events(thread: Thread) -> tuple[TranscriptEvent, ...]: if not thread.parent: return () return ( - ForkOriginTranscript(f"Forked from @{thread.parent}. This thread started with:"), - *((ForkTaskTranscript(thread.task),) if thread.task else ()), + ForkOriginTranscript(f"Forked from @{thread.parent}. This thread started with:", + timestamp=thread.created_at), + *((ForkTaskTranscript(thread.task, timestamp=thread.created_at),) if thread.task else ()), ) def thread_transcript_page( From bc1822b47b0d24e891fa708e3615662d4346d414 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 18:00:32 -0400 Subject: [PATCH 4/4] Keep send_guard as on main; P260 owns the one shell parse Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/commander/send_guard.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/src/agent_comms/commander/send_guard.py b/src/agent_comms/commander/send_guard.py index d302f042d4..642e93b2e8 100644 --- a/src/agent_comms/commander/send_guard.py +++ b/src/agent_comms/commander/send_guard.py @@ -35,14 +35,11 @@ def simple_commands(command: str) -> tuple[Words, ...]: lexer = shlex.shlex(command, posix=True, punctuation_chars=True) lexer.whitespace_split = True commands: list[list[str]] = [[]] - try: - for word in lexer: - if word and set(word) <= set(';&|()'): - commands.append([]) - else: - commands[-1].append(word) - except ValueError: - return (tuple(command.split()),) + for word in lexer: + if word and set(word) <= set(';&|()'): + commands.append([]) + else: + commands[-1].append(word) return tuple(tuple(words) for words in commands if words)