Stamp derived projections and index answer eligibility at append

plans/commander-native-plan.md

@@ -211,6 +211,42 @@ The fix is one derived model with four renderers. **Core is the engine, Toad is
 
 **Done when** a commander runs a full session with no script process running, every signal the scripts raised appears from Core, and replaying one recorded hour of the live feed on a copy delivers to the commander only decision events, with no duplicates, and the same hour yields a `UnitTime` rollup whose classes add up to each unit's elapsed time.
 
+### Current control-plane work: heartbeat source identity (2026-10-23)
+
+Before this repair, `HeartbeatSilent.violations` read the newest row in the
+two-way DM conversation and then checked its sender. A commander reply after
+the last beat therefore hid a silent heartbeat. `BusReader.latest_addressed`
+now selects one candidate per exact service/commander alias pair through the
+existing typed table query, then decodes only the newest message. `Addressed`
+declares its compound sender/target/sequence index; registry provenance supplies
+the aliases. Ordinary DM display scopes retain their original contract.
+`CommanderHeartbeat.requested_by` derives
+live eligibility and the oldest open request from `OwnerRequestStore`; the shared
+silence scan derives its grace from that request and the latest actual beat.
+`HeartbeatNotStarted` and `MissedHeartbeat` own their distinct evidence and wording.
+A newer request, service registration, commander reply or rename cannot erase the
+older debt. No parallel timestamp, ping store or scheduler is added.
+
+### Reply eligibility: the message owns the answer
+
+`Message.role.response_policy` owns eligibility. SQL does not reimplement its
+recipient or response rules, and `MessageIdentitySource` does not retain a
+second policy identity. `Delivered.projections` materializes the direct-turn
+predicate once for each frozen recipient in `DirectReplySource`.
+`BusReader.direct_replies` uses that index with a bounded selection before
+loading deliveries. Installed Core commit stamps invalidate and rebuild these
+projections when their producing implementation changes.
+
+Reply pages retain exact original-reference matching, participant aliases and
+the original request's response eligibility. Latest reply sequence and greatest
+reply timestamp are separate selections. The query indexes must cover that
+existing scope before a SQL limit is applied; reading every original to decide
+eligibility is not the design.
+
+Schema-family carries preserve converter-owned `HistoryOffset` rows until
+`BusHistory.after_acceptance` retires them. Durable history is preserved;
+rebuildable projections follow their producing Core build.
+
 ### Phase 8: cutover
 
 **Changes.**

src/agent_comms/bus.py

@@ -5,9 +5,11 @@
 import json
 import sqlite3
 from abc import ABC, abstractmethod
-from collections.abc import Callable, Iterable, Iterator
+from collections.abc import Callable, Generator, Iterable, Iterator
 from contextlib import AbstractContextManager, ExitStack, closing, contextmanager
 from dataclasses import dataclass, replace
+from heapq import nlargest
+from itertools import islice, product
 from pathlib import Path
 from typing import TYPE_CHECKING, Self
 
@@ -21,8 +23,10 @@
     BusRow,
     BusSchema,
     ClaimOperations,
+    DirectReplySource,
     HeldResources,
     MessageIdentitySource,
+    ParticipantSource,
     ReceivedSource,
     SentSource,
 )
@@ -83,6 +87,22 @@ def matching(
             f'WHERE {addresses._sql(predicate)}', parameters))
 
 
+    @classmethod
+    def latest(
+        cls, db: sqlite3.Connection, senders: frozenset[str], targets: frozenset[str],
+    ) -> tuple[Message, ...]:
+        """One indexed candidate per alias pair; an empty selection stays empty."""
+        positions = nlargest(1, (
+            row.seq for sender, target in product(senders, targets)
+            for row in Addressed.table.select(
+                db, sender=sender, target=target,
+                order_by=("seq",), descending=True, limit=1,
+            )
+        ))
+        headers = (BusRow.require(db, seq=seq) for seq in positions)
+        return tuple(header.kind.read_from(db, header.seq) for header in headers)
+
+
 class BusReader(RouteCounts, ABC):
     """One query implementation for individual reads and a held read snapshot."""
 
@@ -231,6 +251,12 @@ def latest_conversations(
                 (*parameters, through),
             )
 
+    def latest_addressed(
+        self, senders: frozenset[str], targets: frozenset[str],
+    ) -> tuple[Message, ...]:
+        with self.observing() as db:
+            return AddressedRow.latest(db, senders, targets)
+
     def messages_for_references(
         self, references: Iterable[MessageReference],
     ) -> tuple[Message, ...]:
