diff --git a/src/agent_comms/cli_bridges/claude_protocol.py b/src/agent_comms/cli_bridges/claude_protocol.py index f2f284bdc1..c0dcd0db3a 100644 --- a/src/agent_comms/cli_bridges/claude_protocol.py +++ b/src/agent_comms/cli_bridges/claude_protocol.py @@ -61,6 +61,8 @@ from agent_comms.tool_calls import CliToolArguments, ToolCall from agent_comms.transcript_events import ( AssistantTranscript, + NarrationTranscript, + ThinkingTranscript, NoticeTranscript, TranscriptEvent, UserTranscript, @@ -468,10 +470,13 @@ def said(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 () def said(self) -> tuple[str, ...]: """The words this block shows the reader; a tool call or hidden reasoning shows none.""" @@ -495,8 +500,8 @@ class AssistantText(TextWire, AssistantBlock, TextContent, declared_name='text') def said(self) -> tuple[str, ...]: return (self.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'): @@ -504,6 +509,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 @@ -516,12 +524,10 @@ def said(self) -> tuple[str, ...]: """Saved reasoning is empty; narration is what the reader saw.""" return (self.thinking,) if self.thinking else () - def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: - activity = super().updates(record, session) + def events(self, record: TimedRecord) -> tuple[TranscriptEvent, ...]: if not self.thinking: - return activity - return (*activity, *session.finished( - record.uuid, (AssistantTranscript(self.thinking, timestamp=record.timestamp, position=record.position),))) + return super().events(record) + return (NarrationTranscript(self.thinking, timestamp=record.timestamp, position=record.position),) class ToolUse(AssistantBlock, declared_name='tool_use'): @@ -567,8 +573,11 @@ def said(self) -> tuple[str, ...]: return tuple(words for block in self.message.content for words in block.said()) 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=""): diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index 6cfc031b51..e0d5c4d302 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -264,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): @@ -363,6 +372,10 @@ def reply_window[Window](self, window: Window, answering: Callable[[], Window]) return answering() +class NarrationTranscript(ModelOutput, AgentTextTranscript): + """Visible assistant progress closes the window for new automatic replies but is not an answer or DM final.""" + + class AssistantTranscript(ModelOutput, AgentTextTranscript): @property def answers_turn(self) -> bool: @@ -391,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. diff --git a/tests/test_claude_automatic_reply.py b/tests/test_claude_automatic_reply.py index d49a53991a..480b264b61 100644 --- a/tests/test_claude_automatic_reply.py +++ b/tests/test_claude_automatic_reply.py @@ -291,12 +291,26 @@ async def deliver(original: Message) -> None: 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), + ((opening_record('directive-turn'), decoded_record(dict( + type='assistant', uuid=transcript_record('empty-thinking'), timestamp='2026-10-29T01:00:00Z', + message=dict(content=[dict(type='thinking', thinking='')]), + ))), (assistant_record('after-empty-thinking', 'Answering the older input'), TURN_FINISHED), False), ((), (assistant_record('interrupted-output', 'A partial answer'), TURN_INTERRUPTED), False), ((), (assistant_record('empty-final', ''), TURN_FINISHED), False), -), ids=('committed-final', 'thinking-only', 'still-open', 'mid-output', 'interrupted', 'empty-final')) +), ids=('committed-final', 'thinking-only', 'narration-with-final', 'still-open', 'mid-output', + 'hidden-thinking-before-input', 'empty-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: @@ -434,7 +448,7 @@ async def test_original_kind_owns_automatic_answer_without_suppressing_the_turn( @pytest.mark.asyncio @pytest.mark.parametrize(('transition', 'targets'), ( (Registration.archive, ()), - (Registration.begin_delete, ()), + (lambda registry, name: (registry.unregister(name), registry.begin_delete(name)), ()), (Registration.status, ('claude-code',)), ), ids=('archived', 'deleting', 'visible')) async def test_requester_presence_keeps_the_final_and_the_next_delivery( diff --git a/tests/test_cli_error_family.py b/tests/test_cli_error_family.py index 3dfaea4629..db47816791 100644 --- a/tests/test_cli_error_family.py +++ b/tests/test_cli_error_family.py @@ -69,9 +69,10 @@ async def test_journal_concurrent_initialization_and_reopen_preserve_rows(tmp_pa from pathlib import Path from unittest.mock import patch from agent_comms.cli_bridges.shared import Binding +from agent_comms.store_files import LendingConnection connect = sqlite3.connect -class PausedConnection(sqlite3.Connection): +class PausedConnection(LendingConnection): first_create = True def execute(self, statement, *args, **kwargs): if self.first_create and statement.startswith('CREATE TABLE '): @@ -80,7 +81,7 @@ def execute(self, statement, *args, **kwargs): assert sys.stdin.readline() == 'go\n' return super().execute(statement, *args, **kwargs) -with patch('sqlite3.connect', side_effect=lambda *a, **k: connect(*a, factory=PausedConnection, **k)): +with patch('sqlite3.connect', side_effect=lambda *a, **k: connect(*a, **{**k, 'factory': PausedConnection})): with Binding.open(Path(sys.argv[1])): pass print('opened', flush=True) @@ -116,7 +117,7 @@ async def open_together(*, fresh): assert binding.bind(db) == binding await open_together(fresh=False) with Binding.open(tmp_path) as db: - assert StoredBinding().bind(db) == binding.address() + assert StoredBinding().bind(db) == binding @pytest.mark.parametrize('conflicting', [False, True]) @@ -134,7 +135,7 @@ def test_concurrent_initial_binding_preserves_one_identity( with Binding.open(tmp_path) as db: winner, = Binding.select(db) assert winner in (first, other) - assert StoredBinding().bind(db) == winner.address() + assert StoredBinding().bind(db) == winner assert winner.bind(db) == winner loser = other if winner == first else first with pytest.raises(ValueError): @@ -151,7 +152,7 @@ def test_existing_binding_lookup_does_not_acquire_a_writer(tmp_path, bus): db.set_trace_callback(statements.append) with closing(SQLiteFile.writer(tmp_path / 'delivery.sqlite3')) as writer: writer.execute('BEGIN IMMEDIATE') - assert StoredBinding().bind(db) == binding.address() + assert StoredBinding().bind(db) == binding assert writer.in_transaction assert not any(sql.startswith(('BEGIN', 'COMMIT', 'ROLLBACK')) for sql in statements) diff --git a/tests/test_monitor_delivery.py b/tests/test_monitor_delivery.py index 2cd850e13c..49c2975940 100644 --- a/tests/test_monitor_delivery.py +++ b/tests/test_monitor_delivery.py @@ -54,6 +54,8 @@ AllTurnsEnded, FinishedItem, FinishedTurn, HiddenItem, RuntimeUpdate, ) from agent_comms.turn_item import NativeItemId, NativeTurnId +from agent_comms.external_transcript import ExternalTurnInput +from agent_comms.thread_execution import ExternalThreadExecution from agent_comms.messages import AlertKind, InfoKind, Message, NoticeRole, QuestionKind from agent_comms.channel_targets import ThreadTarget from agent_comms.diagnostics import record_terminal_failure @@ -131,9 +133,11 @@ async def test_empty_journal_starts_at_binding_and_ignores_user_reads(comms, tmp @pytest.fixture def bridge(comms, tmp_path): - declare_test_bus(comms, threads=tuple( - Thread(name, frozenset(), Checkout(tmp_path)) - for name in ('claude-code', 'viewer', 'peer') + declare_test_bus(comms, threads=( + Thread('claude-code', frozenset(), Checkout(tmp_path)), + # A production bridge's session is external: its answers live in the external store. + Thread('viewer', frozenset(), Checkout(tmp_path), execution=ExternalThreadExecution), + Thread('peer', frozenset(), Checkout(tmp_path)), )) practice = PracticeCopy(comms.root, tmp_path / 'copied-bus') practice.include(tuple(RootFile.existing_in(practice.root))) @@ -335,7 +339,6 @@ async def test_renamed_reader_handoff_and_confirmation_share_the_prepared_owner( from agent_comms.coordination_database import CoordinationStore from agent_comms.coordination_tables.assignments import WakeAssignment from agent_comms.coordinator import Coordination - from agent_comms.thread_execution import ExternalThreadExecution comms = prepare_native_comms(wire(tmp_path / 'wire')) comms.registry.declare(Thread('peer', frozenset(), Checkout(tmp_path))) @@ -904,7 +907,6 @@ async def test_harness_outcome_lives_on_the_threads_claim_across_a_rehome(tmp_pa from agent_comms.coordination_database import CoordinationStore from agent_comms.coordination_tables.assignments import WakeAssignment from agent_comms.coordinator import Coordination - from agent_comms.thread_execution import ExternalThreadExecution comms = prepare_native_comms(wire(tmp_path / 'wire')) comms.registry.declare(Thread('peer', frozenset(), Checkout(tmp_path))) @@ -986,7 +988,6 @@ async def test_a_bridge_waits_out_maintenance_and_then_records_its_delivery(tmp_ from agent_comms.coordination_tables.assignments import WakeAssignment from agent_comms.coordinator import Coordination from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator - from agent_comms.thread_execution import ExternalThreadExecution comms = prepare_native_comms(wire(tmp_path / 'wire')) comms.registry.declare(Thread('peer', frozenset(), Checkout(tmp_path))) @@ -1038,13 +1039,14 @@ async def observed(): await asyncio.wait_for(pending, 5) assert type(claim()) is DeliveredAssignment -@pytest.fixture(params=(FinishedTurn(NativeTurnId('reply-turn')), AllTurnsEnded())) -def reply_end(request: pytest.FixtureRequest) -> RuntimeUpdate: +@pytest.fixture(params=(FinishedTurn, lambda _turn: AllTurnsEnded())) +def reply_end(request: pytest.FixtureRequest) -> Callable[[NativeTurnId], RuntimeUpdate]: return request.param async def confirm_and_finish_reply( - bridge: ClaudeBridge, message: Message, expected: tuple[str, ...], end: RuntimeUpdate, + bridge: ClaudeBridge, message: Message, expected: tuple[str, ...], + end: Callable[[NativeTurnId], RuntimeUpdate], ) -> None: """Real monitor consumption feeds the ordinary staged output and terminal update.""" output = bridge.output_type.for_comms( @@ -1071,18 +1073,24 @@ def replies() -> tuple[tuple[str, str], ...]: if row.sender == bridge.binding.bus) output.publish_batch(updates) - turn, item = NativeTurnId('reply-turn'), NativeItemId('answer') + # Only the turn the delivery opened answers it. + with output.transcript.snapshot() as connection: + opened = ExternalTurnInput.select_keys(connection, ( + dict(thread=output.transcript.lookup, reference=message.reference),)) + # A delivery that opens no turn leaves any other turn's final with nothing to answer. + turn = next((row.turn_id for row in opened), NativeTurnId('unopened-turn')) + item = NativeItemId('answer') output.publish_batch([FinishedItem(turn, item, (AssistantTranscript( - 'unrelated assistant final', timestamp=2., position=item),))]) + 'the opened turn final', timestamp=2., position=item),))]) assert replies() == () - output.publish_batch([end]) - assert replies() == tuple((target, 'unrelated assistant final') for target in expected) + output.publish_batch([end(turn)]) + assert replies() == tuple((target, 'the opened turn final') for target in expected) # Completion is idempotent, and another turn cannot inherit these routes. later, later_item = NativeTurnId('later-turn'), NativeItemId('later-answer') output.publish_batch([FinishedTurn(turn), AllTurnsEnded(), FinishedItem( later, later_item, (AssistantTranscript( 'a later final', timestamp=3., position=later_item),)), AllTurnsEnded()]) - assert replies() == tuple((target, 'unrelated assistant final') for target in expected) + assert replies() == tuple((target, 'the opened turn final') for target in expected) @pytest.mark.asyncio @@ -1105,6 +1113,7 @@ async def test_diagnostic_notice_is_consumed_without_replying_to_the_failed_send async def test_confirmed_reply_routes_follow_direct_and_channel_policy( bridge, reply_end, target, body, kind, expected, ): + bridge.comms.channels.create_tag('monitor') message = bridge.comms.messaging.send('claude-code', target, body, kind).message await confirm_and_finish_reply(bridge, message, expected, reply_end) @@ -1122,7 +1131,6 @@ def test_only_commanded_sessions_reply_to_their_delivered_senders(comms): mid-turn never receives that turn's answer as a direct message.""" from agent_comms.registry_document import RegisteredCommanderProject from agent_comms.runtime_updates import FinishedItem, ReplyingThreadOutput - from agent_comms.thread_execution import ExternalThreadExecution from agent_comms.transcript_events import AssistantTranscript from agent_comms.turn_item import NativeItemId, NativeTurnId @@ -1131,7 +1139,7 @@ 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) - original = comms.messaging.send(sender.name, speaker.name, "Open this answer").message + original = comms.messaging.send(sender.name, speaker.name, "Open this answer", QuestionKind).message turn = NativeTurnId(f"{speaker.name}-turn") output.delivered(turn, (original.reference,)) item_id = NativeItemId("a")