Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 12 additions & 17 deletions src/agent_comms/cli_bridges/claude_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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,
Expand Down Expand Up @@ -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 = ''

Expand All @@ -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):
Expand Down Expand Up @@ -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)


Expand Down Expand Up @@ -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):
Expand Down
71 changes: 35 additions & 36 deletions src/agent_comms/cli_bridges/codex_protocol.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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):
Expand All @@ -335,17 +335,17 @@ 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):
wire_tag = 'userMessage'
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.
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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 ()

Expand All @@ -545,23 +545,23 @@ 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 ()


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):
Expand All @@ -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:
Expand All @@ -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):
Expand All @@ -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):
Expand All @@ -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:
Expand Down Expand Up @@ -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):
Expand All @@ -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):
Expand All @@ -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):
Expand All @@ -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.
Expand Down Expand Up @@ -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):
Expand Down
24 changes: 12 additions & 12 deletions src/agent_comms/cli_bridges/codex_rollout.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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, ...]:
Expand Down Expand Up @@ -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
Expand All @@ -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):
Expand Down Expand Up @@ -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)),)),)


Expand All @@ -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(
Expand All @@ -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())
Expand Down
Loading
Loading