@@ -248,60 +274,74 @@ def deliveries_for_references(
         return tuple(delivery for record in self.messages_for_references(references)
                      for delivery in record.deliveries())
 
-    def _source_page(
+    def _source_sequences(
         self, query: str, parameters: tuple[object, ...], *, after: int = 0,
         before: int | None = None, limit: int = 100, descending: bool = True,
         sequences: tuple[int, ...] = (),
-    ) -> tuple[BusRecord, ...]:
+    ) -> Generator[int]:
+        """Stream selected frozen participant positions on the held snapshot."""
         with self.observing() as db:
             before = BusRow.last_seq(db) + 1 if before is None else before
             selected = ""
             if sequences:
                 selected = " AND seq IN (" + ",".join("?" for _ in sequences) + ")"
             order = "DESC" if descending else "ASC"
-            rows = db.execute(
+            with closing(db.execute(
                 f"SELECT seq FROM ({query}) WHERE seq > ? AND seq < ?{selected} "
                 f"ORDER BY seq {order} LIMIT ?",
                 (*parameters, after, before, *sequences, limit),
-            )
-            headers = (BusRow.require(db, seq=FieldCodec.decode(int, row[0])) for row in rows)
-            return tuple(header.kind.read_from(db, header.seq) for header in headers)
+            )) as rows:
+                for row in rows:
+                    yield FieldCodec.decode(int, row[0])
 
     def conversation(
         self, lookup: str, *, after: int = 0, before: int | None = None, limit: int = 100,
         descending: bool = True, sequences: tuple[int, ...] = (),
-    ) -> tuple[BusRecord, ...]:
-        return self._source_page(
-            f'SELECT seq FROM "{SentSource.declared_name}" WHERE lookup=? UNION '
-            f'SELECT seq FROM "{ReceivedSource.declared_name}" WHERE lookup=?',
-            (lookup, lookup), after=after, before=before, limit=limit,
-            descending=descending, sequences=sequences,
-        )
+    ) -> tuple[Delivered, ...]:
+        with self.reading() as source:
+            positions: set[int] = set()
+            for participant in ParticipantSource.schema_rows():
+                with closing(source._source_sequences(
+                    *participant.query(lookup), after=after, before=before, descending=descending,
+                    sequences=sequences, limit=limit,
+                )) as selected:
+                    positions.update(selected)
+            return tuple(source.delivery(seq) for seq in islice(
+                sorted(positions, reverse=descending), limit,
+            ))
 
     def sent(
         self, lookup: str, *, after: int = 0, before: int | None = None,
         limit: int = 100, descending: bool = True,
-    ) -> tuple[BusRecord, ...]:
-        return self._source_page(
-            f'SELECT seq FROM "{SentSource.declared_name}" WHERE lookup=?', (lookup,),
-            after=after, before=before, limit=limit, descending=descending,
-        )
+    ) -> tuple[Delivered, ...]:
+        with self.reading() as source, closing(source._source_sequences(
+            *SentSource.query(lookup),
+            after=after, before=before, descending=descending, limit=limit,
+        )) as positions:
+            return tuple(source.delivery(seq) for seq in positions)
 
     def received(
         self, lookup: str, *, after: int = 0, before: int | None = None, limit: int = 100,
         sender_lookups: tuple[str, ...] = (), descending: bool = True,
-        source: type[ReceivedSource] = ReceivedSource, sequences: tuple[int, ...] = (),
-    ) -> tuple[BusRecord, ...]:
-        sender = ""
-        if sender_lookups:
-            marks = ",".join("?" for _ in sender_lookups)
-            sender = (f' AND seq IN (SELECT seq FROM "{SentSource.declared_name}" '
-                      f'WHERE lookup IN ({marks}))')
-        return self._source_page(
-            f'SELECT seq FROM "{source.declared_name}" WHERE lookup=?{sender}',
-            (lookup, *sender_lookups), after=after, before=before, limit=limit,
-            descending=descending, sequences=sequences,
-        )
+        sequences: tuple[int, ...] = (),
+    ) -> tuple[Delivered, ...]:
+        with self.reading() as source, closing(source._source_sequences(
+            *ReceivedSource.query(lookup, sender_lookups), after=after, before=before,
+            descending=descending, sequences=sequences, limit=limit,
+        )) as positions:
+            return tuple(source.delivery(seq) for seq in positions)
+
+    def direct_replies(
+        self, lookup: str, *, after: int = 0, limit: int = 100,
+        sender_lookups: tuple[str, ...] = (), descending: bool = True,
+        sequences: tuple[int, ...] = (),
+    ) -> tuple[Delivered, ...]:
+        """Read the bounded append-time policy index before loading deliveries."""
+        with self.reading() as source, closing(source._source_sequences(
+            *DirectReplySource.query(lookup, sender_lookups), after=after,
+            descending=descending, sequences=sequences, limit=limit,
+        )) as positions:
+            return tuple(source.delivery(seq) for seq in positions)
 
     def unanswered_references(
         self, sequences: tuple[int, ...], responders: frozenset[str],

src/agent_comms/bus_publication.py

@@ -22,6 +22,7 @@
 from agent_comms.typed_table import Column, Generated, sql_literal
 
 from .field_codec import FieldCodec
+from .read_basis import RepliesDisplayScope
 from .wake_policy import RecipientDecision
 
 if TYPE_CHECKING:
@@ -42,22 +43,28 @@ def sender_lookup(self) -> str:
         return self.audience.sender_lookup
 
     def projections(self):
-        from .bus_tables import DirectReplySource, ReceivedSource, SentSource
+        from .bus_tables import AwaitingAnswerSource, ReceivedSource, SentSource
 
         yield from super().projections()
         yield SentSource(self.audience.sender_lookup, self.seq)
         for recipient in self.audience.recipients:
             yield ReceivedSource(recipient.recipient_lookup, self.seq)
-            if self.direct_for(recipient.recipient_lookup):
-                yield DirectReplySource(recipient.recipient_lookup, self.seq)
+            scope = RepliesDisplayScope(
+                frozenset({recipient.canonical_thread}), frozenset({self.sender}),
+            )
+            if scope.includes(self):
+                yield AwaitingAnswerSource(
+                    recipient.recipient_lookup, self.seq, recipient.canonical_thread,
+                    self.role.response_policy(self).starts_direct_turn,
+                )
 
     def deliveries(self) -> tuple[Self, ...]:
         return (self,)
 
     def direct_for(self, recipient_lookup: str) -> bool:
         """Direct reply semantics and the committed addressed incarnation agree."""
         policy = self.message.role.response_policy(self.message)
-        return (policy.starts_turn and policy.separate_turn
+        return (policy.starts_direct_turn
                 and self.audience.includes_lookup(recipient_lookup))
 
     def compaction_messages_for(self, recipient_lookup: str) -> tuple[Message, ...]: