P924: separate automatic answers from turn wake permission

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."""
@@ -465,27 +470,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.py

@@ -61,14 +61,13 @@ async def wake(self, message: Message) -> type[Outcome]:
         return await self.deliver(replace(message, body=self.board_wake))
 
     def history_start(self, stream: BinaryIO, boundary: int) -> int:
-        """On restart only the last 4 MiB is published again; Core's finish is idempotent, so
-        completed items may overlap. The first record starts after a newline in that window."""
+        """Retain the recent window's owning occurrence, never an arbitrary mid-turn prefix."""
         start = max(0, boundary - self.RECENT_BYTES)
         if start:
             stream.seek(start - 1)
             stream.readline()
             start = stream.tell()
-        return start
+        return self.opening_before(stream, start)
 
     async def transcript(self) -> AsyncIterator[RuntimeUpdate]:
         session = self.start_state()

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,
@@ -51,6 +51,7 @@
     StatusUpdate,
     Summarized,
     ToolDraft,
+    TurnEnded,
     TurnStarted,
 )
 from agent_comms.thread_identity import OfferedTools, ToolName, UnobservedTools
@@ -63,7 +64,7 @@
     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 +245,26 @@
         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."""
 
+    def opened(self, record: TimedRecord) -> ClaudeTurn:
+        """Further prompt records belong to the turn already running."""
+        return self
+
+    @property
+    def source(self) -> SourceTurn:
+        return NoSourceTurn()
+
     @abstractmethod
     def started(self, tool: ClaudeToolDraft,
                 phase: ClaudeReaderPhase) -> tuple[RuntimeUpdate, ...]: ...
@@ -269,6 +279,9 @@
 
 
 class UnknownTurn(ClaudeTurn):
+    def opened(self, record: TimedRecord) -> ClaudeTurn:
+        return KnownTurn(NativeTurnId(record.uuid))
+
     def started(self, tool: ClaudeToolDraft,
                 phase: ClaudeReaderPhase) -> tuple[RuntimeUpdate, ...]:
         return phase.unturned_finish()
@@ -286,6 +299,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,11 +367,15 @@
                 *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)
 
+    def retired(self, ending: TurnEnded) -> tuple[RuntimeUpdate, ...]:
+        self.turn = UnknownTurn()
+        return (ending,)
+
 
 class TimedRecord(TimestampedWire):
     position: int = Field(exclude=True)
@@ -403,6 +424,12 @@
         """Whether Claude replaced everything before this record (compaction)."""
         return False
 
+    def ends_turn(self) -> bool:
+        return False
+
+    def opening_position(self, following: int) -> int:
+        return following
+
     def request(self) -> str:
         """The prompt this record carries, read without the session around it."""
         return ''
@@ -590,7 +617,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'):
@@ -617,6 +644,9 @@
 
 class UserBlock(TaggedWire, ABC, affix=""):
 
+    def ends_turn(self) -> bool:
+        return False
+
     def updates(self, record: TimedRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]:
         return (TurnStarted(),)
 
@@ -629,9 +659,12 @@
     def prompt_text(self) -> str:
         return self.text
 
+    def ends_turn(self) -> bool:
+        return self.text.startswith('[Request interrupted by user')
+
     def updates(self, record: UserRecord, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]:
-        if self.text.startswith('[Request interrupted by user'):
-            return (AllTurnsEnded(),)
+        if self.ends_turn():
+            return session.retired(AllTurnsCancelled())
         return record.origin.prompt(record, self.text, session)
 
 
@@ -662,13 +695,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 +743,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 +759,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 +774,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 +785,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 +807,29 @@
         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):
+    """The first prompt after retirement owns the turn; promptId may be reused."""
+
+    promptId: NativeTurnId
+
+    def opening_position(self, following: int) -> int:
+        return self.position
 
     def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]:
-        session.turn = KnownTurn(FieldCodec.decode(NativeTurnId, self.promptId))
+        session.turn = session.turn.opened(self)
+        return self.opened_updates(session)
+
+    @abstractmethod
+    def opened_updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]: ...
+
+
+class UserRecord(PromptRecord, Originated, declared_name='user'):
+    message: UserMessage
+
+    def ends_turn(self) -> bool:
+        return any(block.ends_turn() for block in self.message.content)
+
+    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 +842,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):
@@ -815,6 +864,9 @@
 class SystemEvent(TaggedWire, ABC, affix=""):
     header = SubtypeHeader
 
+    def ends_turn(self) -> bool:
+        return False
+
     def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]:
         return ()
 
@@ -831,8 +883,11 @@
 
 class TurnDuration(SystemEvent, declared_name='turn_duration'):
 
+    def ends_turn(self) -> bool:
+        return True
+
     def updates(self, session: ClaudeSession) -> tuple[RuntimeUpdate, ...]:
-        return (AllTurnsEnded(),)
+        return session.retired(AllTurnsEnded())
 
 
 class CompactBoundary(SystemEvent, TimedRecord, declared_name='compact_boundary'):
@@ -888,6 +943,9 @@
     def begins_history(self) -> bool:
         return self.subtype.begins_history()
 
+    def ends_turn(self) -> bool:
+        return self.subtype.ends_turn()
+
 
 class MetadataRecord(ClaudeRecord, ABC, declared_name=None):
     pass

src/agent_comms/cli_bridges/claude_transcript.py

@@ -30,6 +30,16 @@ class ClaudeJsonLines(JsonLines):
     def decode_line(self, raw: bytes, position: int) -> ClaudeRecord:
         return self.records.validate_python({**json.loads(raw), 'position': position})
 
+    def opening_before(self, stream: BinaryIO, boundary: int) -> int:
+        """Recover the actual opening, including every prompt folded into that occurrence."""
+        opening = boundary
+        for offset, raw in self.reverse_lines(stream, boundary):
+            record = self.decode_line(raw, offset)
+            if record.ends_turn():
+                break
+            opening = record.opening_position(opening)
+        return opening
+
     def record_updates(self, record: ClaudeRecord, state: ClaudeSession, offset: int,
                        raw: bytes) -> Iterator[RuntimeUpdate]:
         yield from state.located(record.cwd)