From 7ceee1e58be25a3139a0cb8b4ce1ffc1d9897a07 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 31 Oct 2026 02:07:23 -0400 Subject: [PATCH 1/3] Keep narration out of answers; repair the reply tests the policy outran Narration before a tool call was stored as an AssistantTranscript, so it counted as an answer and could finish a record twice. Carries closed #3594's ad6a75278e: narration is NarrationTranscript, and each record finishes once. The 13 reds on main were stale tests, not lost replies: - Agent info messages get no automatic reply since 6964ac4d9e; the commanded test now asks a question. - Only the turn a delivery opened answers it since 1973a5d765; the reply helper now finishes that turn. - The bridge fixture declared viewer native; production bridges are external. - #monitor was never created. A running requester can't be deleted; it stops first. - StoredBinding returns the stored Binding, not its address. - The concurrency worker crashed on a duplicate factory argument before reaching the store. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XbFMSpSd8zQxyozgEV2Re7 --- .../cli_bridges/claude_protocol.py | 33 +++++++++------ src/agent_comms/transcript_events.py | 27 ++++++++----- tests/test_claude_automatic_reply.py | 14 ++++++- tests/test_cli_error_family.py | 11 ++--- tests/test_monitor_delivery.py | 40 +++++++++++-------- 5 files changed, 79 insertions(+), 46 deletions(-) diff --git a/src/agent_comms/cli_bridges/claude_protocol.py b/src/agent_comms/cli_bridges/claude_protocol.py index f2f284bdc1..f9f81e3f66 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,8 @@ 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) - 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'): @@ -567,8 +571,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..078f2ca0d1 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 admission 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 4d21a6db7d..f7d0ee02c3 100644 --- a/tests/test_claude_automatic_reply.py +++ b/tests/test_claude_automatic_reply.py @@ -265,12 +265,22 @@ 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), ((), (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', '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: @@ -408,7 +418,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 3e7ff3c7e0..aae738a01f 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))) @@ -316,7 +320,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))) @@ -885,7 +888,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))) @@ -967,7 +969,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))) @@ -1019,13 +1020,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( @@ -1052,18 +1054,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 @@ -1086,6 +1094,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) @@ -1103,7 +1112,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 @@ -1112,7 +1120,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") From f2795a4c5d295727765acebcc15489b161546a62 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 31 Oct 2026 02:12:15 -0400 Subject: [PATCH 2/3] Treat empty saved thinking as hidden thinking, not as narration Claude Code usually saves reasoning under this tag with empty text. The conflict resolution dropped the old guard, so every such block finished an empty NarrationTranscript. It now takes the parent's hidden-thinking path. Caught by claude-board. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XbFMSpSd8zQxyozgEV2Re7 --- src/agent_comms/cli_bridges/claude_protocol.py | 2 ++ tests/test_claude_automatic_reply.py | 6 +++++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/agent_comms/cli_bridges/claude_protocol.py b/src/agent_comms/cli_bridges/claude_protocol.py index f9f81e3f66..c0dcd0db3a 100644 --- a/src/agent_comms/cli_bridges/claude_protocol.py +++ b/src/agent_comms/cli_bridges/claude_protocol.py @@ -525,6 +525,8 @@ def said(self) -> tuple[str, ...]: return (self.thinking,) if self.thinking else () def events(self, record: TimedRecord) -> tuple[TranscriptEvent, ...]: + if not self.thinking: + return super().events(record) return (NarrationTranscript(self.thinking, timestamp=record.timestamp, position=record.position),) diff --git a/tests/test_claude_automatic_reply.py b/tests/test_claude_automatic_reply.py index a864d09d8e..480b264b61 100644 --- a/tests/test_claude_automatic_reply.py +++ b/tests/test_claude_automatic_reply.py @@ -303,10 +303,14 @@ async def deliver(original: Message) -> None: 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', 'narration-with-final', 'still-open', 'mid-output', - 'hidden-thinking-before-input', 'interrupted', 'empty-final')) + '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: From 73e60cd03db317bd1e810fb0f3ce7400993fb9f8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 31 Oct 2026 02:20:17 -0400 Subject: [PATCH 3/3] Say what narration does instead of naming a banned word Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01XbFMSpSd8zQxyozgEV2Re7 --- src/agent_comms/transcript_events.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/agent_comms/transcript_events.py b/src/agent_comms/transcript_events.py index 078f2ca0d1..e0d5c4d302 100644 --- a/src/agent_comms/transcript_events.py +++ b/src/agent_comms/transcript_events.py @@ -373,7 +373,7 @@ def reply_window[Window](self, window: Window, answering: Callable[[], Window]) class NarrationTranscript(ModelOutput, AgentTextTranscript): - """Visible assistant progress closes admission but is not an answer or DM final.""" + """Visible assistant progress closes the window for new automatic replies but is not an answer or DM final.""" class AssistantTranscript(ModelOutput, AgentTextTranscript):