diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 57efe574de..52578363a1 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -16,6 +16,7 @@ from agent_comms.bus_activity_index import BusActivityIndex from agent_comms.bus_display_index import BusDisplayIndex from agent_comms.bus_publication import Delivered, KeyedPublication +from agent_comms.channel_targets import NamedTarget from agent_comms.bus_route_counts import RouteCounts from agent_comms.bus_tables import ( AnsweredSource, @@ -53,7 +54,7 @@ from agent_comms.private_path import FileIdentity, PresentPath from agent_comms.read_basis import CursorRow from agent_comms.read_basis import ( - DisplayUnreadSelection, MessageDisplayScope, ReadCursor, RepliesDisplayScope, UnreadQuery, + DisplayUnreadSelection, ReadCursor, RepliesDisplayScope, UnreadQuery, ) from agent_comms.typed_table import ( CarriedWriter, PracticeRebuild, ReconciledFile, SchemaRebuild, @@ -63,8 +64,6 @@ if TYPE_CHECKING: from agent_comms.registry_document import RegistrySnapshot - from agent_comms.registry_provenance import RegistryProvenance - from agent_comms.thread_identity import RecordedThreadIncarnation class ClaimRecord(PublicRecord, ClaimMessage): @@ -162,6 +161,12 @@ class BusReader(RouteCounts, ABC): def observing(self) -> AbstractContextManager[sqlite3.Connection]: """Borrow one read snapshot through the reader's declared lifetime.""" + def answering(self, references: tuple[MessageReference, ...], route: NamedTarget, + snapshot: RegistrySnapshot) -> tuple[MessageReference, ...]: + """Select answer links on this reader's own connection.""" + with self.observing() as connection: + return MessageIdentitySource.answering(connection, references, route, snapshot) + @contextmanager def reading(self) -> Iterator[BusSnapshot]: """Keep queries and their physical identity on one held snapshot.""" @@ -472,27 +477,6 @@ def count_unread( ).fetchone()[0]) return counts - def delivered_unanswered( - self, recipient: RecordedThreadIncarnation, scope: MessageDisplayScope, - seen: ReadCursor, snapshot: RegistryProvenance, - ) -> tuple[MessageReference, ...]: - """Unanswered originals in this recipient's recorded delivered cursor.""" - bindings = ReadCursor.sql_bindings(snapshot, {"viewer": seen.counted(scope, snapshot)}) - predicate, parameters = scope.index_predicate - received, recipients = ReceivedSource.query(recipient.lookup) - with self.observing() as db: - rows = db.execute( - f'WITH {ReadCursor.sql("?")}, requests AS (' - f'SELECT \'viewer\' AS actor,* FROM "{Addressed.table.declared_name}" ' - f'JOIN "{MessageInput.table.declared_name}" USING(seq) ' - f'WHERE seq > 0 AND ({predicate}) AND seq IN ({received})), ' - f'{ReadCursor.joined_sql} ' - f'SELECT DISTINCT seq FROM cursors WHERE {ReadCursor.read_sql}', - (*bindings.values(), *parameters, *recipients), - ) - sequences = tuple(FieldCodec.decode(int, row[0]) for row in rows) - return self.unanswered_references(sequences, snapshot.names_for(recipient.name)) - def selected_sequences( self, predicate: str, parameters: tuple[object, ...], through: int, ) -> list[AddressedRow]: diff --git a/src/agent_comms/cli_bridges/claude_protocol.py b/src/agent_comms/cli_bridges/claude_protocol.py index ad3406f693..00b763dc3b 100644 --- a/src/agent_comms/cli_bridges/claude_protocol.py +++ b/src/agent_comms/cli_bridges/claude_protocol.py @@ -38,7 +38,7 @@ from agent_comms.recorded_time import EventTime from agent_comms.runtime_updates import ( ActiveItem, - AllTurnsEnded, + AllTurnsCancelled, AllTurnsEnded, FinishedItem, HiddenItem, HistoryNotice, @@ -59,11 +59,13 @@ from agent_comms.tool_calls import CliToolArguments, ToolCall from agent_comms.transcript_events import ( AssistantTranscript, + NarrationTranscript, + ThinkingTranscript, NoticeTranscript, TranscriptEvent, UserTranscript, ) -from agent_comms.turn_item import NativeItemId, NativeTurnId +from agent_comms.turn_item import NativeItemId, NativeTurnId, NamedSourceTurn, NoSourceTurn, SourceTurn from agent_comms.turn_phase import ModelWaitPhase, ToolRunningPhase, TurnPhase from ..wire_models import WireModel @@ -244,17 +246,22 @@ def completed( return (*self.events, self.call.ended(text, ok, record.position, record.timestamp)) def read_results(self, record: TimedRecord, text: str, ok: bool, - reading: frozenset[str]) -> tuple[RuntimeUpdate, ...]: + reading: frozenset[str], source_turn: SourceTurn) -> tuple[RuntimeUpdate, ...]: """What a Read showed the model, timed by its own result record, beside the paths other Reads in this session still have open. A failed Read showed no file but ends its call.""" observed = ReadResultObserved if ok else FailedReadObserved - return tuple(observed(path, text, ObservedMonitor(record.uuid, record.timestamp), reading) + return tuple(observed(path, text, ObservedMonitor(record.uuid, record.timestamp), reading, + source_turn=source_turn) for path in self.call.context_read_paths()) class ClaudeTurn(ABC): """The transcript's identified turn owns item identities.""" + @property + def source(self) -> SourceTurn: + return NoSourceTurn() + @abstractmethod def started(self, tool: ClaudeToolDraft, phase: ClaudeReaderPhase) -> tuple[RuntimeUpdate, ...]: ... @@ -286,6 +293,10 @@ def active(self, tool_id: str, phase: ClaudeReaderPhase, class KnownTurn(ClaudeTurn): id: NativeTurnId + @property + def source(self) -> SourceTurn: + return NamedSourceTurn(self.id) + def started(self, tool: ClaudeToolDraft, phase: ClaudeReaderPhase) -> tuple[RuntimeUpdate, ...]: return (StartedItem(self.id, NativeItemId(tool.call.id), @@ -350,7 +361,7 @@ def tool_result( *self.turn.active(tool_id, self.phase, ModelWaitPhase()), *tool.read_results(record, text, ok, frozenset( path for other in self.tools.values() - for path in other.call.context_read_paths()))) + for path in other.call.context_read_paths()), self.turn.source)) def finished(self, item: str, events: tuple[TranscriptEvent, ...]) -> tuple[RuntimeUpdate, ...]: return self.turn.finished(item, events, self.phase) @@ -412,10 +423,13 @@ def owner_words(self) -> tuple[str, ...]: return () -class AssistantBlock(TaggedWire, ABC, affix=""): +class AssistantBlock(TaggedWire, ABC, affix="", declared_name=None): - @abstractmethod - def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: ... + def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: + return () + + def events(self, record: TimedRecord) -> tuple[TranscriptEvent, ...]: + return () class TextContent(WireModel): @@ -432,8 +446,8 @@ def result_text(self) -> str: class AssistantText(TextWire, AssistantBlock, TextContent, declared_name='text'): - def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return session.finished(record.uuid, (AssistantTranscript(self.text, timestamp=record.timestamp, position=record.position),)) + def events(self, record: TimedRecord) -> tuple[TranscriptEvent, ...]: + return (AssistantTranscript(self.text, timestamp=record.timestamp, position=record.position),) class RedactedThinking(AssistantBlock, declared_name='redacted_thinking'): @@ -441,6 +455,9 @@ class RedactedThinking(AssistantBlock, declared_name='redacted_thinking'): def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return session.turn.active(record.uuid, session.phase, ModelWaitPhase()) + def events(self, record: TimedRecord) -> tuple[TranscriptEvent, ...]: + return (ThinkingTranscript('', timestamp=record.timestamp, position=record.position),) + class NarrationBlock(RedactedThinking, declared_name='thinking'): """Claude Code saves its reasoning under this tag with empty text, and the @@ -449,12 +466,8 @@ class NarrationBlock(RedactedThinking, declared_name='thinking'): thinking: str - def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - activity = super().updates(record, session) - if not self.thinking: - return activity - return (*activity, *session.finished( - record.uuid, (AssistantTranscript(self.thinking, timestamp=record.timestamp, position=record.position),))) + def events(self, record: TimedRecord) -> tuple[TranscriptEvent, ...]: + return (NarrationTranscript(self.thinking, timestamp=record.timestamp, position=record.position),) class ToolUse(AssistantBlock, declared_name='tool_use'): @@ -497,8 +510,11 @@ class AssistantRecord(ClaudeRecord, TimedRecord, declared_name='assistant'): message: AssistantMessage def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return (*(update for block in self.message.content for update in block.updates(self, session)), - *session.answered(self.message)) + updates = [update for block in self.message.content for update in block.updates(self, session)] + events = tuple(event for block in self.message.content for event in block.events(self)) + if events: + updates.extend(session.finished(self.uuid, events)) + return (*updates, *session.answered(self.message)) class ResultPart(TaggedWire, ABC, affix=""): @@ -590,7 +606,7 @@ class TaskNotification(PromptOrigin, declared_name='task-notification'): def absorbed(self, record: TimedRecord, text: str, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return TaskEnvelope.from_text(text).updates(record) + return TaskEnvelope.from_text(text).updates(record, session.turn.source) class PeerMessage(AgentPrompt, declared_name='peer'): @@ -631,7 +647,7 @@ def prompt_text(self) -> str: def updates(self, record: UserRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: if self.text.startswith('[Request interrupted by user'): - return (AllTurnsEnded(),) + return (AllTurnsCancelled(),) return record.origin.prompt(record, self.text, session) @@ -662,13 +678,13 @@ def from_text(cls, text: str) -> TaskEnvelope: return MonitorTaskNotification(frames=ClaudeMonitorFrame.decode_event(''.join(event.itertext()))) @abstractmethod - def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: ... + def updates(self, record: TimedRecord, source_turn: SourceTurn) -> tuple[RuntimeUpdate, ...]: ... class ForeignTaskNotification(TaskEnvelope): text: str - def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: + def updates(self, record: TimedRecord, source_turn: SourceTurn) -> tuple[RuntimeUpdate, ...]: return () @@ -710,8 +726,8 @@ def frames(self) -> tuple[MonitorFrame, ...]: lines.pop() return tuple(ClaudeMonitorFrame.decode_line(line) for line in lines) - async def confirmed_by(self, bridge: LineMonitorBridge) -> RuntimeUpdate: - return replace(self, targets=await bridge.read_export(self)) + async def confirmed_by(self, bridge: LineMonitorBridge) -> ConfirmedDelivery: + return replace(self, references=await bridge.read_export(self)) @dataclass(frozen=True) @@ -726,9 +742,9 @@ def frames(self) -> tuple[MonitorFrame, ...]: class MonitorTaskNotification(TaskEnvelope): frames: tuple[MonitorFrame, ...] - def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: - return tuple(MonitorObservation(frame, - ObservedMonitor(record.uuid, record.timestamp)) + def updates(self, record: TimedRecord, source_turn: SourceTurn) -> tuple[RuntimeUpdate, ...]: + return tuple(MonitorObservation(frame, ObservedMonitor(record.uuid, record.timestamp), + source_turn=source_turn) for frame in self.frames) @@ -741,8 +757,8 @@ class TaskContent(WireModel): def envelope(cls, value: str) -> TaskEnvelope: return TaskEnvelope.from_text(value) - def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: - return self.text.updates(record) + def updates(self, record: TimedRecord, source_turn: SourceTurn) -> tuple[RuntimeUpdate, ...]: + return self.text.updates(record, source_turn) class TaskText(TaskContent): @@ -752,8 +768,8 @@ class TaskText(TaskContent): class TaskMessage(ContentBlocks): content: list[TaskText] - def updates(self, record: TimedRecord) -> tuple[RuntimeUpdate, ...]: - return tuple(update for block in self.content for update in block.updates(record)) + def updates(self, record: TimedRecord, source_turn: SourceTurn) -> tuple[RuntimeUpdate, ...]: + return tuple(update for block in self.content for update in block.updates(record, source_turn)) class ToolResult(UserBlock, ContentBlocks, declared_name='tool_result'): @@ -774,12 +790,23 @@ def text(self) -> str: return '\n'.join(text for block in self.content if (text := block.prompt_text())) -class UserRecord(ClaudeRecord, TimedRecord, Originated, declared_name='user'): - promptId: str - message: UserMessage +class PromptRecord(ClaudeRecord, TimedRecord, ABC, declared_name=None): + """Every prompt shape opens its own source turn at the wire boundary.""" + + promptId: NativeTurnId def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - session.turn = KnownTurn(FieldCodec.decode(NativeTurnId, self.promptId)) + session.turn = KnownTurn(self.promptId) + return self.opened_updates(session) + + @abstractmethod + def opened_updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: ... + + +class UserRecord(PromptRecord, Originated, declared_name='user'): + message: UserMessage + + def opened_updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return tuple(update for block in self.message.content for update in block.updates(self, session)) def request(self) -> str: @@ -792,20 +819,19 @@ def owner_words(self) -> tuple[str, ...]: class CompactSummary(UserRecord, declared_name='compact-summary'): """Claude's summary of the conversation before its latest compaction.""" - def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - session.turn = KnownTurn(FieldCodec.decode(NativeTurnId, self.promptId)) + def opened_updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: return (TurnStarted(), Summarized(self.message.text())) def request(self) -> str: return '' -class TaskUserRecord(ClaudeRecord, TimedRecord, declared_name='task-user'): +class TaskUserRecord(PromptRecord, declared_name='task-user'): origin: TaskNotification message: TaskMessage - def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - return self.message.updates(self) + def opened_updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: + return self.message.updates(self, session.turn.source) class SubtypeHeader(WireHeader): diff --git a/src/agent_comms/cli_bridges/codex_protocol.py b/src/agent_comms/cli_bridges/codex_protocol.py index cc0b293011..cec34a037f 100644 --- a/src/agent_comms/cli_bridges/codex_protocol.py +++ b/src/agent_comms/cli_bridges/codex_protocol.py @@ -8,6 +8,7 @@ import json import time from abc import ABC, abstractmethod +from dataclasses import dataclass from typing import TYPE_CHECKING, Annotated, ClassVar, Generic, Literal, TypeVar from pydantic import ( @@ -20,7 +21,7 @@ ) from agent_comms.cli_protocol import Declared, TaggedWire, TextWire -from agent_comms.declared_family import UnknownFamilyMember +from agent_comms.declared_family import DeclaredFamily, UnknownFamilyMember from agent_comms.external_transcript import EarlierPage, NoEarlierPage from agent_comms.field_codec import ( FieldCodec, @@ -58,10 +59,11 @@ TranscriptEvent, UserTranscript, ) -from agent_comms.turn_item import NativeItemId, NativeTurnId +from agent_comms.turn_item import NativeItemId, NativeTurnId, NamedSourceTurn from agent_comms.turn_phase import CompactionPhase, ModelWaitPhase, ToolRunningPhase, TurnPhase from ..wire_models import DeclaredWireFamily, WireModel +from .shared import BridgeInputObserved class UserInput(TaggedWire, ABC, affix=""): @@ -187,17 +189,63 @@ class MessageInput(TurnInput): input: tuple[TextInput, ...] = Field(min_length=1) -class BusInput(MessageInput): - client_prefix: ClassVar[str] = 'comms:' +class ClientId(DeclaredFamily, ABC): + """Codex's optional wire key is decoded once, before item behavior is selected.""" @classmethod - def for_message(cls, thread: str, message_id: MessageId, text: str) -> BusInput: - return cls(threadId=thread, clientUserMessageId=cls.client_prefix + message_id, - input=(TextInput(text=text),)) + @boundary_decoder + def decode(cls, value: str | None) -> ClientId: + if value is None: + return NoClientId() + if value.startswith(BusClientId.prefix): + return BusClientId(FieldCodec.decode(MessageId, value[len(BusClientId.prefix):])) + return ForeignClientId(value) + + @property + @abstractmethod + def external(self) -> str | None: ... + + def completed(self, item: UserItem, turn_id: NativeTurnId, + events: tuple[TranscriptEvent, ...]) -> RuntimeUpdate: + return CompletedItem(turn_id, item.id, events) + + +@dataclass(frozen=True) +class NoClientId(ClientId): + @property + def external(self) -> None: + return None + + +@dataclass(frozen=True) +class ForeignClientId(ClientId): + text: str + @property + def external(self) -> str: + return self.text + + +@dataclass(frozen=True) +class BusClientId(ClientId): + message_id: MessageId + prefix: ClassVar[str] = 'comms:' + + @property + def external(self) -> str: + return self.prefix + self.message_id + + def completed(self, item: UserItem, turn_id: NativeTurnId, + events: tuple[TranscriptEvent, ...]) -> RuntimeUpdate: + return BridgeInputObserved(events, self.message_id, item.input_text, + source_turn=NamedSourceTurn(turn_id)) + + +class BusInput(MessageInput): @classmethod - def identifies(cls, client_id: str | None) -> bool: - return client_id is not None and client_id.startswith(cls.client_prefix) + def for_message(cls, thread: str, message_id: MessageId, text: str) -> BusInput: + return cls(threadId=thread, clientUserMessageId=BusClientId(message_id).external, + input=(TextInput(text=text),)) SEND_TOOL = 'comms_send' @@ -428,18 +476,22 @@ def accumulator(self, started_at: EventTime) -> ParagraphDraft: class UserItem(Item): wire_tag = 'userMessage' content: list[Declared[UserInput]] - clientId: str | None = None + clientId: Annotated[ + ClientId, ValidateAs(str | None, ClientId.decode), + PlainSerializer(lambda identity: identity.external), + ] = Field(default_factory=NoClientId) + + @property + def input_text(self) -> str: + return '\n'.join(part.display for part in self.content) def events(self, timestamp: float) -> tuple[TranscriptEvent, ...]: - return (UserTranscript('\n'.join(part.display for part in self.content), position=self.id, timestamp=timestamp),) + return (UserTranscript(self.input_text, position=self.id, timestamp=timestamp),) def completed( self, turn_id: NativeTurnId, events: tuple[TranscriptEvent, ...] ) -> RuntimeUpdate: - # Bus inputs already appear in Core's inbox; the live view does not echo them as new input. - if BusInput.identifies(self.clientId): - return HiddenItem(events) - return super().completed(turn_id, events) + return self.clientId.completed(self, turn_id, events) class ToolItem(ToolCallItem, Item, ABC): diff --git a/src/agent_comms/cli_bridges/shared.py b/src/agent_comms/cli_bridges/shared.py index ad1b83fa99..52d00b6020 100644 --- a/src/agent_comms/cli_bridges/shared.py +++ b/src/agent_comms/cli_bridges/shared.py @@ -42,7 +42,7 @@ from agent_comms.cli_arguments import CliArguments from agent_comms.cli_protocol import TaggedWire, WatchLine, raise_child_exit from agent_comms.harness_delivery_states import ( - EmittedOutcome, FilteredOutcome, MismatchedMonitor, ObservedMonitor, + DeliveredOutcome, EmittedOutcome, FilteredOutcome, MismatchedMonitor, ObservedMonitor, Outcome, ReceivingMonitor, UnconfirmedOutcome, ) from agent_comms.comms import Comms, wire @@ -67,7 +67,8 @@ MessageReference, ) from agent_comms.registration import Registration -from agent_comms.runtime_updates import ReplyingThreadOutput, RuntimeUpdate, ThreadOutput +from agent_comms.runtime_updates import HiddenItem, ReplyingThreadOutput, RuntimeUpdate, ThreadOutput +from agent_comms.turn_item import NoSourceTurn, SourceTurn from agent_comms.send_result import SendResult from agent_comms.thread_execution import ExternalThreadExecution from agent_comms.thread_identity import ThreadIncarnation @@ -1278,7 +1279,7 @@ class TranscriptBridge(PublishesTranscript, CliBridge[BridgeOptions]): output_type: ClassVar[type[ThreadOutput]] = ThreadOutput - async def confirm_delivery(self, update: ConfirmedDelivery) -> RuntimeUpdate: + async def confirm_delivery(self, update: ConfirmedDelivery) -> ConfirmedDelivery: """A transcript this bridge did not deliver into confirms nothing.""" return update @@ -1314,10 +1315,10 @@ async def read() -> None: class MonitorFrame(TaggedWire, ABC, affix=""): """Our line protocol inside the harness's task-notification envelope.""" - @abstractmethod async def confirm( self, bridge: LineMonitorBridge, record: ObservedMonitor, - ) -> tuple[str, ...]: ... + ) -> tuple[MessageReference, ...]: + return () def exported_part(self, message_id: MessageId) -> tuple[MonitorPart, ...]: """Only a body part of that message counts when read back from its export.""" @@ -1338,7 +1339,7 @@ class MonitorProbe(MonitorProtocolFrame, declared_name='probe'): session: str token: str - async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[str, ...]: + async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[MessageReference, ...]: await bridge.confirm_probe(self) return () @@ -1416,7 +1417,7 @@ def retire(self) -> None: with self.directory() as directory, suppress(FileNotFoundError): os.unlink(Path(self.path).name, dir_fd=directory) - async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[str, ...]: + async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[MessageReference, ...]: """The notice frees only its own reservation; it confirms no body.""" await bridge.part_observed(self.reference.message_id, self.reservation) return () @@ -1435,10 +1436,6 @@ def announce(cls, claim: WatchClaim) -> None: 'agent-comms-claude-bridge connect restarts it and this Monitor resumes.')) stopped.emit() - async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[str, ...]: - """A notice carries no delivery to confirm.""" - return () - class MonitorBridgeSwapping(MonitorProtocolFrame, declared_name='bridge-swapping'): notice: str @@ -1450,10 +1447,6 @@ def announce(cls, claim: WatchClaim) -> None: 'an install holds its claim and is replacing it.' )).emit() - async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[str, ...]: - """A notice carries no delivery to confirm.""" - return () - class MonitorPart(MonitorFrame, declared_name='part'): id: InstanceOf[ @@ -1477,43 +1470,67 @@ def valid_number(self): def exported_part(self, message_id: MessageId) -> tuple[MonitorPart, ...]: return (self,) if self.id == message_id else () - async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[str, ...]: + async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[MessageReference, ...]: """The whole confirmation (attempt read, observed parts, mismatch, final outcome) is one contention boundary: a busy coordinator cannot tear down the monitor mid-confirmation.""" - targets: list[str] = [] + references: list[MessageReference] = [] with bridge.uncertain_on_contention(self.id): for delivery in await bridge.attempts(self.id): - targets.extend(await delivery.monitor.confirm(bridge, delivery, self, record)) - return tuple(targets) + references.extend(await delivery.monitor.confirm(bridge, delivery, self, record)) + return tuple(references) class ForeignMonitorFrame(MonitorFrame, declared_name='foreign'): text: str - async def confirm(self, bridge: LineMonitorBridge, record: ObservedMonitor) -> tuple[str, ...]: - """Another monitor's output is not comms delivery evidence.""" - return () - MONITOR_FRAME = WireModel.adapter(MonitorFrame.declared_union()) @dataclass(frozen=True) -class ConfirmedDelivery(RuntimeUpdate): - """A transcript observation that may confirm deliveries.""" +class DeliveredInput(RuntimeUpdate): + """The observed original references, bound by their source turn.""" - targets: tuple[str, ...] = field(default=(), kw_only=True) + references: tuple[MessageReference, ...] = field(default=(), kw_only=True) + source_turn: SourceTurn = field(default_factory=NoSourceTurn, kw_only=True) def publish(self, output: ThreadOutput) -> None: - """Keep confirmed reply routes in transcript order, without a second input row.""" - output.delivered_targets(self.targets) + output.delivered(self.source_turn.require_id(), self.references) + + async def observed(self, bridge: TranscriptBridge) -> DeliveredInput: + return self + + +class ConfirmedDelivery(DeliveredInput): + """An observation confirmed through the session's monitor protocol.""" - async def observed(self, bridge: TranscriptBridge) -> RuntimeUpdate: + async def observed(self, bridge: TranscriptBridge) -> ConfirmedDelivery: return await bridge.confirm_delivery(self) @abstractmethod - async def confirmed_by(self, bridge: LineMonitorBridge) -> RuntimeUpdate: - """This observation with the reply targets its delivery confirms through that bridge.""" + async def confirmed_by(self, bridge: LineMonitorBridge) -> ConfirmedDelivery: + """This observation with the original references its delivery confirms.""" + + +@dataclass(frozen=True) +class BridgeInputObserved(DeliveredInput, HiddenItem): + """A harness echoes our exact client input in its own ordered transcript.""" + + message_id: MessageId + text: str + + async def observed(self, bridge: TranscriptBridge) -> DeliveredInput: + attempts = await bridge.attempts(self.message_id) + if not attempts: + raise RelationViolationError("Observed bus input has no original delivery attempt") + references: list[MessageReference] = [] + for delivery in attempts: + message = await bridge.message(delivery.reference) + if self.text != IncomingFrame(message).text: + raise RelationViolationError("Observed bus input differs from its original delivery") + await bridge.finish(delivery, partial(replace, outcome=DeliveredOutcome)) + references.extend(message.opening_references(bridge.binding.bus)) + return replace(self, references=tuple(references)) @dataclass(frozen=True) @@ -1521,8 +1538,8 @@ class MonitorObservation(ConfirmedDelivery): frame: MonitorFrame record: ObservedMonitor - async def confirmed_by(self, bridge: LineMonitorBridge) -> RuntimeUpdate: - return replace(self, targets=await self.frame.confirm(bridge, self.record)) + async def confirmed_by(self, bridge: LineMonitorBridge) -> ConfirmedDelivery: + return replace(self, references=await self.frame.confirm(bridge, self.record)) class TranscriptConfirmsDelivery: @@ -1620,17 +1637,17 @@ def exports(self) -> Path: """Where this session's messages too large for one event are read whole.""" return self.options.state.absolute() / 'messages' - async def confirm_delivery(self, update: ConfirmedDelivery) -> RuntimeUpdate: + async def confirm_delivery(self, update: ConfirmedDelivery) -> ConfirmedDelivery: return await update.confirmed_by(self) - async def read_export(self, read: ReadResultObserved) -> tuple[str, ...]: + async def read_export(self, read: ReadResultObserved) -> tuple[MessageReference, ...]: """A visible Read of this session's own export confirms the parts it shows, through the same all-parts gate a Monitor event uses. The export names an exact reference: only the attempt at that reference takes its parts, so a file at another sequence with the same message id confirms nothing. Any other file, or a read whose every line is not a part of the exported message, confirms nothing either. The export goes once that attempt, reread, is consumed and no other Read of it is still running.""" - targets: list[str] = [] + references: list[MessageReference] = [] for notice in MonitorReadFile.exported_at(self.exports, read.path): identity = notice.reference.message_id frames = read.frames() @@ -1642,11 +1659,11 @@ async def read_export(self, read: ReadResultObserved) -> tuple[str, ...]: for exact in (attempt for attempt in attempted if attempt.reference == notice.reference): for part in parts: - targets.extend(await part.confirm(self, read.record)) + references.extend(await part.confirm(self, read.record)) for unread in read.closing(notice): for stored in await self.coordinator_read(exact.stored): stored.outcome.finished_with(unread) - return tuple(targets) + return tuple(references) def new_attachment(self, peer: MonitorPeer, token: str) -> MonitorAttachment: return MonitorAttachment(peer, MonitorProbe(session=self.binding.thread, token=token)) diff --git a/src/agent_comms/external_transcript.py b/src/agent_comms/external_transcript.py index d3b5e5e2e6..c3fcae2269 100644 --- a/src/agent_comms/external_transcript.py +++ b/src/agent_comms/external_transcript.py @@ -10,7 +10,7 @@ from __future__ import annotations -from collections.abc import Generator, Iterable, Iterator +from collections.abc import Callable, Generator, Iterable, Iterator from contextlib import aclosing, closing, contextmanager from dataclasses import dataclass, field from pathlib import Path @@ -18,6 +18,7 @@ from typing import TYPE_CHECKING, Annotated, ClassVar from .declared_family import DeclaredFamily +from .errors import RelationViolationError from .field_codec import ( FieldCodec, InputId, @@ -26,6 +27,7 @@ ToolCallId, ) from .locked_store import RegistryFile +from .message_reference import MessageReference from .native_runtime_input import NativeRuntimeInput from .native_tools import MutatingPathTool from .native_transcript import RecordsCut @@ -58,9 +60,81 @@ from .turn_digests import TurnBound +class ReplyWindow(DeclaredFamily, affix="Replies"): + """The turn owns its opening deliveries, before and after output starts.""" + + def delivered(self, connection: Connection, turn: ExternalTurn, + references: tuple[MessageReference, ...]) -> None: + raise RelationViolationError( + f"Stopped source turn {turn.turn_id} cannot accept opening references {references!r}") + + def answering(self) -> ReplyWindow: + return self + + def references_in(self, connection: Connection, turn: ExternalTurn) -> tuple[MessageReference, ...]: + return () + + def answered(self, connection: Connection, turn: ExternalTurn) -> tuple[MessageReference, ...]: + return () + + def observed(self, events: tuple[TranscriptEvent, ...]) -> ReplyWindow: + window = self + for event in events: + window = event.reply_window(window, window.answering) + return window + + def finished(self) -> ReplyWindow: + return self + + def cancelled(self) -> ReplyWindow: + return self + + +class ActiveReplies(ReplyWindow, declared_name=None): + """Only active windows admit context or transition on turn retirement.""" + + def delivered(self, connection: Connection, turn: ExternalTurn, + references: tuple[MessageReference, ...]) -> None: + """Later context is observed, but cannot reopen the answer's input window.""" + + def references_in(self, connection: Connection, turn: ExternalTurn) -> tuple[MessageReference, ...]: + return turn.opening_references(connection) + + def finished(self) -> ReplyWindow: + return FinishedReplies() + + def cancelled(self) -> ReplyWindow: + return StoppedReplies() + + @dataclass(frozen=True) -class ExternalTurn(TypedRow): - turn_id: NativeTurnId +class OpeningReplies(ActiveReplies): + def delivered(self, connection: Connection, turn: ExternalTurn, + references: tuple[MessageReference, ...]) -> None: + for reference in references: + ExternalTurnInput(turn.thread, turn.turn_id, reference).insert_if_absent(connection) + + def answering(self) -> ReplyWindow: + return AnsweringReplies() + + +@dataclass(frozen=True) +class AnsweringReplies(ActiveReplies): + """Later context cannot change which deliveries opened this answer.""" + + +@dataclass(frozen=True) +class StoppedReplies(ReplyWindow): + """A retired window inherits refusal and cannot be reopened by output.""" + + +class FinishedReplies(StoppedReplies): + def answered(self, connection: Connection, turn: ExternalTurn) -> tuple[MessageReference, ...]: + opening = turn.opening_references(connection) + return opening if any(event.answers_turn + for item in ExternalItemRow.select(connection, thread=turn.thread, turn_id=turn.turn_id, + order_by=("seq",), descending=True, limit=1) + for event in item.events_in(connection)) else () class ExternalPage(DeclaredFamily): @@ -183,6 +257,48 @@ class ExternalTranscriptRows(StoreRows): """The tables of ExternalTranscript; each row declares itself one.""" +@dataclass(frozen=True) +class ExternalTurn(ExternalTranscriptRows, TypedTable): + """One external turn's opening deliveries, retained after its completion.""" + + seq: Annotated[int, Column(primary_key=True, auto_increment=True)] + thread: ThreadLookup + turn_id: NativeTurnId + replies: ReplyWindow + unique = (("thread", "turn_id"),) + indexes = (Index(("thread", "seq")), Index(("thread", "replies"))) + + @classmethod + def opened(cls, connection: Connection, thread: ThreadLookup, turn_id: NativeTurnId) -> ExternalTurn: + for turn in cls.select(connection, thread=thread, turn_id=turn_id): + return turn + cls.insert_values(connection, thread=thread, turn_id=turn_id, replies=OpeningReplies()) + return cls.require(connection, thread=thread, turn_id=turn_id) + + def store(self, connection: Connection, replies: ReplyWindow) -> None: + if replies != self.replies: + self.update(connection, where="seq=?", parameters=(self.seq,), replies=replies) + + def opening_references(self, connection: Connection) -> tuple[MessageReference, ...]: + return tuple(sorted((row.reference for row in ExternalTurnInput.select( + connection, thread=self.thread, turn_id=self.turn_id)), key=lambda reference: reference.seq)) + + +@dataclass(frozen=True) +class ExternalTurnInput(ExternalTranscriptRows, TypedTable): + """An indexed opening edge, not a second copy in the turn's state payload.""" + + thread: Annotated[ThreadLookup, Column(primary_key=True)] + turn_id: Annotated[NativeTurnId, Column(primary_key=True)] + reference: Annotated[MessageReference, Column(primary_key=True)] + indexes = (Index(("thread", "reference")),) + + @classmethod + def references(cls) -> tuple[ForeignKey, ...]: + return (ForeignKey(columns=("thread", "turn_id"), target=ExternalTurn, + target_columns=("thread", "turn_id")),) + + @dataclass(frozen=True) class ExternalItemRow(ExternalTranscriptRows, ExternalPageRow, TypedTable): """A stable item position; older pages prepend without moving saved cursors.""" @@ -212,7 +328,7 @@ def recent_turn_ids( f'WHERE previous.thread={table}.thread AND previous.turn_id={table}.turn_id ' f'AND previous.seq<{table}.seq)' ) - with closing(cls.stream(connection, row_type=ExternalTurn, thread=thread, + with closing(cls.stream(connection, row_type=ExternalPageRow, thread=thread, where=first_item, order_by=("seq",), descending=True)) as rows: for row in rows: yield row.turn_id @@ -467,10 +583,53 @@ def last_tool_call(self) -> EventTime: ).fetchone()[0] return GeneratedTimestamp.decode(timestamp) + def delivered(self, turn_id: NativeTurnId, references: tuple[MessageReference, ...]) -> None: + if not references: + return # Protocol notices and already-confirmed history introduce no input. + with self.writing() as connection: + turn = ExternalTurn.opened(connection, self.lookup, turn_id) + turn.replies.delivered(connection, turn, references) + + def opening_for(self, turn_id: NativeTurnId) -> tuple[MessageReference, ...]: + with self.snapshot() as connection: + turns = ExternalTurn.select(connection, thread=self.lookup, turn_id=turn_id) + return next((turn.replies.references_in(connection, turn) for turn in turns), ()) + + def answered_inputs(self, references: tuple[MessageReference, ...]) -> frozenset[MessageReference]: + """Probe only named originals and the final items of the turns they opened.""" + with self.snapshot() as connection: + inputs = ExternalTurnInput.select_keys(connection, tuple( + dict(thread=self.lookup, reference=reference) for reference in references)) + turns = tuple(dict.fromkeys(row.turn_id for row in inputs)) + answered = frozenset(reference + for turn_id in turns + for turn in ExternalTurn.select(connection, thread=self.lookup, turn_id=turn_id) + for reference in turn.replies.answered(connection, turn)) + return answered.intersection(references) + + def finish_replies(self, turn_id: NativeTurnId) -> None: + with self.writing() as connection: + for turn in ExternalTurn.select(connection, thread=self.lookup, turn_id=turn_id): + turn.store(connection, turn.replies.finished()) + + def transition_replies(self, transition: Callable[[ReplyWindow], ReplyWindow]) -> None: + """Indexed active-state probes retire all open turns without decoding history.""" + with self.writing() as connection: + for turn in ExternalTurn.select_keys(connection, tuple( + dict(thread=self.lookup, replies=member()) + for member in ReplyWindow.members_with(ActiveReplies))): + turn.store(connection, transition(turn.replies)) + + def observe_output(self, connection: Connection, turn_id: NativeTurnId, + events: tuple[TranscriptEvent, ...]) -> None: + turn = ExternalTurn.opened(connection, self.lookup, turn_id) + turn.store(connection, turn.replies.observed(events)) + def finish(self, *, turn_id: NativeTurnId, item_id: NativeItemId, events: tuple[TranscriptEvent, ...]) -> None: """Append a finished item once; an item with no presentation adds nothing.""" with self.writing() as connection: + self.observe_output(connection, turn_id, events) if events: self._insert_once(connection, turn_id, item_id, events, self._cut(connection).offset + 1) DraftRow.delete(connection, thread=self.lookup, turn_id=turn_id, item_id=item_id) @@ -526,6 +685,7 @@ async def watch_requests(self, emit) -> None: def draft(self, *, turn_id: NativeTurnId, item_id: NativeItemId, events: tuple[TranscriptEvent, ...]) -> None: with self.writing() as connection: + self.observe_output(connection, turn_id, events) DraftRow(self.lookup, turn_id, item_id, events).upsert(connection) def end_turn(self) -> None: diff --git a/src/agent_comms/harness_delivery_states.py b/src/agent_comms/harness_delivery_states.py index 079388dcac..ae58158de1 100644 --- a/src/agent_comms/harness_delivery_states.py +++ b/src/agent_comms/harness_delivery_states.py @@ -11,7 +11,7 @@ from functools import partial from typing import TYPE_CHECKING -from .channel_targets import CommandTarget +from .message_reference import MessageReference from .declared_family import DeclaredFamily from .field_codec import TranscriptRecordId @@ -85,7 +85,7 @@ class MonitorEvidence(DeclaredFamily): """Unsent and terminal observations cannot accept another delivery.""" async def confirm(self, bridge: LineMonitorBridge, delivery: HarnessDelivery, - notification: MonitorPart, record: ObservedMonitor) -> tuple[str, ...]: + notification: MonitorPart, record: ObservedMonitor) -> tuple[MessageReference, ...]: return () def evidence(self, recorded: type[Outcome]) -> tuple[type[Outcome], MonitorEvidence]: @@ -104,7 +104,7 @@ class ReceivingMonitor(MonitorEvidence): total: int async def confirm(self, bridge: LineMonitorBridge, delivery: HarnessDelivery, - notification: MonitorPart, record: ObservedMonitor) -> tuple[str, ...]: + notification: MonitorPart, record: ObservedMonitor) -> tuple[MessageReference, ...]: if notification.total != self.total: return () parts = await bridge.observe_part(delivery, notification, record) @@ -116,9 +116,7 @@ async def confirm(self, bridge: LineMonitorBridge, delivery: HarnessDelivery, await bridge.record_mismatch(delivery, 'Observed parts do not equal the original message') return () await bridge.finish(delivery, partial(replace, outcome=DeliveredOutcome, monitor=record)) - return (CommandTarget.named(message.target).automatic_reply_targets(message.sender) - if message.role.response_policy(message).starts_turn_for(message, bridge.binding.bus) - else ()) + return message.opening_references(bridge.binding.bus) @dataclass(frozen=True) diff --git a/src/agent_comms/historical_native_inputs.py b/src/agent_comms/historical_native_inputs.py index b010183b17..37195dce59 100644 --- a/src/agent_comms/historical_native_inputs.py +++ b/src/agent_comms/historical_native_inputs.py @@ -11,12 +11,20 @@ from agent_comms.native_session_files import NativeSessionFiles from abc import abstractmethod from dataclasses import dataclass +from pathlib import Path from typing import Annotated, TYPE_CHECKING from agent_comms.coordination_errors import IdentityConflict from agent_comms.coordinator import Coordination from .cohort_schema import CohortTable +from .coordination_database import CoordinationStore +from .coordination_session import ReadNativeCoordinationSession +from .message_reference import MessageReference +from .native_entries import NativeEvidenceScope +from .native_transcript import NativeTranscript +from .private_registry_guard import RegistryPrivacy +from .thread_identity import ThreadIncarnation from agent_comms.native_runtime_input import NativeRuntimeTable from .declared_family import DeclaredFamily from .field_codec import AssignmentId, InputId, MessageId, WireName, RootId, ThreadLookup @@ -33,8 +41,6 @@ from .selected_triage import SelectedTriage from .typed_table import TypedRow -if TYPE_CHECKING: - from .native_entries import NativeEvidenceScope @dataclass(frozen=True, kw_only=True) @@ -89,6 +95,34 @@ def proves_triage_source(self, evidence) -> bool: return False +@dataclass(frozen=True) +class HistoricalNativeAnswers: + """Join retained input proofs to their own bounded transcript output.""" + + root: Path + owner: ThreadIncarnation + privacy: RegistryPrivacy + + def answered_inputs(self, references: tuple[MessageReference, ...]) -> frozenset[MessageReference]: + with Coordination(str(CoordinationStore.path_in(self.root)), + session_type=ReadNativeCoordinationSession) as store, NativeEvidenceScope() as reads: + sources: dict[MessageReference, tuple[HistoricalNativeInput, ...]] = {} + for reference in references: + proofs = read_historical_native_inputs( + store, wire_root_id=self.privacy.namespace, + recipient_lookup=self.owner.require_recorded().lookup, + source_seq=reference.seq, source_reads=reads) + if any(proof.source_message_id != reference.message_id for proof in proofs): + raise IdentityConflict("Native answer source differs from its named message") + sources[reference] = tuple(proof for proof in proofs + if proof.execution.proves_full_source(proof)) + contexts = dict.fromkeys(proof.context for proofs in sources.values() for proof in proofs) + completed = frozenset(context for context in contexts + if NativeTranscript(context.session_file).completed_input(context)) + return frozenset(reference for reference, proofs in sources.items() + if any(proof.context in completed for proof in proofs)) + + @dataclass(frozen=True) class _HistoricalSource(TypedRow): input_id: InputId diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py index 7425d003dd..b489bcee7e 100644 --- a/src/agent_comms/messages.py +++ b/src/agent_comms/messages.py @@ -1032,6 +1032,11 @@ def exact_reply_target(self) -> str: """The exact response destination owned by this opening message.""" return CommandTarget.named(self.target).reply_target(self.sender) + def opening_references(self, recipient: str) -> tuple[MessageReference, ...]: + """A confirmed delivery can open a reply only for its policy's intended reader.""" + return ((self.reference,) + if self.role.response_policy(self).starts_turn_for(self, recipient) else ()) + autonomous: ClassVar[bool] = False @property diff --git a/src/agent_comms/messaging.py b/src/agent_comms/messaging.py index 8210b6cf69..16e111e243 100644 --- a/src/agent_comms/messaging.py +++ b/src/agent_comms/messaging.py @@ -17,6 +17,9 @@ ) from .message_bus import MessageBus from .message_reference import MessageReference +from .coordination_session import NativeCapture +from .external_transcript import ExternalTranscript +from .turn_item import NoSourceTurn, SourceTurn from .messages import ConversationRole, InfoKind, Message, MessageKind, MessageRole, NoticeRole from .registration import Registration from .send_result import CorrectionComplete, CorrectionFailed, CorrectionResult, SendResult @@ -71,6 +74,7 @@ def send( report: ReportEvidence = NoChangeReport(), answers: Sequence[MessageReference] = (), origin: AgentOrigin = AgentOrigin(), + source_turn: SourceTurn = NoSourceTurn(), ) -> SendResult: """Return the committed envelope and its separate wording-correction outcome.""" type.require_sendable() @@ -97,7 +101,7 @@ def send( role=role, task=task, report=report, - answers=tuple(dict.fromkeys((*answers, *self._answered_by(owner, target)))), + answers=tuple(dict.fromkeys((*answers, *self._answered_by(owner, target, source_turn)))), ) committed = self._publish(message, origin, destination=CommandTarget.named(target)) correction: CorrectionResult @@ -109,9 +113,17 @@ def send( correction = CorrectionComplete() return SendResult(committed, correction) - def _answered_by(self, owner: Thread, target: str) -> tuple[MessageReference, ...]: - """What this owner's message to target answers among its running turn's inputs.""" - return owner.turn_lease.answered_inputs(self, owner, target) + def _answered_by(self, owner: Thread, target: str, + source_turn: SourceTurn) -> tuple[MessageReference, ...]: + """Native lease or explicitly named external source; never a newest-row guess.""" + references = list(owner.turn_lease.answered_inputs(self, owner, target)) + with NativeCapture(self.bus) as boundary, self.bus.bus.reading() as source: + transcript = ExternalTranscript(self.root, owner.incarnation) + route = CommandTarget.named(target) + for turn_id in source_turn.ids(): + opening = transcript.opening_for(turn_id) + references.extend(source.answering(opening, route, boundary.registry)) + return tuple(dict.fromkeys(references)) def user_identity(self, worktree: Checkout) -> Thread: """One durable human sender, never an executor or a tag-derived agent role.""" diff --git a/src/agent_comms/native_entries.py b/src/agent_comms/native_entries.py index 8f390b31d3..cac84ce458 100644 --- a/src/agent_comms/native_entries.py +++ b/src/agent_comms/native_entries.py @@ -107,6 +107,16 @@ def imported_child(self, message: ImportedMessage, timestamp: datetime) -> dict[ def input_boundary(self): return any(message.user for message in self.messages) + @property + def tracked_user(self) -> MessageEntry | None: + return None + + def require_tracked_user(self) -> MessageEntry: + tracked = self.tracked_user + if tracked is None: + raise MalformedRecord("Native publication requires an original tracked input") + return tracked + def require_entry_id(self) -> str: return self.source_coordinates.original_id @@ -207,16 +217,6 @@ def open_evidence(cls, session_file): with NativeEvidenceRead.open(session_file) as evidence: yield evidence - @property - def tracked_user(self) -> MessageEntry | None: - return None - - def require_tracked_user(self) -> MessageEntry: - tracked = self.tracked_user - if tracked is None: - raise MalformedRecord("Native publication requires an original tracked input") - return tracked - @property def model_choice(self) -> tuple[str, str] | None: return None diff --git a/src/agent_comms/native_input_owner.py b/src/agent_comms/native_input_owner.py index 0a2dc623f4..c56265f66a 100644 --- a/src/agent_comms/native_input_owner.py +++ b/src/agent_comms/native_input_owner.py @@ -23,8 +23,8 @@ RunningTurnRegistryCheck, ) from .reservation_rules import ReservationViolationError -from .threads import Thread from .turn_goal_permission import GoalPermission, OwnerGoalPermission +from .threads import Thread if TYPE_CHECKING: from .goals import Goal diff --git a/src/agent_comms/native_input_record.py b/src/agent_comms/native_input_record.py index 81422f126b..4ee1181f07 100644 --- a/src/agent_comms/native_input_record.py +++ b/src/agent_comms/native_input_record.py @@ -37,6 +37,7 @@ WireName, ) from .native_session_reopen import PiSessionId +from .turn_item import FreshNativeId from .transcript_routes import InputDisplay, NoDisplayText from .typed_table import Column, IntegerStorage, SqlStorage, TextStorage @@ -65,7 +66,7 @@ class NativeInputCommit: Column(primary_key=True), ] session_id: Annotated[PiSessionId, WireName("sessionId")] - session_entry_id: Annotated[str, WireName("sessionEntryId")] + session_entry_id: Annotated[FreshNativeId, WireName("sessionEntryId")] @property def input_commit(self) -> NativeInputCommit: diff --git a/src/agent_comms/native_runtime_input.py b/src/agent_comms/native_runtime_input.py index d2086fc1bd..710f5fc2a8 100644 --- a/src/agent_comms/native_runtime_input.py +++ b/src/agent_comms/native_runtime_input.py @@ -30,6 +30,7 @@ ToolCallId, ) from .native_claim_epoch import NativeClaimEpoch, UnrecordedNativeClaim +from .turn_item import FreshNativeId from .native_input_record import ( NativeColumn, NativeInputContext, @@ -290,7 +291,7 @@ def published_replies(self, db, user, owner_lookup): ] = field(default_factory=UnrecordedNativeClaim) session_id: PiSessionId | NoPiSessionId = NoPiSessionId() session_file: str | None = None - session_entry_id: Annotated[NativeColumn, NativeContext(str), Column(storage=TextStorage, nullable=True)] = UnrecordedNativeColumn() + session_entry_id: Annotated[NativeColumn, NativeContext(FreshNativeId), Column(storage=TextStorage, nullable=True)] = UnrecordedNativeColumn() session_entry_start: Annotated[ NativeInputPosition, NativeInputLocation.position_column, diff --git a/src/agent_comms/native_transcript.py b/src/agent_comms/native_transcript.py index a6a7a106c0..b494db649a 100644 --- a/src/agent_comms/native_transcript.py +++ b/src/agent_comms/native_transcript.py @@ -4,18 +4,18 @@ from abc import abstractmethod from collections import defaultdict -from collections.abc import Generator +from collections.abc import Generator, Sequence from contextlib import closing, contextmanager, nullcontext +from itertools import takewhile from dataclasses import dataclass from math import inf from pathlib import Path -from typing import TYPE_CHECKING, BinaryIO +from typing import TYPE_CHECKING, BinaryIO, Never from .errors import RelationViolationError -from .native_pi import NativeContextJournal +from .native_pi import NativeContextJournal, NativeContextProof if TYPE_CHECKING: - from .native_pi import NativeContextProof from .native_output import NativeRecordLocation from .read_ledger import ReadLedger from .thread_identity import RecordedThreadIncarnation @@ -64,9 +64,11 @@ def reverse_records( stream.seek(start) yield start, end, stream.read(end - start) end = start - if floor == 0 and end: - stream.seek(0) - yield 0, end, stream.read(end) + if end > floor: + stream.seek(max(0, floor - 1)) + if floor == 0 or stream.read(1) == b"\n": + stream.seek(floor) + yield floor, end, stream.read(end - floor) def record_end(stream: BinaryIO, size: int) -> int: @@ -190,15 +192,34 @@ def annotation_id(self) -> FreshNativeId | NoEntryId: def events(self): return self.project(lambda original: original.entry.events(TranscriptProjection(position=original.start))) - def publication_producer(self) -> NativeEntry | None: + @property + def input_reported(self) -> bool: + return any(message.input_id.reported + for message in self.project(lambda original: original.entry.messages)) + + def require_entry(self) -> RecordedEntry: if not self.complete or self.entry is None: raise MalformedRecord("Native publication has incomplete evidence") - if self.entry.final_reply: - self.entry.require_entry_id() - return self.entry + return self.entry + + def publication_producer(self) -> RecordedEntry | None: + entry = self.require_entry() + if entry.final_reply: + entry.require_entry_id() + return entry return None +class NoNativeInput(NativeRecord): + """A parent chain that reached its root without an input record.""" + + def __init__(self) -> None: + super().__init__(0, 0, None, False) + + def require_entry(self) -> Never: + raise RelationViolationError("Native publication has no tracked input ancestry") + + class StoredRecord(NativeRecord): @property def size(self): @@ -209,6 +230,23 @@ class NativeTranscript(TranscriptSource): def __init__(self, path: Path): self.path = path + def completed_input(self, context: NativeContextProof) -> bool: + """Start at the proved input byte offset; never borrow the following input's final.""" + cut = self.cut() + with closing(self.forward(context.session_entry_start, cut.offset)) as records: + anchor = next(records) + if not anchor.project(lambda original: original.entry.require_entry_id() == context.session_entry_id): + raise TranscriptChanged("Native answer input changed after its proof was read") + own_turn = takewhile( + lambda record: not record.project(lambda original: original.entry.input_boundary), records) + finals = tuple(record for record in own_turn if record.project(lambda original: + original.entry.final_reply and any( + text for event in original.events for text in event.assistant_parts()))) + answered = anchor in self.inputs_for(finals, floor=context.session_entry_start) + if not cut.holds(self.cut(), cut.offset): + raise TranscriptChanged("Native answer source changed during its read") + return answered + def _tool_call_times(self, since: float = -inf) -> Generator[float]: try: with closing(self.tail()) as entries: @@ -353,7 +391,7 @@ def require_publication_context( self.path.absolute(), header, user, session_entry_start=session_entry_start, ) - def inputs_for(self, records) -> tuple: + def inputs_for(self, records: Sequence[NativeRecord], *, floor: int = 0) -> tuple[NativeRecord, ...]: """Find the user input behind each record in one reverse walk for the batch. Parent IDs are followed back from the latest record; every intervening @@ -361,7 +399,7 @@ def inputs_for(self, records) -> tuple: chain reaches a user boundary or a root. Tool rounds and branches cannot lend another input's publication proof. """ - found = [None] * len(records) + found: list[NativeRecord] = [NoNativeInput()] * len(records) waiting: dict[str | RootEntry, list[int]] = defaultdict(list) latest = 0 for position, record in enumerate(records): @@ -376,7 +414,7 @@ def inputs_for(self, records) -> tuple: # A chain that reaches a root (no parent) has no input. waiting.pop(RootEntry(), []) if waiting: - for ancestor in self.reverse(latest): + for ancestor in self.reverse(latest, floor=floor): entry = ancestor.entry if entry is None or entry.id not in waiting: continue @@ -391,7 +429,7 @@ def inputs_for(self, records) -> tuple: break return tuple(found) - def input_for(self, record: NativeRecord): + def input_for(self, record: NativeRecord) -> NativeRecord: """Follow the native record's parents to find the user input that started it.""" return self.inputs_for((record,))[0] @@ -408,9 +446,7 @@ def publication_input(self, *, after: int, through: int, files: NativeSessionFil if producer is None: continue user = self.input_for(record) - if user is None: - raise RelationViolationError("Native publication has no tracked input ancestry") - entry = user.entry.require_tracked_user() + entry = user.require_entry().require_tracked_user() self.require_publication_context(entry, session_entry_start=user.start, files=files) return record, entry raise RelationViolationError("Native turn has no final reply after its checkpoint") @@ -433,6 +469,6 @@ def forward(self, after: int, through: int): return yield NativeRecord.read(start, stream.tell(), raw) - def reverse(self, before: int): - for start, end, raw in reverse_records(self.path, before): + def reverse(self, before: int, *, floor: int = 0) -> Generator[NativeRecord]: + for start, end, raw in reverse_records(self.path, before, floor=floor): yield NativeRecord.read(start, end, raw) diff --git a/src/agent_comms/private_send_rules.py b/src/agent_comms/private_send_rules.py index aef7c5b264..14dfd0cbb9 100644 --- a/src/agent_comms/private_send_rules.py +++ b/src/agent_comms/private_send_rules.py @@ -12,8 +12,8 @@ from .coordination_tables.participants import OwnerGenerations from .reservation_rules import ReservationRule, RuleCheck from .text_digest import TextDigest -from .threads import Thread from .turn_goal_permission import GoalPermission +from .threads import Thread if TYPE_CHECKING: from .native_input_record import NativeInputRecord diff --git a/src/agent_comms/read_basis.py b/src/agent_comms/read_basis.py index 275375d3a8..3658398384 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -396,7 +396,6 @@ def sql(parameter: str) -> str: FROM json_each({parameter}))""" unread_sql = "seq > high_water AND seq NOT IN (SELECT value FROM json_each(exceptions))" - read_sql = f"NOT ({unread_sql})" def unread_rule(self) -> str: """Which joined request rows this cursor still counts; a human read is the whole rule.""" diff --git a/src/agent_comms/runtime_updates.py b/src/agent_comms/runtime_updates.py index add4b6e25c..300047964c 100644 --- a/src/agent_comms/runtime_updates.py +++ b/src/agent_comms/runtime_updates.py @@ -16,6 +16,8 @@ from .agent_activity import AgentActivity from .cli_protocol import EarlierItems from .external_transcript import ExternalPage, ExternalTranscript +from .channel_targets import CommandTarget +from .message_reference import MessageReference from .recorded_time import EventTime, UnrecordedEventTime from .transcript_events import ( LiveTextTranscript, @@ -25,7 +27,7 @@ from .thread_text import NoThreadText, ThreadText from .thread_identity import OfferedTools from .turn_phase import IdlePhase, ModelWaitPhase, TurnPhase -from .turn_item import NativeItemId, NativeTurnId, TurnItem +from .turn_item import NativeItemId, NativeTurnId, NamedSourceTurn, TurnItem if TYPE_CHECKING: from .agent_events import RuntimeInfoObservation @@ -174,9 +176,9 @@ class ThreadOutput(activity.ThreadOutput): def for_comms(cls, transcript: ExternalTranscript, comms: Comms) -> ThreadOutput: return cls(transcript, comms.agents) - def delivered_targets(self, targets: tuple[str, ...]) -> None: - """Ordinary transcript outputs do not publish session replies.""" - return None + def delivered(self, turn_id: NativeTurnId, references: tuple[MessageReference, ...]) -> None: + """Every external turn records the original deliveries that opened it.""" + self.transcript.delivered(turn_id, references) def publish_item(self, item: FinishedItem) -> None: self.transcript.finish(turn_id=item.turn_id, item_id=item.item_id, events=item.events) @@ -223,11 +225,17 @@ def finish_item(self, update: CompletedItem) -> FinishedItem: return finished def end_turn(self, turn_id: NativeTurnId) -> None: + self.transcript.finish_replies(turn_id) for key in tuple(self.drafts): if key.turn_id == turn_id: del self.drafts[key] def end_all_turns(self) -> None: + self.transcript.transition_replies(lambda window: window.finished()) + self.drafts.clear() + + def cancel_all_turns(self) -> None: + self.transcript.transition_replies(lambda window: window.cancelled()) self.drafts.clear() @property @@ -237,19 +245,16 @@ def activity_thread(self) -> str: @dataclass class ReplyingThreadOutput(ThreadOutput): - """The last finished assistant item replies to this turn's confirmed deliveries.""" + """The last finished assistant item replies only to deliveries that opened its turn.""" messaging: Messaging = field(kw_only=True) - reply_targets: set[str] = field(default_factory=set) pending: dict[TurnItem, FinishedItem] = field(default_factory=dict) def staged(self) -> Self: - return replace(super().staged(), reply_targets=self.reply_targets.copy(), - pending=self.pending.copy()) + return replace(super().staged(), pending=self.pending.copy()) def commit(self, staged: Self) -> None: super().commit(staged) - self.reply_targets = staged.reply_targets self.pending = staged.pending @classmethod @@ -260,9 +265,6 @@ def for_comms(cls, transcript: ExternalTranscript, comms: Comms) -> ThreadOutput commanded=cls(transcript, comms.agents, messaging=comms.messaging), ) - def delivered_targets(self, targets: tuple[str, ...]) -> None: - self.reply_targets.update(targets) - def flush_before(self, key: TurnItem) -> None: for previous in tuple(self.pending): if previous != key: @@ -270,7 +272,7 @@ def flush_before(self, key: TurnItem) -> None: def publish_item(self, item: FinishedItem) -> None: self.flush_before(item.key) - if self.reply_targets and item.assistant_text: + if item.assistant_text: self.pending[item.key] = item self.transcript.draft(turn_id=item.turn_id, item_id=item.item_id, events=item.events) else: @@ -281,6 +283,9 @@ def publish_draft(self, item: DraftItem) -> None: super().publish_draft(item) def end_turn(self, turn_id: NativeTurnId) -> None: + opening = self.messaging.bus.bus.messages_for_references(self.transcript.opening_for(turn_id)) + targets = {target for message in opening + for target in CommandTarget.named(message.target).automatic_reply_targets(message.sender)} for key in tuple(self.pending): if key.turn_id != turn_id: continue @@ -288,22 +293,25 @@ def end_turn(self, turn_id: NativeTurnId) -> None: publications = tuple( self.messaging.send( sender=self.activity_thread, target=target, body=item.assistant_text, + source_turn=NamedSourceTurn(turn_id), ).message.reference - for target in sorted(self.reply_targets) + for target in sorted(targets) ) super().publish_item(replace(item, events=tuple( event.with_publications(publications) for event in item.events ))) del self.pending[key] - self.reply_targets.clear() super().end_turn(turn_id) def end_all_turns(self) -> None: for turn_id in dict.fromkeys(key.turn_id for key in self.pending): self.end_turn(turn_id) - self.reply_targets.clear() super().end_all_turns() + def cancel_all_turns(self) -> None: + self.pending.clear() + super().cancel_all_turns() + class ReadingPhase(ABC): """Where a reader is in its source: catching up on what was already written, or following it.""" @@ -623,3 +631,10 @@ class AllTurnsEnded(TurnEnded): def retire(self, output: ThreadOutput) -> None: output.end_all_turns() + + +class AllTurnsCancelled(TurnEnded): + """Interrupted output is retained context, not a completed answer.""" + + def retire(self, output: ThreadOutput) -> None: + output.cancel_all_turns() diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py index 2fc7c3238e..988b84844f 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -31,6 +31,7 @@ ) from agent_comms.held_orders import HeldOrders from agent_comms.messages import Message +from agent_comms.message_reference import MessageReference from agent_comms.owner_requests import OwnerRequestId, RequestDirective from agent_comms.recorded_time import EventTime, RecordedEventTime from agent_comms.repository_name import RepositoryName @@ -642,13 +643,27 @@ def opened(cls, comms: Comms, commander: str) -> tuple[Self, ...]: @classmethod def read_scan(cls, comms: Comms, commander: str, scan: HeldOrders) -> tuple[Self, ...]: # Capture committed order/directive references before the bus cut. - snapshot = comms.registry.snapshot() - requests = comms.goals.todos.open_requests(snapshot.require(commander).incarnation) + cut = comms.registry.store.snapshot_with_authority() + snapshot = cut.snapshot + thread = snapshot.require(commander) + requests = comms.goals.todos.open_requests(thread.incarnation) held = scan.holders with comms.bus.bus.reading() as source: opened = tuple(rule for member in cls.members_with(cls) for rule in member.read_open(source, snapshot, held, commander, requests)) - return tuple(rule for rule in opened if rule.undecided(comms.goals.todos)) + references = tuple(dict.fromkeys(reference for rule in opened + for reference in rule.session_questions())) + answered = comms.transcripts.answered_inputs(thread, references, cut.privacy) + return tuple(rule for rule in opened if rule.undecided(comms.goals.todos) + and not rule.answered_in_session(answered)) + + def session_questions(self) -> tuple[MessageReference, ...]: + """Ordinary owed answers still require their existing bus answer relation.""" + return () + + def answered_in_session(self, answered: frozenset[MessageReference]) -> bool: + questions = self.session_questions() + return bool(questions) and all(question in answered for question in questions) def answered_by(self, named: tuple[Message, ...]) -> bool: """Whether an action naming these open messages answers this one.""" @@ -1122,6 +1137,9 @@ class DirectiveUnacted(OwedAnswer, DatedStandingRule): request: OwnerRequestId directive: RequestDirective + def session_questions(self) -> tuple[MessageReference, ...]: + return (self.message.reference,) + @classmethod def read_open(cls, source: BusReader, snapshot: RegistrySnapshot, held: Mapping[ThreadIncarnation, Sequence[Todo]], commander: str, diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 157824b9ee..fcbdd926e4 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -846,7 +846,9 @@ class SendTool(PublishingTool[ThreadCaller]): description = ( "Send a message to another thread or channel. Address specific threads with " "@thread-name in the body: the named threads are told they are the intended " - "readers, and others dismiss quietly. Unknown names stay plain text." + "readers, and others dismiss quietly. Unknown names stay plain text. " + "External sends may name received originals in answers; " + "source-turn identity is not a model argument." ) target: Annotated[str, *tool_field("Target thread name or channel", wire_name="to")] body: Annotated[str, *tool_field("Message text")] @@ -861,6 +863,11 @@ class SendTool(PublishingTool[ThreadCaller]): *tool_field("Change report fields; required for the change-report kind"), ] = NoChangeReport() + answers: Annotated[ + tuple[MessageReference, ...], + *tool_field("Original received messages explicitly answered; external sends without a named source turn must name these to link answers"), + ] = () + def publish(self, comms: Comms, caller: ThreadCaller) -> MessagePublished: sender = caller.own_thread() target = CommandTarget.named(self.target) @@ -869,7 +876,8 @@ def publish(self, comms: Comms, caller: ThreadCaller) -> MessagePublished: body = target.addressed(self.audience, self.body) result = comms.messaging.send( sender, target.name, body, self.type, report=self.report, - task=self.attachment(comms, target), origin=caller.origin()) + task=self.attachment(comms, target), origin=caller.origin(), + answers=self.answers) except ToolRefusedError as error: # The caller hears the refusal now; the record waits for the wire lock on its own. threading.Thread(target=SendRefusal.record, diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index de4b8f8185..a0576f1e5c 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -135,6 +135,15 @@ def for_assigned_source(self, source, published: bool) -> tuple[TranscriptEvent, def assistant_parts(self) -> tuple[str, ...]: return () + def reply_window[Window](self, window: Window, answering: Callable[[], Window]) -> Window: + """Input and system evidence do not start an assistant answer.""" + return window + + @property + def answers_turn(self) -> bool: + """Non-answer events do not clear a turn's opening questions.""" + return False + def with_publications(self, publications: tuple[MessageReference, ...]) -> TranscriptEvent: return self @@ -255,6 +264,15 @@ def routed(self) -> bool: class AgentTextTranscript(OutgoingRoute, LiveTextTranscript): """Native output that can continue streaming inside one mounted block.""" + unrouted_category: ClassVar[type[MessageCategory]] = AgentCategory + + @property + def category(self) -> type[MessageCategory]: + return OutboundCategory if self.routed else self.unrouted_category + + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + ImportedReply(self.text).import_to(accept) + @dataclass(frozen=True, kw_only=True) class WireTextTranscript(MarkdownTranscript): @@ -347,7 +365,22 @@ def for_assigned_source(self, source, published: bool) -> tuple[TranscriptEvent, return (replace(self, routing=self.routing.with_requests(remaining)),) -class AssistantTranscript(AgentTextTranscript): +class ModelOutput: + """Assistant text, thinking and tool output close the window for new automatic reply routes.""" + + def reply_window[Window](self, window: Window, answering: Callable[[], Window]) -> Window: + return answering() + + +class NarrationTranscript(ModelOutput, AgentTextTranscript): + """Visible assistant progress closes admission but is not an answer or DM final.""" + + +class AssistantTranscript(ModelOutput, AgentTextTranscript): + @property + def answers_turn(self) -> bool: + return bool(self.text) + def append_to(self, events: list[TranscriptEvent]) -> None: if events and (merged := events[-1].followed_by_assistant(self)) is not None: events[-1] = merged @@ -371,18 +404,12 @@ def for_external_source(self, source: TranscriptSource) -> tuple[TranscriptEvent published = bool(references) and source.sent_messages(references).keys() >= set(references) return self.for_assigned_source(source, published) - @property - def category(self) -> type[MessageCategory]: - return OutboundCategory if self.routed else AgentCategory - - def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: - ImportedReply(self.text).import_to(accept) - class NoticeTranscript(AgentTextTranscript): - @property - def category(self) -> type[MessageCategory]: - return OutboundCategory if self.routed else OtherCategory + unrouted_category = OtherCategory + + def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: + """System notices are not conversation replies.""" def as_context(self) -> NoticeTranscript: # Native failures remain visible notices, including bounded triage. @@ -437,7 +464,7 @@ def core_message(self, source) -> Message: raise ValueError("Sent report no longer matches its original assigned wire message") -class ThinkingTranscript(SilentTranscript, LiveTextTranscript): +class ThinkingTranscript(ModelOutput, SilentTranscript, LiveTextTranscript): category = ThinkingCategory @property @@ -453,7 +480,7 @@ def import_to(self, accept: Callable[[ImportedMessage], None]) -> None: @dataclass(frozen=True, kw_only=True) -class ToolTranscript(SilentTranscript, RecordedTranscript): +class ToolTranscript(ModelOutput, SilentTranscript, RecordedTranscript): category = ToolCategory tool_call_id: ToolCallId tool_name: str diff --git a/src/agent_comms/transcript_sources.py b/src/agent_comms/transcript_sources.py index 182550c4fa..1ab870b23b 100644 --- a/src/agent_comms/transcript_sources.py +++ b/src/agent_comms/transcript_sources.py @@ -642,22 +642,18 @@ def native_records( lookup = self.recipient.lookup for fragment in fragments: fragment = tuple(fragment) - selected = tuple( - (record, user.entry if user else None) - for record, user in zip(fragment, reader.inputs_for(fragment), strict=True) - ) + selected = tuple(zip(fragment, reader.inputs_for(fragment), strict=True)) with TranscriptRead.publication_source(self.root) as db: # Missing coordination remains detached native history. Do not # introduce header failures where no SQL source was selected. session_id = (reader.session_id if db is not None and any( - user and user.message.input_id.reported - for _, user in selected) else None) + user.input_reported for _, user in selected) else None) capture = partial( NativeRuntimeInput.transcript_projection, db, reader, owner_lookup=lookup, session_id=session_id, ) - originals = tuple((record, record.project(partial(capture, user=user))) + originals = tuple((record, record.project(partial(capture, user=user.entry))) for record, user in selected) for record, projection in originals: yield record, record.project( diff --git a/src/agent_comms/transcripts.py b/src/agent_comms/transcripts.py index 7f98da9363..41d6b6b2bd 100644 --- a/src/agent_comms/transcripts.py +++ b/src/agent_comms/transcripts.py @@ -49,7 +49,7 @@ TranscriptSource, WireFrontier, ) -from .turn_digests import ThreadTurns, TurnBound, TurnDigestRead +from .turn_digests import InputAnswerRead, ThreadTurns, TurnBound, TurnDigestRead from .turn_lease import TurnLeaseFence if TYPE_CHECKING: @@ -515,6 +515,12 @@ def __init__(self, root: Path, registry: Registration, bus: MessageBus, messagin self.routes = TranscriptRoutes(root) self.page_reads = 0 + def answered_inputs( + self, thread: Thread, references: tuple[MessageReference, ...], privacy: RegistryPrivacy, + ) -> frozenset[MessageReference]: + """The same source family owns bounded answer evidence for native and external turns.""" + return InputAnswerRead(self.root, thread, references, privacy).read() + def bind_page_read( self, name: str, diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 46bbb2c55b..c985a3dc7d 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -18,6 +18,9 @@ from .external_transcript import ExternalTranscript, StoreRead from .field_codec import FieldCodec from .held_orders import HeldOrders, LiveHold +from .historical_native_inputs import HistoricalNativeAnswers +from .message_reference import MessageReference +from .private_registry_guard import RegistryPrivacy from .order_work import OrderWork from .repo import Checkout, GitCommandError, Repo from .storage_rules import CheckoutAllocation, StorageRule @@ -168,16 +171,38 @@ def full(self, taken: int) -> bool: @dataclass(frozen=True) -class TranscriptRead[Result](ConversationPreparation): +class ThreadRead[Result](ConversationPreparation): """The original native/external source selection, shared by typed readers.""" root: Path thread: Thread - bound: TurnBound def read(self) -> Result: return self.thread.execution.prepare_conversation(self) + +@dataclass(frozen=True) +class InputAnswerRead(ThreadRead[frozenset[MessageReference]]): + references: tuple[MessageReference, ...] + privacy: RegistryPrivacy + + def read(self) -> frozenset[MessageReference]: + if not self.references: + return frozenset() + return super().read() + + def native(self) -> frozenset[MessageReference]: + return HistoricalNativeAnswers(self.root, self.thread.incarnation, self.privacy).answered_inputs( + self.references) + + def external(self) -> frozenset[MessageReference]: + return ExternalTranscript(self.root, self.thread.incarnation).answered_inputs(self.references) + + +@dataclass(frozen=True) +class TranscriptRead[Result](ThreadRead[Result]): + bound: TurnBound + def native(self) -> Result: source = self.thread.runtime.history_source(self.root, self.thread.incarnation) return self.from_native(source, source.turns(self.bound)) diff --git a/src/agent_comms/turn_item.py b/src/agent_comms/turn_item.py index 747f1c1a2e..5c5e6201cd 100644 --- a/src/agent_comms/turn_item.py +++ b/src/agent_comms/turn_item.py @@ -6,7 +6,8 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Self -from .field_codec import Id128, NominalSpelling, NominalText +from .errors import RelationViolationError +from .field_codec import Id128, NominalSpelling, NominalText, NullMember, RecordMember, SubclassUnion if TYPE_CHECKING: from .routing import Routing @@ -23,6 +24,31 @@ class NativeTurnIdText(NominalSpelling[NativeTurnId]): spelling = re.compile(r"\S{1,4096}") +class SourceTurn(SubclassUnion, affix="SourceTurn"): + """A source-named turn, never a guess from the newest transcript row.""" + + def ids(self) -> tuple[NativeTurnId, ...]: + return () + + def require_id(self) -> NativeTurnId: + for turn_id in self.ids(): + return turn_id + raise RelationViolationError("Confirmed input has no source turn") + + +@dataclass(frozen=True) +class NoSourceTurn(NullMember, SourceTurn): + """No external turn was supplied; no external answer links may be inferred.""" + + +@dataclass(frozen=True) +class NamedSourceTurn(RecordMember, SourceTurn): + id: NativeTurnId + + def ids(self) -> tuple[NativeTurnId, ...]: + return (self.id,) + + class FreshNativeId(NominalText): """An opaque harness key Core may mint, with the harness's free text spelling.""" diff --git a/src/agent_comms/turn_lease.py b/src/agent_comms/turn_lease.py index d8804536c0..aa6b12a7c1 100644 --- a/src/agent_comms/turn_lease.py +++ b/src/agent_comms/turn_lease.py @@ -13,7 +13,6 @@ from .channel_targets import CommandTarget from .child_process import ProcessIdentity, ProcessState from .coordination_session import NativeCapture -from .bus_tables import MessageIdentitySource from .errors import RelationViolationError from .field_codec import ( NullMember, @@ -32,7 +31,6 @@ PatternFile, ) from .message_reference import MessageReference -from .read_basis import ChannelConversation, DMConversation from .reported_turn import NoReportedTurn, ReportedTurn, ReportedTurnState from .routing import NoTurnRouting, Routing from .store_files import DependentStateTier, SingleFileStore @@ -390,21 +388,7 @@ def held(self) -> tuple[TurnLeaseFence, ...]: def answered_inputs( self, messaging: Messaging, owner: Thread, target: str, ) -> tuple[MessageReference, ...]: - with NativeCapture(messaging.bus) as boundary, messaging.bus.bus.reading() as source: - snapshot = boundary.registry - route = CommandTarget.named(target) - conversation = route.conversation_key( - owner.name, snapshot, ChannelConversation, DMConversation, - ) - reads = messaging.bus.reads - document = reads.read_conversation(owner.name, conversation, snapshot) - cursor = reads.cursor(owner.name, snapshot, document=document) - references = source.delivered_unanswered( - owner.incarnation.require_recorded(), conversation.scope(snapshot), - cursor, snapshot, - ) - with source.observing() as db: - return MessageIdentitySource.answering(db, references, route, snapshot) + return () def attests(self, claim_scope: int, turn_id: TurnId) -> bool: return False diff --git a/stack/native/native-proof-schema.mjs b/stack/native/native-proof-schema.mjs index 92f335f476..7e45d1e43d 100644 --- a/stack/native/native-proof-schema.mjs +++ b/stack/native/native-proof-schema.mjs @@ -1,7 +1,7 @@ // Rendered at build from NativeContextJournal; never edited. export const nativeProofSchema = { "objects": { - "native_context_journal": "CREATE TABLE \"native_context_journal\" (\"input_id\" TEXT NOT NULL CHECK (\"input_id\" IS NULL OR typeof(\"input_id\") = 'text') CHECK (length(\"input_id\") BETWEEN 32 AND 32) CHECK (\"input_id\" NOT GLOB '*[^0-9a-f]*') CHECK (instr(\"input_id\", char(0)) = 0), \"session_id\" TEXT NOT NULL CHECK (\"session_id\" IS NULL OR typeof(\"session_id\") = 'text'), \"session_entry_id\" TEXT NOT NULL, \"request_generation\" INTEGER NOT NULL CHECK (request_generation BETWEEN 1 AND 9007199254740991), \"llm_context_digest\" TEXT NOT NULL CHECK (\"llm_context_digest\" IS NULL OR typeof(\"llm_context_digest\") = 'text') CHECK (length(\"llm_context_digest\") BETWEEN 64 AND 64) CHECK (\"llm_context_digest\" NOT GLOB '*[^0-9a-f]*') CHECK (instr(\"llm_context_digest\", char(0)) = 0), \"committed_at_ms\" INTEGER NOT NULL CHECK (committed_at_ms BETWEEN 1 AND 9007199254740991), \"schema\" INTEGER NOT NULL CHECK (\"schema\" IN (2)), \"type\" TEXT NOT NULL CHECK (\"type\" IN ('context_committed')), PRIMARY KEY (\"input_id\", \"request_generation\")) STRICT, WITHOUT ROWID", + "native_context_journal": "CREATE TABLE \"native_context_journal\" (\"input_id\" TEXT NOT NULL CHECK (\"input_id\" IS NULL OR typeof(\"input_id\") = 'text') CHECK (length(\"input_id\") BETWEEN 32 AND 32) CHECK (\"input_id\" NOT GLOB '*[^0-9a-f]*') CHECK (instr(\"input_id\", char(0)) = 0), \"session_id\" TEXT NOT NULL CHECK (\"session_id\" IS NULL OR typeof(\"session_id\") = 'text'), \"session_entry_id\" TEXT NOT NULL CHECK (\"session_entry_id\" IS NULL OR typeof(\"session_entry_id\") = 'text'), \"request_generation\" INTEGER NOT NULL CHECK (request_generation BETWEEN 1 AND 9007199254740991), \"llm_context_digest\" TEXT NOT NULL CHECK (\"llm_context_digest\" IS NULL OR typeof(\"llm_context_digest\") = 'text') CHECK (length(\"llm_context_digest\") BETWEEN 64 AND 64) CHECK (\"llm_context_digest\" NOT GLOB '*[^0-9a-f]*') CHECK (instr(\"llm_context_digest\", char(0)) = 0), \"committed_at_ms\" INTEGER NOT NULL CHECK (committed_at_ms BETWEEN 1 AND 9007199254740991), \"schema\" INTEGER NOT NULL CHECK (\"schema\" IN (2)), \"type\" TEXT NOT NULL CHECK (\"type\" IN ('context_committed')), PRIMARY KEY (\"input_id\", \"request_generation\")) STRICT, WITHOUT ROWID", "native_context_journal_request_generation_idx": "CREATE INDEX \"native_context_journal_request_generation_idx\" ON \"native_context_journal\" (\"request_generation\")", "native_context_journal_session_id_idx": "CREATE INDEX \"native_context_journal_session_id_idx\" ON \"native_context_journal\" (\"session_id\")", "native_context_journal_insert_before_abort": "CREATE TRIGGER \"native_context_journal_insert_before_abort\" BEFORE INSERT ON \"native_context_journal\" WHEN NEW.request_generation < COALESCE((SELECT MAX(request_generation)\n FROM native_context_journal), 0) OR EXISTS(SELECT 1 FROM native_context_journal\n WHERE request_generation=NEW.request_generation\n AND (llm_context_digest != NEW.llm_context_digest\n OR committed_at_ms != NEW.committed_at_ms)) OR NEW.session_id\n != (SELECT session_id FROM native_context_journal ORDER BY session_id LIMIT 1) BEGIN SELECT RAISE(ABORT, 'Insert violates row membership'); END", diff --git a/stack/pi-native.sha256 b/stack/pi-native.sha256 index ae1305f692..0ed44a2d97 100644 --- a/stack/pi-native.sha256 +++ b/stack/pi-native.sha256 @@ -22,7 +22,7 @@ fef47246f4ce18621bae844a940e7d009cefc7df7cdf00ccb98198d9c407a696 dist/core/sess 4fa4e9a0ba80f43b560eb827dbd0985ea56624170b51401b39398d76ffcbcefa dist/core/export-html/index.js d7ad93bcb341424fa79014ce2fec4226326cfab1e8d5fee27c1859138e120161 dist/core/sdk.js 37ae95f1f0ca7c88a75f3c83101e3d7b621c15afced8ac8ec287f958f150cb30 dist/core/native-proof-journal.js -42b8245b0f6c25304792047c99046724402658034be26343a26aed6755223646 dist/core/native-proof-schema.js +bc883f29dbb2a9ba342cb4635a483a5d350ee7d8cb208d43d424449776d41177 dist/core/native-proof-schema.js aae7f609bfa65d29ed16e311456921a88221622e08cf6a1f5c7aab8e58bd69e6 node_modules/@earendil-works/pi-ai/dist/api/openai-completions.js 99f2649a743a3bf2b8765b502c04cd8d45bf0f954bb1419b446669c07de0c4db agent-comms-extensions/global-agent-comms/index.mjs 8b9fce004ea6e178729afa9d2252c5031592125191b9bee796663d2b525c21f5 dist/agent-comms-imports.json @@ -56,5 +56,5 @@ d64e8c05c7932d377034820b1db0df45612afe64885c0765f8f348a12e1bc955 dist/core/skil 8e8276bbabcd898136d431345611928626fad6b6f03ac572f1d8506e6ece326f dist/core/system-prompt.d.ts 0391543cf703a93a0163b37b3ccebe53e0f12bff6f6162b86d56fb017fe49011 dist/core/system-prompt.js # agent-comms-native-stock-v1 @earendil-works/pi-coding-agent@0.85.1 -# agent-comms-native-tree-v1 09b00ac37b5c2e652588df901eb61c3b1983248071781cd17c9247f21e782d93 -# agent-comms-native-inputs-v1 4991f959f91f0ea0f2fc10f847303993fe4c95f443bb4e10782cb8bb67fe605a +# agent-comms-native-tree-v1 724dbf68818d1932e6981b737b8e1a8cf4280e592f4d63013306e41e9d15feb8 +# agent-comms-native-inputs-v1 5d32a1691f83443a9c82235830923331f195ce524046354b075d400e08496eb9 diff --git a/tests/test_bus_files.py b/tests/test_bus_files.py index 11f16b0386..8b5672a02c 100644 --- a/tests/test_bus_files.py +++ b/tests/test_bus_files.py @@ -23,6 +23,7 @@ from agent_comms.todos import TodoStore from agent_comms.native_session_reopen import PiSessionId from agent_comms.typed_table import ReconciledFile +from agent_comms.turn_item import FreshNativeId from bus_record_fixture import prepare_native_comms from test_todos import order @@ -96,7 +97,7 @@ def test_parameterized_sqlite_resources_follow_their_file_owner(tmp_path: Path, db.execute(statement) db.commit() NativeContextJournal( - input_id=InputId.mint(), session_id=PiSessionId("session"), session_entry_id="entry", + input_id=InputId.mint(), session_id=PiSessionId("session"), session_entry_id=FreshNativeId("entry"), request_generation=1, llm_context_digest="c" * 64, committed_at_ms=1_791_964_800_001, schema=2, type="context_committed", ).insert(db) @@ -113,7 +114,7 @@ def test_installer_reconciles_a_native_proof_journal_onto_its_declaration(tmp_pa for statement in NativeContextJournal.schema_objects().values(): db.execute(statement) NativeContextJournal( - input_id=InputId.mint(), session_id=PiSessionId("session"), session_entry_id="entry", + input_id=InputId.mint(), session_id=PiSessionId("session"), session_entry_id=FreshNativeId("entry"), request_generation=1, llm_context_digest="c" * 64, committed_at_ms=1_791_964_800_001, schema=2, type="context_committed", ).insert(db) diff --git a/tests/test_claude_automatic_reply.py b/tests/test_claude_automatic_reply.py index 62d20f7be5..73c9ea93ef 100644 --- a/tests/test_claude_automatic_reply.py +++ b/tests/test_claude_automatic_reply.py @@ -1,50 +1,335 @@ -"""A session's final answers a DM, never an outstanding channel mention.""" +"""Only a turn's opening DMs receive its final; channels always require explicit sends.""" + +import json +from pathlib import Path import pytest +from aiohttp import web -from agent_comms.cli_bridges.claude_protocol import ClaudeSession, KnownTurn +from agent_comms.cli_bridges.claude import ClaudeBridge +from agent_comms.cli_bridges.claude_protocol import ClaudeRecord, ClaudeSession +from agent_comms.cli_bridges.codex import CodexBridge, CodexOptions +from agent_comms.cli_bridges.codex_protocol import ItemStream, RpcNotification +from agent_comms.cli_bridges.shared import Binding, StoredBinding, Watch +from agent_comms.comms import Comms +from agent_comms.connector import ConnectorSendTool +from agent_comms.connector_caller import DirectorCaller +from agent_comms.standing_rules import DirectiveUnacted from agent_comms.harness_delivery_states import DeliveredOutcome -from agent_comms.turn_item import NativeTurnId +from agent_comms.messages import Message, QuestionKind +from agent_comms.registry_document import RegisteredCommanderProject +from agent_comms.errors import RelationViolationError +from agent_comms.runtime_updates import AllTurnsCancelled, AllTurnsEnded, FinishedTurn, RuntimeUpdate, ThreadOutput +from agent_comms.turn_item import NamedSourceTurn, NativeTurnId from transcript_record_fixture import transcript_record from test_monitor_delivery import attached_monitor, bridge, decoded_record, notification -@pytest.mark.asyncio -async def test_channel_mention_and_dm_share_a_turn_without_broadcasting_the_final(bridge): - comms = bridge.comms - comms.channels.create_tag('comms') - for name in ('claude-code', 'viewer', 'peer'): - comms.channels.update_tags(name, add=frozenset({'comms'})) - channel = comms.messaging.send('peer', '#comms', '@viewer please investigate').message - direct = comms.messaging.send('claude-code', 'viewer', 'please answer me privately').message - output = bridge.output_type.for_comms(comms.transcripts.external('viewer'), comms) - session = ClaudeSession(turn=KnownTurn(NativeTurnId('reply-turn'))) +def assistant_record(label: str, text: str) -> ClaudeRecord: + return decoded_record(dict( + type='assistant', uuid=transcript_record(label), timestamp='2026-10-29T01:00:00Z', + message=dict(content=[dict(type='text', text=text)]), + )) + + +def opening_record(prompt_id: str) -> ClaudeRecord: + return decoded_record(dict( + type='user', uuid=transcript_record(prompt_id), timestamp='2026-10-29T01:00:00Z', + sessionId='session', promptId=prompt_id, origin=dict(kind='human'), + message=dict(content='An earlier user turn'), + )) + + +async def confirm_messages( + bridge: ClaudeBridge, output: ThreadOutput, session: ClaudeSession, messages: tuple[Message, ...], + *, prompt_id: str = 'reply-turn', +) -> tuple[ClaudeRecord, ...]: + """A real attached monitor and Claude notification feed the ordinary staged output.""" + observations: list[ClaudeRecord] = [] async with attached_monitor(bridge) as reader: - for message in (channel, direct): + for message in messages: await bridge.accept(message) line = (await reader.readline()).decode().rstrip('\n') - output.publish_batch([ - await update.observed(bridge) for update in notification( - line, record_id=transcript_record(str(message.seq))).updates(session) - ]) + record = notification(line, record_id=transcript_record(str(message.seq)), prompt_id=prompt_id) + observations.append(record) + output.publish_batch([await update.observed(bridge) for update in record.updates(session)]) delivered, = await bridge.attempts(message.message_id) assert delivered.outcome is DeliveredOutcome + return tuple(observations) - def replies(): - return tuple((message.target, message.body) for message in comms.bus.bus.message_history() - if message.sender == 'viewer' and message.seq > channel.seq) - final = decoded_record(dict( - type='assistant', uuid=transcript_record('final'), timestamp='2026-10-29T01:00:00Z', - message=dict(content=[dict(type='text', text='Private answer for Tristan')]), - )) - output.publish_batch(list(final.updates(session))) - assert replies() == () +def replies_after(bridge: ClaudeBridge, sequence: int) -> tuple[tuple[str, str], ...]: + return tuple((message.target, message.body) + for message in bridge.comms.bus.bus.message_history() + if message.sender == bridge.binding.bus and message.seq > sequence) + + +def finish_reply(output: ThreadOutput, session: ClaudeSession) -> None: ended = decoded_record(dict(type='system', subtype='turn_duration')) output.publish_batch(list(ended.updates(session))) - assert replies() == (('claude-code', 'Private answer for Tristan'),) - output.publish_batch(list(ended.updates(session))) - assert replies() == (('claude-code', 'Private answer for Tristan'),) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(('prefix', 'targets'), ( + ((), ('claude-code',)), + ((assistant_record('already-answering', 'An answer already in progress'),), ()), + ((decoded_record(dict( + type='assistant', uuid=transcript_record('already-using-tool'), + timestamp='2026-10-29T01:00:00Z', message=dict(content=[dict( + type='tool_use', id='already-running', name='Bash', input=dict(command='pwd'))]), + )),), ()), + ((decoded_record(dict( + type='user', uuid=transcript_record('human-input'), timestamp='2026-10-29T01:00:00Z', + sessionId='session', promptId='reply-turn', origin=dict(kind='human'), + message=dict(content='Tristan asks a question'), + )),), ('claude-code',)), +)) +async def test_only_opening_deliveries_receive_the_turns_final( + bridge: ClaudeBridge, prefix: tuple[ClaudeRecord, ...], targets: tuple[str, ...], +) -> None: + output = bridge.output_type.for_comms(bridge.comms.transcripts.external('viewer'), bridge.comms) + session = ClaudeSession() + output.publish_batch(list(opening_record('reply-turn').updates(session))) + for record in prefix: + output.publish_batch(list(record.updates(session))) + direct = bridge.comms.messaging.send('claude-code', 'viewer', 'a delivered DM').message + observations = await confirm_messages(bridge, output, session, (direct,)) + output.publish_batch(list(assistant_record('final', 'Private answer').updates(session))) + assert replies_after(bridge, direct.seq) == () + finish_reply(output, session) + expected = tuple((target, 'Private answer') for target in targets) + assert replies_after(bridge, direct.seq) == expected + finish_reply(output, session) + assert replies_after(bridge, direct.seq) == expected + # A subscription rereads retained records, not an input or a provider attempt. + for record in observations: + output.publish_batch([await update.observed(bridge) for update in record.updates(session)]) + assert replies_after(bridge, direct.seq) == expected + + # Completion opens a fresh window; another sender can open the next turn. + following = session + next_dm = bridge.comms.messaging.send('peer', 'viewer', 'a later opening DM').message + await confirm_messages(bridge, output, following, (next_dm,), prompt_id='following-turn') + output.publish_batch(list(assistant_record('next-final', 'Next answer').updates(following))) + finish_reply(output, following) + assert replies_after(bridge, next_dm.seq) == (('peer', 'Next answer'),) + + +@pytest.mark.asyncio +async def test_channel_mention_and_dm_share_a_turn_without_broadcasting_the_final( + bridge: ClaudeBridge, +) -> None: + comms = bridge.comms + comms.channels.create_tag('comms') + for name in ('claude-code', 'viewer', 'peer'): + comms.channels.update_tags(name, add=frozenset({'comms'})) + channel = comms.messaging.send('peer', '#comms', '@viewer please investigate').message + direct = comms.messaging.send('claude-code', 'viewer', 'please answer me privately').message + output = bridge.output_type.for_comms(comms.transcripts.external('viewer'), comms) + session = ClaudeSession() + await confirm_messages(bridge, output, session, (channel, direct)) + output.publish_batch(list(assistant_record('final', 'Private answer for Tristan').updates(session))) + assert replies_after(bridge, channel.seq) == () + finish_reply(output, session) + assert replies_after(bridge, channel.seq) == (('claude-code', 'Private answer for Tristan'),) + finish_reply(output, session) + assert replies_after(bridge, channel.seq) == (('claude-code', 'Private answer for Tristan'),) # A channel remains writable through the explicit send path. comms.messaging.send('viewer', '#comms', 'explicit channel update') - assert replies()[-1] == ('#comms', 'explicit channel update') + assert replies_after(bridge, channel.seq)[-1] == ('#comms', 'explicit channel update') + + +def assert_opening_send(comms: Comms, opening: Message, later: Message, turn_id: str) -> None: + unbound = comms.messaging.send('viewer', opening.sender, 'No source identity was supplied').message + assert unbound.answers == () + answered = comms.messaging.send('viewer', opening.sender, 'The opening question is answered', + source_turn=NamedSourceTurn(NativeTurnId(turn_id))).message + assert answered.answers == (opening.reference,) + unrelated = comms.messaging.send('viewer', later.sender, 'A separate status update').message + assert unrelated.answers == () + assert comms.bus.bus.unanswered_references( + (opening.seq, later.seq), frozenset({'viewer'})) == (later.reference,) + + +@pytest.mark.asyncio +async def test_commander_send_answers_only_the_question_that_opened_its_turn(bridge: ClaudeBridge) -> None: + comms = bridge.comms + commander = comms.registry.require('viewer') + comms.registry.set_commander_project(RegisteredCommanderProject(commander.incarnation, ())) + output = bridge.output_type.for_comms(comms.transcripts.external('viewer'), comms) + session = ClaudeSession() + opening = comms.messaging.send('peer', 'viewer', 'Please investigate', QuestionKind).message + await confirm_messages(bridge, output, session, (opening,), prompt_id='commander-turn') + output.publish_batch(list(assistant_record('commander-output', 'Investigating').updates(session))) + later = comms.messaging.send('claude-code', 'viewer', 'Another question', QuestionKind).message + await confirm_messages(bridge, output, session, (later,), prompt_id='commander-turn') + assert_opening_send(comms, opening, later, 'commander-turn') + finish_reply(output, session) + outside = comms.messaging.send('viewer', 'claude-code', 'Outside that completed turn').message + assert outside.answers == () + + +@pytest.mark.asyncio +async def test_codex_send_answers_only_the_question_that_opened_its_turn( + bridge: ClaudeBridge, tmp_path: Path, +) -> None: + starts: list[dict] = [] + + async def serve(request: web.Request) -> web.WebSocketResponse: + socket = web.WebSocketResponse() + await socket.prepare(request) + async for message in socket: + frame = json.loads(message.data) + method = frame['method'] + if method == 'initialized': + continue + if method == 'initialize': + result = {'userAgent': 'codex/0.0.0'} + else: + assert method == 'turn/start' + starts.append(frame['params']) + result = {} + await socket.send_json({'id': frame['id'], 'result': result}) + return socket + + app = web.Application() + app.router.add_get('/', serve) + runner = web.AppRunner(app) + await runner.setup() + socket_path = tmp_path.parent / 'opening-codex.sock' + await web.UnixSite(runner, str(socket_path)).start() + comms = bridge.comms + state = tmp_path / 'codex-bridge' + binding = Binding(root=str(comms.root), owner=comms.registry.require('viewer').incarnation, + thread='codex-session', start=Binding.bus_head(str(comms.root))) + try: + with Binding.open(state) as db: + binding.bind(db) + options = CodexOptions(Watch(), StoredBinding(), state, state.parent, str(socket_path)) + codex = CodexBridge(options, binding.address(), db) + await options.operation.prepare(codex) + output = codex.output_type.for_comms(comms.transcripts.external('viewer'), comms) + stream = ItemStream('codex-session') + + async def receive(item: dict) -> None: + updates = stream.receive(RpcNotification(method='item/completed', params=dict( + threadId='codex-session', turnId='codex-turn', item=item, + completedAtMs=1761699600000))) + output.publish_batch([await update.observed(codex) for update in updates]) + + async def deliver(original: Message) -> None: + await codex.accept(original) + sent = starts[-1] + await receive(dict(type='userMessage', id=f'input-{original.seq}', + content=sent['input'], clientId=sent['clientUserMessageId'])) + + opening = comms.messaging.send('peer', 'viewer', 'Please investigate', QuestionKind).message + await deliver(opening) + await receive(dict(type='agentMessage', id='working', text='Investigating')) + later = comms.messaging.send('claude-code', 'viewer', 'Another question', QuestionKind).message + await deliver(later) + assert_opening_send(comms, opening, later, 'codex-turn') + finally: + await runner.cleanup() + + +TURN_FINISHED = decoded_record(dict(type='system', subtype='turn_duration')) + + +@pytest.mark.asyncio +@pytest.mark.parametrize(('prefix', 'suffix', 'cleared'), ( + ((), (assistant_record('directive-final', 'The requested work is explained'), TURN_FINISHED), True), + ((), (decoded_record(dict( + type='assistant', uuid=transcript_record('thinking-only'), timestamp='2026-10-29T01:00:00Z', + message=dict(content=[dict(type='thinking', thinking='Still considering the work')]), + )), TURN_FINISHED), False), + ((), (decoded_record(dict( + type='assistant', uuid=transcript_record('narration-and-final'), timestamp='2026-10-29T01:00:00Z', + message=dict(content=[dict(type='thinking', thinking='Checking the source'), + dict(type='text', text='The requested work is explained')]), + )), TURN_FINISHED), True), + ((), (assistant_record('not-finished', 'An unfinished answer'),), False), + ((opening_record('directive-turn'), assistant_record('before-directives', 'Already answering something else')), + (assistant_record('following-directives', 'Finishing that older answer'), TURN_FINISHED), False), + ((opening_record('directive-turn'), decoded_record(dict( + type='assistant', uuid=transcript_record('hidden-thinking'), timestamp='2026-10-29T01:00:00Z', + message=dict(content=[dict(type='redacted_thinking')]), + ))), (assistant_record('after-hidden-thinking', 'Answering the older input'), TURN_FINISHED), False), + ((), (assistant_record('interrupted-output', 'A partial answer'), decoded_record(dict( + type='user', uuid=transcript_record('interrupted'), timestamp='2026-10-29T01:00:00Z', + sessionId='session', promptId='directive-turn', + message=dict(content='[Request interrupted by user]'), + ))), False), + ((), (assistant_record('empty-final', ''), TURN_FINISHED), False), +), ids=('committed-final', 'thinking-only', 'narration-with-final', 'still-open', 'mid-output', + 'hidden-thinking-before-input', 'interrupted', 'empty-final')) +async def test_twelve_directives_require_the_final_of_the_turn_they_opened( + bridge: ClaudeBridge, prefix: tuple[ClaudeRecord, ...], suffix: tuple[ClaudeRecord, ...], cleared: bool, +) -> None: + comms = bridge.comms + commander = comms.registry.require('viewer') + comms.registry.set_commander_project(RegisteredCommanderProject(commander.incarnation, ())) + output = bridge.output_type.for_comms(comms.transcripts.external('viewer'), comms) + session = ClaudeSession() + output.publish_batch(list(opening_record('previous-turn').updates(session))) + finish_reply(output, session) + for record in prefix: + output.publish_batch(list(record.updates(session))) + directives = [] + for number in range(12): + request = comms.goals.todos.add_request(commander.incarnation, f'Independent request {number}') + ConnectorSendTool(target='viewer', request=request.id, + body=f'Explain independent request {number}.').apply( + comms, DirectorCaller('claude-code')) + directive, = comms.goals.todos.owner_request(request.id).directives + with comms.bus.bus.reading() as source: + directives.append(directive.original(source)) + originals = tuple(directives) + assert {rule.message.reference for rule in DirectiveUnacted.opened(comms, 'viewer')} == { + message.reference for message in originals} + await confirm_messages(bridge, output, session, originals, prompt_id='directive-turn') + before = comms.bus.bus.revision() + for record in suffix: + output.publish_batch(list(record.updates(session))) + expected = set() if cleared else {message.reference for message in originals} + assert {rule.message.reference for rule in DirectiveUnacted.opened(comms, 'viewer')} == expected + assert comms.bus.bus.revision() == before + for record in (opening_record('later-source'), assistant_record('later-final', 'A different turn answers')): + output.publish_batch(list(record.updates(session))) + output.publish_batch([FinishedTurn(NativeTurnId('later-source'))]) + assert {rule.message.reference for rule in DirectiveUnacted.opened(comms, 'viewer')} == expected + assert comms.bus.bus.unanswered_references( + tuple(message.seq for message in originals), frozenset({'viewer'})) == tuple( + message.reference for message in originals) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('ending', (AllTurnsEnded(), AllTurnsCancelled()), ids=('finish', 'cancel')) +async def test_retirement_visits_all_live_windows_not_the_newest_row( + bridge: ClaudeBridge, ending: RuntimeUpdate, +) -> None: + output = bridge.output_type.for_comms(bridge.comms.transcripts.external('viewer'), bridge.comms) + session = ClaudeSession() + for prompt_id in ('older-open', 'newer-open'): + question = bridge.comms.messaging.send('peer', 'viewer', prompt_id, QuestionKind).message + await confirm_messages(bridge, output, session, (question,), prompt_id=prompt_id) + assert output.transcript.opening_for(NativeTurnId(prompt_id)) == (question.reference,) + output.publish_batch([ending]) + assert output.transcript.opening_for(NativeTurnId('older-open')) == () + assert output.transcript.opening_for(NativeTurnId('newer-open')) == () + + +@pytest.mark.asyncio +async def test_same_prompt_id_wake_after_retirement_refuses_instead_of_dropping( + bridge: ClaudeBridge, +) -> None: + output = bridge.output_type.for_comms(bridge.comms.transcripts.external('viewer'), bridge.comms) + session = ClaudeSession() + first = bridge.comms.messaging.send('peer', 'viewer', 'First source turn').message + await confirm_messages(bridge, output, session, (first,), prompt_id='reused-prompt') + finish_reply(output, session) + later = bridge.comms.messaging.send('peer', 'viewer', 'New wake with reused prompt', QuestionKind).message + with pytest.raises(RelationViolationError, match='reused-prompt'): + await confirm_messages(bridge, output, session, (later,), prompt_id='reused-prompt') + assert output.transcript.opening_for(NativeTurnId('reused-prompt')) == () + assert bridge.comms.bus.bus.unanswered_references((later.seq,), frozenset({'viewer'})) == (later.reference,) diff --git a/tests/test_codex_earlier_pages.py b/tests/test_codex_earlier_pages.py index 84a0c507f8..0adf4c91b6 100644 --- a/tests/test_codex_earlier_pages.py +++ b/tests/test_codex_earlier_pages.py @@ -8,6 +8,8 @@ from aiohttp import web from agent_comms.cli_bridges.codex import CodexBridge, CodexOptions +from agent_comms.cli_bridges.codex_protocol import BusClientId +from agent_comms.field_codec import MessageId from agent_comms.cli_bridges.shared import Binding, StoredBinding, Watch from agent_comms.cli_protocol import WATCH_LINE, EarlierItems from agent_comms.comms import wire @@ -26,8 +28,12 @@ def entry(name, timestamp): async def test_pager_requests_two_older_pages_through_the_bridges_real_socket(tmp_path): seen = [] + historical_id = MessageId.mint('old-delivery-with-no-retained-attempt') + historical_input = dict(turnId='old-turn', completedAtMs=4500, item=dict( + type='userMessage', id='old-bus-input', clientId=BusClientId(historical_id).external, + content=[dict(type='text', text='An old rendering, not the current IncomingFrame')])) pages = { - None: {'data': [entry('recent2', 5000), entry('recent1', 4000)], 'nextCursor': 'p1'}, + None: {'data': [entry('recent2', 5000), historical_input, entry('recent1', 4000)], 'nextCursor': 'p1'}, 'p1': {'data': [entry('older2', 3000), entry('older1', 2000)], 'nextCursor': 'p2'}, 'p2': {'data': [entry('oldest', 1000)], 'nextCursor': None}, } @@ -74,7 +80,8 @@ async def serve(request): bridge = CodexBridge(options, binding.address(), db) await options.operation.prepare(bridge) output = ThreadOutput(transcript, comms.agents) - output.publish_batch([update async for update in bridge.history()]) + output.publish_batch([await update.observed(bridge) async for update in bridge.history()]) + assert await bridge.attempts(historical_id) == () assert seen == [None] newest = comms.transcripts.capture_page_read('cli').read() assert [event.text for event in newest.events] == ['recent1', 'recent2'] @@ -108,7 +115,10 @@ async def serve(request): 'oldest', 'older1', 'older2', 'recent1', 'recent2'] with transcript.snapshot() as connection: positions = ExternalItemRow.select(connection, thread=transcript.lookup, order_by=('seq',)) - assert [row.seq for row in positions] == [-2, -1, 0, 1, 2] + assert [row.seq for row in positions] == [-2, -1, 0, 1, 2, 3] + # Resume passes historical inputs through PublishedPage, never live confirmation. + output.publish_batch([await update.observed(bridge) async for update in bridge.history()]) + assert await bridge.attempts(historical_id) == () finally: watching.cancel() with suppress(asyncio.CancelledError): @@ -152,7 +162,7 @@ def test_hidden_page_items_keep_import_events_but_own_no_visible_events(tmp_path page = ItemsPage.model_validate({'data': [{ 'turnId': 't', 'completedAtMs': 1000, 'item': {'type': 'userMessage', 'id': 'bus', - 'clientId': BusInput.for_message('thread', 'm1', 'input').clientUserMessageId, + 'clientId': BusInput.for_message('thread', MessageId.mint('hidden-page-input'), 'input').clientUserMessageId, 'content': [{'type': 'text', 'text': 'Already on the bus'}]}, }], 'nextCursor': ''}).publication() item, = page.items diff --git a/tests/test_codex_items.py b/tests/test_codex_items.py index 652078caa1..33da550fa0 100644 --- a/tests/test_codex_items.py +++ b/tests/test_codex_items.py @@ -1,6 +1,11 @@ from agent_comms.repo import Checkout -from agent_comms.cli_bridges.codex_protocol import BusInput, ItemsPage +import pytest + +from agent_comms.cli_bridges.codex_protocol import ( + BusClientId, BusInput, ForeignClientId, ItemsPage, NoClientId, UserItem, +) +from agent_comms.field_codec import MessageId from agent_comms.transcript_events import UserTranscript from agent_comms.threads import Thread from agent_comms.turn_item import NativeItemId, NativeTurnId @@ -35,10 +40,26 @@ def test_items_page_decodes_every_declared_user_input(): def test_bus_input_encodes_its_text_with_the_codex_tag(): - params = BusInput.for_message("thread", "m1", "hi").model_dump(mode="json") + params = BusInput.for_message("thread", MessageId.mint("test-bus-input"), "hi").model_dump(mode="json") assert params["input"] == [{"type": "text", "text": "hi"}] +@pytest.mark.parametrize(('wire', 'decoded'), ( + (None, NoClientId()), + ('foreign-client', ForeignClientId('foreign-client')), + ('', ForeignClientId('')), + (BusClientId(MessageId.mint('client-round-trip')).external, + BusClientId(MessageId.mint('client-round-trip'))), +)) +def test_client_id_boundary_retains_its_member_through_item_use(wire, decoded): + item = UserItem.model_validate(dict(id='input', content=[dict(type='text', text='body')], clientId=wire)) + assert item.clientId == decoded + assert item.model_dump(mode='json')['clientId'] == wire + member = item.clientId + item.completed(NativeTurnId('source-turn'), item.events(1.0)) + assert item.clientId is member + + def test_codex_stream_reports_activity_and_finishes_items(tmp_path): t1_turn_id = NativeTurnId('t1') from agent_comms.turn_phase import ModelWaitPhase, ToolRunningPhase diff --git a/tests/test_importing.py b/tests/test_importing.py index 9fd4fe5f0f..ee5d689f8a 100644 --- a/tests/test_importing.py +++ b/tests/test_importing.py @@ -29,7 +29,7 @@ from agent_comms.cli_bridges.claude_transcript import ClaudeTranscript, RunningClaude from agent_comms.cli_bridges.codex import CodexAppServer from agent_comms.cli_bridges.codex_connection import CodexConnection -from agent_comms.cli_bridges.codex_protocol import InitializeResponse, RequestParams, ItemsList +from agent_comms.cli_bridges.codex_protocol import BusInput, InitializeResponse, RequestParams, ItemsList from agent_comms.cli_bridges.codex_rollout import ROLLOUT_RECORD, ForeignItem, ForeignRecord, ResponseItemRecord from agent_comms.cli_bridges.history import ( AnySession, DefaultAddress, NamedSession, OfflineHistory, OnlineHistory, @@ -38,7 +38,7 @@ from agent_comms.cli_bridges.opencode_parts import SavedMessage from agent_comms.cli_bridges.opencode_protocol import Session from agent_comms.cli_bridges.opencode_store import OpenCodeDatabase, OpenCodeExport -from agent_comms.field_codec import FieldCodec +from agent_comms.field_codec import FieldCodec, MessageId from agent_comms.harness_install import ClaudeCodeHarness, CodexHarness, OpenCodeHarness, PiHarness from agent_comms.importing import ImportBuffer, ThreadImported, ImportSnapshot, ImportedSessionMetadata from agent_comms.native_entries import NativeEntry, ImportedMetadataEntry @@ -630,9 +630,7 @@ def codex_entry(turn: str, item: dict[str, JsonValue]) -> dict[str, JsonValue]: def test_codex_app_server_reads_only_newest_page_and_keeps_bus_inputs() -> None: - from agent_comms.cli_bridges.codex_protocol import BusInput - - bus = BusInput.for_message("thread", "m1", "from the bus") + bus = BusInput.for_message("thread", MessageId.mint("imported-bus-input"), "from the bus") newest: dict[str, JsonValue] = {"data": [ codex_entry("t2", {"type": "agentMessage", "id": "i3", "text": "second answer"}), codex_entry("t2", {"type": "userMessage", "id": "i1", "clientId": bus.clientUserMessageId, diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index 12518c2c05..57218847f0 100644 --- a/tests/test_monitor_delivery.py +++ b/tests/test_monitor_delivery.py @@ -401,10 +401,10 @@ def decoded_record(data): return ClaudeJsonLines().decode_line(json.dumps(data).encode(), 0) -def notification(line, record_id=transcript_record('notification-record')): +def notification(line, record_id=transcript_record('notification-record'), *, prompt_id='prompt'): return decoded_record(dict( type='user', uuid=record_id, timestamp='2026-10-16T16:40:05.123Z', - sessionId='session', promptId='prompt', origin=dict(kind='task-notification'), + sessionId='session', promptId=prompt_id, origin=dict(kind='task-notification'), message=dict(content=''+escape(line)+ ''))) @@ -1031,14 +1031,16 @@ def test_only_commanded_sessions_reply_to_their_delivered_senders(comms): comms.registry.set_commander_project(RegisteredCommanderProject(lead.incarnation, ())) for speaker, sender in ((lead, unit), (unit, lead)): output = ReplyingThreadOutput.for_comms(comms.transcripts.external(speaker.name), comms) - output.delivered_targets((sender.name,)) + original = comms.messaging.send(sender.name, speaker.name, "Open this answer").message turn = NativeTurnId(f"{speaker.name}-turn") + output.delivered(turn, (original.reference,)) item_id = NativeItemId("a") output.publish_item(FinishedItem(turn, item_id, (AssistantTranscript( f"{speaker.name} answers", timestamp=1.0, position=item_id),))) output.end_turn(turn) - assert comms.bus.pending_count("unit") == 0 - assert comms.bus.pending_count("lead") == 1 + replies = tuple((message.target, message.body) for message in comms.bus.bus.message_history() + if message.body in {"lead answers", "unit answers"}) + assert replies == (("lead", "unit answers"),) @pytest.mark.parametrize(('rows', 'start', 'frames'), ( diff --git a/tests/test_native_evidence_custody.py b/tests/test_native_evidence_custody.py index 1978f924fe..7c12f4e04e 100644 --- a/tests/test_native_evidence_custody.py +++ b/tests/test_native_evidence_custody.py @@ -65,9 +65,9 @@ def test_acquired_input_source_reverifies_rewritten_file_from_start(tmp_path, re .replace(INPUT_ID.encode(), restored_input.encode()) .replace(b"separate", b"restored")) _header, original = NativeTranscript(session).forward(0, session.stat().st_size) - location = NativeInputLocation(INPUT_ID, "sid", "entry", session_entry_start=original.start) + location = NativeInputLocation(INPUT_ID, "sid", original.entry.require_entry_id(), session_entry_start=original.start) restored_location = NativeInputLocation( - restored_input, "new", "entry", session_entry_start=original.start, + restored_input, "new", location.session_entry_id, session_entry_start=original.start, ) with NativeEvidenceScope() as scope: evidence = scope.for_source(session) diff --git a/tests/test_native_pi.py b/tests/test_native_pi.py index 4b735a4679..e3ebdd6a9b 100644 --- a/tests/test_native_pi.py +++ b/tests/test_native_pi.py @@ -235,7 +235,7 @@ def test_provider_failure_notice_does_not_broadcast_untrusted_error_body(tmp_pat files = _evidence(tmp_path) session = files.session _header, record = NativeTranscript(session).forward(0, session.stat().st_size) - context = NativeContextProof(INPUT_ID,'sid','entry',1,DIGEST,1_791_964_800_001,session, + context = NativeContextProof(INPUT_ID,'sid',record.entry.require_entry_id(),1,DIGEST,1_791_964_800_001,session, session_entry_start=record.start) failed = NativePiTerminalFailure('SECRET provider response including prompt data', context, SelectedModel('provider','model'), ModelRequestFailed('SECRET provider response including prompt data')) @@ -264,7 +264,7 @@ def test_recorded_context_remains_verifiable_after_later_tool_rounds(tmp_path): with NativeEntry.open_evidence(session) as evidence: assert recorded.corroborates_input(INPUT_ID, files=files, evidence=evidence) assert not recorded.corroborates_input('e'*32, files=files, evidence=evidence) - input_event = InputCommitted(input_id=INPUT_ID,session_id='sid',session_entry_id='entry',session_entry_start=record.start) + input_event = InputCommitted(input_id=INPUT_ID,session_id='sid',session_entry_id=record.entry.require_entry_id(),session_entry_start=record.start) context_event = ContextCommitted.from_wire({key:value for key,value in first.items() if key != 'schema'}) assert _verify_context(files,INPUT_ID,'sid',input_event,context_event) == recorded with pytest.raises(NativePiUnavailable): diff --git a/tests/test_native_transcript.py b/tests/test_native_transcript.py index 6a62babb47..11db39a0a8 100644 --- a/tests/test_native_transcript.py +++ b/tests/test_native_transcript.py @@ -2,16 +2,20 @@ import json from dataclasses import fields +from pathlib import Path import pytest from agent_comms.field_codec import ( FieldCodec, + InputId, MalformedRecord, ToolCallId, ) from agent_comms.native_entries import MessageEntry, NativeEntry, TranscriptProjection, UnknownEntry from agent_comms.native_transcript import NativeTranscript +from agent_comms.native_pi import NativeContextProof +from agent_comms.native_session_reopen import PiSessionId from agent_comms.pi_payloads import UserMessage from agent_comms.transcript_events import ( AssistantTranscript, @@ -22,7 +26,7 @@ TranscriptEvent, UserTranscript, ) -from agent_comms.turn_item import NativeItemId +from agent_comms.turn_item import FreshNativeId, NativeItemId from conftest import transcript_records STAMPED = 1790812800.0 # "2026-10-01T00:00:00.000Z", the time every fixture row records @@ -137,3 +141,39 @@ def test_large_record_retained_in_tail_and_both_page_directions(tmp_path): assert list(transcript.tail(max_bytes=512)) == [forward[-1].entry] +@pytest.mark.parametrize(('stop_reason', 'body', 'parent', 'next_input', 'complete', 'answered'), ( + ('stop', 'A committed final', 'opening-input', False, True, True), + ('toolUse', 'Before a tool', 'opening-input', False, True, False), + ('aborted', 'Interrupted output', 'opening-input', False, True, False), + ('error', 'Failed output', 'opening-input', False, True, False), + ('stop', '', 'opening-input', False, True, False), + ('stop', 'Another branch', 'older-input', False, True, False), + ('stop', 'Another turn', 'next-input', True, True, False), + ('stop', 'An unfinished record', 'opening-input', False, False, False), +)) +def test_native_final_stays_inside_its_proved_input_and_ancestry( + tmp_path: Path, stop_reason: str, body: str, parent: str, next_input: bool, complete: bool, answered: bool, +) -> None: + path = tmp_path / 'answer.jsonl' + session_id = PiSessionId.fresh() + input_id = InputId.mint() + prefix = encoded(dict(type='session', version=3, id=session_id, cwd=str(tmp_path), + timestamp='2026-10-01T00:00:00.000Z')) + # Neither the forward answer read nor its bounded ancestry walk may decode old history. + prefix += b'{an unrelated older row that is outside this input}\n' + opening = encoded(dict(type='message', id='opening-input', parentId='older-input', + timestamp='2026-10-01T00:00:00.000Z', message=dict(role='user', content='The opening question', + inputId=input_id, inputDigest='b' * 64))) + following = encoded(dict(type='message', id='next-input', parentId='opening-input', + timestamp='2026-10-01T00:00:00.000Z', message=dict(role='user', content='A different question'))) + final = encoded(dict(type='message', id='answer', parentId=parent, + timestamp='2026-10-01T00:00:00.000Z', message=dict(role='assistant', + content=[dict(type='text', text=body)], stopReason=stop_reason))) + path.write_bytes(prefix + opening + (following if next_input else b'') + + (final if complete else final.removesuffix(b'\n'))) + context = NativeContextProof(input_id=input_id, session_id=session_id, + session_entry_id=FreshNativeId('opening-input'), request_generation=1, llm_context_digest='a' * 64, + committed_at_ms=1761699600000, session_file=path, session_entry_start=len(prefix)) + assert NativeTranscript(path).completed_input(context) is answered + +