Reply only to DMs that opened the Claude answer, not mid-turn arrivals

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]:

src/agent_comms/cli_bridges/claude_protocol.py

@@ -1,3 +1,3 @@
 """Claude's completed JSONL blocks own transcript items and activity."""
 from __future__ import annotations
 
@@ -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 @@
         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 @@
 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 @@
                 *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 @@
         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,15 +446,18 @@
 
 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'):
 
     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 @@
 
     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 @@
     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 @@
 
     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 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 @@
         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 @@
             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 @@
 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 @@
     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 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 @@
         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 @@
 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):