Keep narration out of answers; repair the reply tests the policy outran

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,15 +500,18 @@ 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'):
 
     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=""):

src/agent_comms/transcript_events.py

@@ -1,3 +1,3 @@
 """Saved presentation facts: each declaration carries only its valid fields."""
 
 from __future__ import annotations
@@ -264,6 +264,15 @@
 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 @@
         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 @@
         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.

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(

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)
 

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")