diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md
index f9fcec6bd9..fd8a661d0a 100644
--- a/plans/commander-native-plan.md
+++ b/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.**
diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py
index 25f0c9b891..79b4ecb901 100644
--- a/src/agent_comms/bus.py
+++ b/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],
diff --git a/src/agent_comms/bus_publication.py b/src/agent_comms/bus_publication.py
index 7b31db09b5..1a6210c5c8 100644
--- a/src/agent_comms/bus_publication.py
+++ b/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,14 +43,20 @@ 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,)
@@ -57,7 +64,7 @@ def deliveries(self) -> tuple[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, ...]:
diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py
index d1b1968402..e71f1caac1 100644
--- a/src/agent_comms/bus_tables.py
+++ b/src/agent_comms/bus_tables.py
@@ -22,7 +22,6 @@
 from agent_comms.errors import RelationViolationError
 from agent_comms.field_codec import FieldCodec, WireName, WireOrder
 from agent_comms.message_reference import MessageReference
-from agent_comms.response_policy import ResponsePolicy
 from agent_comms.typed_table import (
     AppendOnlyTables,
     Column,
@@ -31,6 +30,8 @@
     DurableTables,
     ForeignKey,
     Index,
+    ProducedRow,
+    ProjectionProducerTables,
     SharedFileTables,
     TypedRow,
     TypedTable,
@@ -46,7 +47,7 @@ class BusTables(SharedFileTables, DurableTables, AppendOnlyTables):
     """Bus history cannot be rebuilt from another source."""
 
 
-class BusProjectionTables(SharedFileTables, DerivedTables):
+class BusProjectionTables(ProducedRow, SharedFileTables, DerivedTables):
     """Reverse lookups are derived from the record, in its append transaction."""
 
     @classmethod
@@ -58,14 +59,21 @@ def create_schema(cls, db: sqlite3.Connection) -> None:
                 header.kind.read_from(db, header.seq).project_into(db)
 
 
-class BusSchema(ComposedSchemas, BusTables, BusProjectionTables):
+class BusSchema(ComposedSchemas, BusTables, ProjectionProducerTables, BusProjectionTables):
     """One file reconciles its declared families using each family's carry policy."""
 
+
 @dataclass(frozen=True)
 class ParticipantSource(TypedTable, BusProjectionTables, declared_name=None):
+    """A recorded participant's positions in actual frozen deliveries."""
+
     lookup: Annotated[str, Column(primary_key=True)]
     seq: Annotated[int, Column(primary_key=True)]
 
+    @classmethod
+    def query(cls, lookup: str) -> tuple[str, tuple[object, ...]]:
+        return f'SELECT seq FROM "{cls.declared_name}" WHERE lookup=?', (lookup,)
+
 
 class SentSource(ParticipantSource):
     """The participant authored this record."""
@@ -74,9 +82,51 @@ class SentSource(ParticipantSource):
 class ReceivedSource(ParticipantSource):
     """The participant belongs to the frozen delivery audience."""
 
+    @classmethod
+    def query(
+        cls, lookup: str, sender_lookups: tuple[str, ...] = (),
+    ) -> tuple[str, tuple[object, ...]]:
+        query, parameters = super().query(lookup)
+        if sender_lookups:
+            marks = ",".join("?" for _ in sender_lookups)
+            query += (f' AND seq IN (SELECT seq FROM "{SentSource.declared_name}" '
+                      f'WHERE lookup IN ({marks}))')
+        return query, (*parameters, *sender_lookups)
+
+
+@dataclass(frozen=True)
+class AwaitingAnswerSource(ReceivedSource):
+    """The scope owner's append-time answer eligibility for a frozen recipient."""
+
+    recipient: str
+    starts_direct_turn: bool
+    renamed_from = ("direct_reply_source",)
+    indexes = (
+        Index(("recipient", "seq")),
+        Index(("lookup", "starts_direct_turn", "seq")),
+    )
+
+    @classmethod
+    def recipient_predicate(
+        cls, sequence: str, recipients: frozenset[str],
+    ) -> tuple[str, tuple[str, ...]]:
+        marks = ",".join("?" for _ in recipients)
+        return (
+            f'EXISTS (SELECT 1 FROM "{cls.declared_name}" AS awaiting '
+            f'WHERE awaiting.seq={sequence} AND awaiting.recipient IN ({marks}))',
+            tuple(recipients),
+        )
+
 
-class DirectReplySource(ReceivedSource):
-    """The record's declared reply policy starts a separate direct turn."""
+class DirectReplySource(AwaitingAnswerSource, declared_name=None):
+    """A bounded selection of the same answer index, with no second stored table."""
+
+    @classmethod
+    def query(
+        cls, lookup: str, sender_lookups: tuple[str, ...] = (),
+    ) -> tuple[str, tuple[object, ...]]:
+        query, parameters = AwaitingAnswerSource.query(lookup, sender_lookups)
+        return query + " AND starts_direct_turn=?", (*parameters, True)
 
 
 @dataclass(frozen=True)
@@ -84,7 +134,6 @@ class MessageIdentitySource(MessageReference, TypedTable, BusProjectionTables):
     """A record's existing reference, searchable without recomputing its content hash."""
 
     sender: str
-    response_policy: type[ResponsePolicy]
     indexes = (
         Index(("seq",), unique=True),
         Index(("message_id",)),
@@ -99,16 +148,6 @@ def reference_sql(cls, alias: str) -> str:
         )
         return f"json_object({fields})"
 
-    @classmethod
-    def eligible_sql(cls, scope: RepliesDisplayScope, identity: str,
-                     content: str, address: str) -> tuple[str, tuple]:
-        return scope.response_predicate(
-            cls._sql(identity + ".{sender}"),
-            Addressed.table._sql(address + ".{target}"),
-            cls._sql(identity + ".{response_policy}"),
-            messages.Message.table._sql(content + ".{mentions}"),
-        )
-
 
 @dataclass(frozen=True)
 class AnsweredSource(TypedTable, BusProjectionTables):
@@ -149,8 +188,8 @@ def unanswered(
         """Read unanswered originals through the existing reference and sender indexes."""
         requesters, responders = scope.second_names - scope.first_names, scope.first_names
         reference = MessageIdentitySource.reference_sql("asked")
-        eligible, eligibility = MessageIdentitySource.eligible_sql(
-            scope, "asked", "message", "asked_route",
+        eligible, eligibility = AwaitingAnswerSource.recipient_predicate(
+            "asked.seq", scope.first_names,
         )
         asking = ",".join("?" for _ in requesters)
         replying = ",".join("?" for _ in responders)
@@ -162,7 +201,6 @@ def unanswered(
         query = (
             f'SELECT asked.seq FROM "{MessageIdentitySource.declared_name}" AS asked '
             f'JOIN "{messages.Message.table.declared_name}" AS message USING(seq) '
-            f'JOIN "{Addressed.table.declared_name}" AS asked_route USING(seq) '
             f'WHERE asked.sender IN ({asking}) AND '
             + messages.Message.table._sql('message.{timestamp} BETWEEN ? AND ? ')
             + f'AND asked.seq < ? {kind_filter}AND ({eligible}) AND NOT EXISTS ('
@@ -184,8 +222,8 @@ def _answer_rows(
         since: float, selection: str,
     ) -> sqlite3.Cursor:
         requesters, responders = scope.second_names, scope.first_names
-        eligible, eligibility = MessageIdentitySource.eligible_sql(
-            scope, "asked", "question", "asked_route",
+        eligible, eligibility = AwaitingAnswerSource.recipient_predicate(
+            "asked.seq", scope.first_names,
         )
         asking = ",".join("?" for _ in requesters)
         replying = ",".join("?" for _ in responders)
@@ -202,7 +240,6 @@ def _answer_rows(
             f'JOIN "{identity}" AS asked ON '
             + cls._sql(f'answered.{{original}}={MessageIdentitySource.reference_sql("asked")} ')
             + f'JOIN "{content.declared_name}" AS question ON question.seq=asked.seq '
-            f'JOIN "{Addressed.table.declared_name}" AS asked_route ON asked_route.seq=asked.seq '
             f'WHERE answered.seq=reply.seq AND asked.seq<reply.seq '
             f'AND asked.sender IN ({asking}) AND '
             + content._sql('question.{timestamp} >= ? ')
@@ -420,3 +457,4 @@ def project_into(self, db: sqlite3.Connection) -> None:
 class Addressed(BusCapability):
     sender: Annotated[str, WireName("from"), WireOrder(2)]
     target: Annotated[str, WireName("to"), WireOrder(3), Column(index=True)]
+    indexes: ClassVar[tuple[Index, ...]] = (Index(("sender", "target", "seq")),)
diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py
index bb631e6b25..ca6f8bf188 100644
--- a/src/agent_comms/commander/standing.py
+++ b/src/agent_comms/commander/standing.py
@@ -9,18 +9,18 @@
 
 from agent_comms.build_identity import RecordedBuildIdentity
 from agent_comms.bus import BusReader
-from agent_comms.channel_targets import ThreadTarget
 from agent_comms.commander.config import CommanderConfig
 from agent_comms.commander.questions import UnansweredUnitAsks
 from agent_comms.commander_heartbeat import CommanderHeartbeat
 from agent_comms.comms import Comms
 from agent_comms.goal_actions import ReissuedOrderGoalAction
 from agent_comms.message_bus import WatchPass
-from agent_comms.message_page import MessageReadRequest, PageBounds
+from agent_comms.message_page import MessageReadRequest
 from agent_comms.messages import Message
+from agent_comms.owner_requests import OwnerRequestId
 from agent_comms.phase_evidence import PlanReading
 from agent_comms.plan_phases import PlanPhase
-from agent_comms.read_basis import NamedViewer, RepliesDisplayScope
+from agent_comms.read_basis import RepliesDisplayScope
 from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.registry_document import RegistrySnapshot
 from agent_comms.standing_rules import (
@@ -337,38 +337,75 @@ def heartbeat_line(self, current: Sequence[WaitingUnownedWheelSpin]) -> str:
 
 
 @dataclass(frozen=True)
-class HeartbeatSilent(IndependentStandingRule, DatedStandingRule):
-    """A commander the heartbeat serves whose newest beat is older than the cadence allows.
+class HeartbeatSilent(IndependentStandingRule, DatedStandingRule, declared_name=None):
+    """Missing or stale heartbeat service for the commander's current open requests.
 
     The heartbeat beats on every pass, so its silence is the failure; this rule reads the
     bus in the commander's own tool process, where a dead heartbeat cannot hide it.
     """
 
     seconds: ClassVar[int] = 180
-    beat: int
     since: float
 
     @classmethod
     def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
-        served = {thread.incarnation
-                  for thread in CommanderHeartbeat.commanders(comms.registry, comms.goals)}
-        if CommanderHeartbeat.sender not in comms.registry or (
-                comms.registry.require(commander).incarnation not in served):
+        return tuple(rule for member in cls.members_with(cls)
+                     for rule in member.violations(comms, commander))
+
+    @staticmethod
+    def beats(comms: Comms, commander: str) -> tuple[Message, ...]:
+        snapshot = comms.registry.snapshot()
+        if snapshot.canonical_name(CommanderHeartbeat.sender) not in snapshot.threads:
             return ()
-        page = comms.views.history_page(ThreadTarget(CommanderHeartbeat.sender),
-                                        PageBounds(limit=1), viewer=NamedViewer(commander))
-        return tuple(cls(beat.seq, beat.timestamp + cls.seconds) for beat in page.messages[-1:]
-                     if beat.sender == CommanderHeartbeat.sender
-                     and beat.timestamp + cls.seconds < time.time())
+        return comms.bus.bus.latest_addressed(
+            snapshot.names_for(CommanderHeartbeat.sender), snapshot.names_for(commander),
+        )
 
     def began(self) -> float:
         return self.since
 
+
+@dataclass(frozen=True)
+class MissedHeartbeat(HeartbeatSilent):
+    beat: int
+
+    @classmethod
+    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
+        requests = CommanderHeartbeat.requested_by(comms.registry.require(commander), comms.goals)
+        if not requests:
+            return ()
+        now = time.time()
+        found = tuple(
+            cls(since=max(request.created_at, beat.timestamp) + cls.seconds, beat=beat.seq)
+            for request in requests for beat in cls.beats(comms, commander)
+        )
+        return tuple(rule for rule in found if rule.since < now)
+
     def line(self) -> str:
         return (f"HEARTBEAT SILENT: no beat since seq {self.beat}, silent past "
                 f"{RecordedEventTime(self.since).clock()}"
                 " -> read the connector journal for the failing pass and fix its owner")
 
+
+@dataclass(frozen=True)
+class HeartbeatNotStarted(HeartbeatSilent):
+    request: OwnerRequestId
+
+    @classmethod
+    def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]:
+        requests = CommanderHeartbeat.requested_by(comms.registry.require(commander), comms.goals)
+        if not requests or cls.beats(comms, commander):
+            return ()
+        now = time.time()
+        found = tuple(cls(since=request.created_at + cls.seconds, request=request.id)
+                      for request in requests)
+        return tuple(rule for rule in found if rule.since < now)
+
+    def line(self) -> str:
+        return (f"HEARTBEAT NOT STARTED: no beat for request {self.request.value}, overdue since "
+                f"{RecordedEventTime(self.since).clock()}"
+                " -> start the connector heartbeat and fix its failing pass")
+
 @dataclass(frozen=True)
 class Unjudged(ReasonedStandingRule, UndatedStandingRule):
     limit: ClassVar[int] = 30
diff --git a/src/agent_comms/commander_heartbeat.py b/src/agent_comms/commander_heartbeat.py
index a82676fd2e..b3da6bbc0c 100644
--- a/src/agent_comms/commander_heartbeat.py
+++ b/src/agent_comms/commander_heartbeat.py
@@ -17,7 +17,7 @@
 if TYPE_CHECKING:
     from agent_comms.comms import Comms
     from agent_comms.goal_management import Goals
-    from agent_comms.todos import Todo
+    from agent_comms.todos import OwnerRequest, Todo
 
 
 @dataclass
@@ -81,14 +81,19 @@ class CommanderHeartbeat(ServiceSender, WatchPass[tuple[CommanderObservation, ..
     sender = "commander-heartbeat"
     reviews: dict[ThreadIncarnation, CommanderReview] = field(default_factory=dict)
 
+    @classmethod
+    def requested_by(cls, thread: Thread, goals: Goals) -> tuple[OwnerRequest, ...]:
+        """The oldest open request makes a live holder owed service, even before its first beat."""
+        return (goals.todos.open_requests(thread.incarnation, limit=1, descending=False)
+                if thread.process_alive else ())
+
     @classmethod
     def commanders(cls, registry: Registration, goals: Goals) -> tuple[Thread, ...]:
         """The live request holders whose current incarnation owns commander work."""
         return tuple(
             thread
             for thread in registry.all_threads().values()
-            if thread.process_alive
-            and goals.todos.open_requests(thread.incarnation, limit=1)
+            if cls.requested_by(thread, goals)
         )
 
     def observe(self, comms: Comms) -> tuple[CommanderObservation, ...]:
diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py
index 3f503c998f..c0764e8122 100644
--- a/src/agent_comms/goal_management.py
+++ b/src/agent_comms/goal_management.py
@@ -42,7 +42,6 @@
     from .native_input_owner import RegistryOwner
     from .registry_document import RegistrySnapshot
     from .thread_identity import ThreadIncarnation
-from .bus_tables import DirectReplySource
 from .goal_presentation import (
     GoalExecution,
     GoalExecutionSource,
@@ -199,18 +198,18 @@ def _goal_input_review(
         with self.bus.bus.reading() as source:
             # Preserve older UNKNOWN rows; new inputs begin at this wait's own cursor.
             # Both selections use the same frozen recipient and sender projections.
-            page = source.received(
+            page = source.direct_replies(
                 lookup, sequences=sequences, limit=len(sequences), descending=False,
-                sender_lookups=senders, source=DirectReplySource,
+                sender_lookups=senders,
             ) if sequences else ()
             while True:
                 replies.update({
                     InputDispositions.bus_key(original.message, thread): original.message
-                    for record in page for original in record.deliveries()
+                    for original in page
                 })
-                page = source.received(
+                page = source.direct_replies(
                     lookup, after=after_seq, limit=100, descending=False,
-                    sender_lookups=senders, source=DirectReplySource,
+                    sender_lookups=senders,
                 )
                 if not page:
                     break
diff --git a/src/agent_comms/goal_waits.py b/src/agent_comms/goal_waits.py
index e838547762..02d0ae0379 100644
--- a/src/agent_comms/goal_waits.py
+++ b/src/agent_comms/goal_waits.py
@@ -10,7 +10,6 @@
 
 from .bus import Bus
 from .bus_publication import Delivered
-from .bus_tables import DirectReplySource
 from .declared_family import DeclaredFamily
 from .field_codec import NullMember
 from .goal_presentation import GoalExecution, GoalExecutionSource, GoalWaitTarget, StandbyExecution
@@ -192,12 +191,10 @@ def has_reply(self, bus: Bus, owner: RecordedThreadIncarnation) -> bool:
             return False
         return any(
             self.matches(original, owner)
-            for record in bus.received(
+            for original in bus.direct_replies(
                 owner.lookup, after=self.after_seq, limit=1, descending=False,
                 sender_lookups=tuple(target.incarnation.lookup for target in self.targets),
-                source=DirectReplySource,
             )
-            for original in record.deliveries()
         )
 
 
diff --git a/src/agent_comms/messages.py b/src/agent_comms/messages.py
index f4f4e43813..b7722c205d 100644
--- a/src/agent_comms/messages.py
+++ b/src/agent_comms/messages.py
@@ -411,9 +411,7 @@ def projections(self):
             yield BusDisplayIndex.from_message(self, participant=participant)
         yield from self.role.activity_rows(self)
         yield from self.type.route_rows(self)
-        yield MessageIdentitySource(
-            self.seq, self.message_id, self.sender, type(self.role.response_policy(self)),
-        )
+        yield MessageIdentitySource(self.seq, self.message_id, self.sender)
         for original in self.answers:
             yield AnsweredSource(original, self.seq)
 
diff --git a/src/agent_comms/private_runtime_schema.py b/src/agent_comms/private_runtime_schema.py
index 0a47f50d1e..8838938914 100644
--- a/src/agent_comms/private_runtime_schema.py
+++ b/src/agent_comms/private_runtime_schema.py
@@ -1,4 +1,4 @@
-"""Private coordinator declarations and the complete target's derived schema view."""
+"""Private coordinator declarations and the mandatory target's derived schema view."""
 
 from __future__ import annotations
 
@@ -57,4 +57,4 @@ def install(cls, store: Coordination) -> None:
 
 
 class PrivateCoordinatorSchema(ComposedSchemas, CoordinatorTable, PrivateCoordinatorTable):
-    """The private file target, derived from its core and private row families."""
+    """Mandatory core/private families; optional awareness owns its separate view."""
diff --git a/src/agent_comms/read_basis.py b/src/agent_comms/read_basis.py
index 8b2b26dcd4..34a01db129 100644
--- a/src/agent_comms/read_basis.py
+++ b/src/agent_comms/read_basis.py
@@ -31,7 +31,6 @@
 from .recorded_bus import RecordedBus, RecordedBusIdentity
 from .store_files import SharedLock
 from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation
-from .typed_table import sql_literal
 
 
 class DisplayViewer(SubclassUnion, affix="Viewer"):
@@ -939,25 +938,6 @@ def includes(self, message: Message) -> bool:
                 or (message.sender in self.second_names
                     and bool(message.response_eligibility(tuple(self.first_names)).recipients)))
 
-    def response_predicate(self, sender: str, target: str, policy: str,
-                           mentions: str) -> tuple[str, tuple]:
-        """Original eligibility on indexed fields, before selecting an answer."""
-        from .response_policy import ResponsePolicy
-
-        first, second = tuple(self.first_names), tuple(self.second_names)
-        a, b = ",".join("?" for _ in first), ",".join("?" for _ in second)
-        direct, parameters = self.address_predicate(sender, target)
-        conditions, selected = [], []
-        for member in ResponsePolicy.members_with(ResponsePolicy):
-            condition, values = member.recipients_sql(mentions, first)
-            conditions.append(f"({policy}={sql_literal(member)} AND ({condition}))")
-            selected.extend(values)
-        return (
-            f"({sender} IN ({a}) OR ({direct}) OR "
-            f"({sender} IN ({b}) AND ({' OR '.join(conditions)})))",
-            (*first, *parameters, *second, *selected),
-        )
-
 
 @dataclass(frozen=True, slots=True)
 class ListedDeliveries(MessageDisplayScope):
diff --git a/src/agent_comms/response_policy.py b/src/agent_comms/response_policy.py
index 9982a67a08..82ef826e5b 100644
--- a/src/agent_comms/response_policy.py
+++ b/src/agent_comms/response_policy.py
@@ -9,8 +9,6 @@
 from typing import TYPE_CHECKING
 
 from .declared_family import DeclaredFamily
-from .field_codec import FieldCodec
-from .mentions import ThreadMention
 
 if TYPE_CHECKING:
     from .messages import Message
@@ -22,6 +20,10 @@ class ResponsePolicy(DeclaredFamily, affix="Policy"):
     starts_turn = False
     separate_turn = False
 
+    @property
+    def starts_direct_turn(self) -> bool:
+        return self.starts_turn and self.separate_turn
+
     @classmethod
     @cache
     def instance(cls) -> ResponsePolicy:
@@ -36,11 +38,6 @@ def recipients(
     ) -> tuple[str, ...]:
         """Resolve this policy's channel recipients."""
 
-    @classmethod
-    @abstractmethod
-    def recipients_sql(cls, mentions: str, audience: Sequence[str]) -> tuple[str, tuple]:
-        """Select the same channel recipients from committed message fields."""
-
     @abstractmethod
     def guidance(self, message: Message, *, aliases: Mapping[str, str] | None = None) -> str:
         """Instructions included in a queued input."""
@@ -61,10 +58,6 @@ def recipients(
     ) -> tuple[str, ...]:
         return ()
 
-    @classmethod
-    def recipients_sql(cls, mentions: str, audience: Sequence[str]) -> tuple[str, tuple]:
-        return "0", ()
-
     def validate_recipients(self, recipients: tuple[str, ...]) -> None:
         if recipients:
             raise ValueError("This eligibility cannot declare channel recipients.")
@@ -90,10 +83,6 @@ def disposition_key(self, message: Message, owner: Thread) -> str:
 class CollectivePolicy(ResponsePolicy):
     starts_turn = True
 
-    @classmethod
-    def recipients_sql(cls, mentions: str, audience: Sequence[str]) -> tuple[str, tuple]:
-        return ("1" if audience else "0"), ()
-
     def recipients(
         self, message: Message, audience: Sequence[str], *, aliases: Mapping[str, str] | None = None
     ) -> tuple[str, ...]:
@@ -104,16 +93,6 @@ def guidance(self, message: Message, *, aliases: Mapping[str, str] | None = None
 
 
 class MentionedOnlyPolicy(ResponsePolicy):
-    @classmethod
-    def recipients_sql(cls, mentions: str, audience: Sequence[str]) -> tuple[str, tuple]:
-        key = FieldCodec.field_plan(ThreadMention, "thread").key
-        marks = ",".join("?" for _ in audience)
-        return (
-            f"EXISTS (SELECT 1 FROM json_each({mentions}) AS mention "
-            f"WHERE json_extract(mention.value, ?) IN ({marks}))",
-            (f"$.{key}", *audience),
-        )
-
     def recipients(
         self, message: Message, audience: Sequence[str], *, aliases: Mapping[str, str] | None = None
     ) -> tuple[str, ...]:
diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py
index 7042536c95..f1008ce2f8 100644
--- a/src/agent_comms/todos.py
+++ b/src/agent_comms/todos.py
@@ -3565,13 +3565,13 @@ def owner_request(self, request: OwnerRequestId) -> OwnerRequest:
             return OwnerRequest.require(db, id=request)
 
     def open_requests(
-        self, commander: ThreadIncarnation, *, limit: int = -1,
+        self, commander: ThreadIncarnation, *, limit: int = -1, descending: bool = True,
     ) -> tuple[OwnerRequest, ...]:
         with self.snapshot() as db:
             return tuple(OwnerRequest.select(
                 db, where="commander_created_at=?",
                 parameters=(commander.require_recorded().created_at,), state=OpenRequestState(),
-                order_by=("created_at", "id"), descending=True, limit=limit,
+                order_by=("created_at", "id"), descending=descending, limit=limit,
             ))
 
     def record_directive(self, request: OwnerRequestId, directive: RequestDirective) -> None:
diff --git a/src/agent_comms/typed_table.py b/src/agent_comms/typed_table.py
index 3a6911cdc8..b08232fbf5 100644
--- a/src/agent_comms/typed_table.py
+++ b/src/agent_comms/typed_table.py
@@ -62,6 +62,7 @@
     StoreCommittedError,
 )
 from .maintenance_barrier import HOLDER_NONCE, MaintenanceBarrier
+from .package_stamp import STAMP
 from .store_files import _store_lock
 
 if TYPE_CHECKING:
@@ -870,6 +871,15 @@ def schema_rows(cls) -> tuple[type[TypedTable], ...]:
             row for family in cls.schema_families() for row in family.schema_rows()
         ))
 
+    @classmethod
+    def schema_differs(cls, db: sqlite3.Connection) -> bool:
+        return any(family.schema_differs(db) for family in cls.schema_families())
+
+    @classmethod
+    def create_schema(cls, db: sqlite3.Connection) -> None:
+        for family in cls.schema_families():
+            family.create_schema(db)
+
     @classmethod
     def carry_schema(cls, db: sqlite3.Connection) -> None:
         for family in cls.schema_families():
@@ -1330,9 +1340,26 @@ class DerivedTables(TableSchema):
 
     __slots__ = ()
 
+    @classmethod
+    def projection_rows(cls) -> tuple[type[TypedTable], ...]:
+        return tuple(row for row in cls.schema_rows() if issubclass(row, ProducedRow))
+
+    @classmethod
+    def schema_differs(cls, db: sqlite3.Connection) -> bool:
+        return (super().schema_differs(db)
+                or not ProjectionProducer.matches(db, cls.projection_rows()))
+
+    @classmethod
+    def create_schema(cls, db: sqlite3.Connection) -> None:
+        super().create_schema(db)
+        for producer in ProjectionProducer.records_for(cls.projection_rows()):
+            producer.upsert(db)
+
     @classmethod
     def reset_schema(cls, db: sqlite3.Connection) -> None:
         """Inside the caller's transaction: rebuild only this schema's derived tables."""
+        for row in cls.schema_rows():
+            row.take_previous_name(db)
         for stored in cls.stored_tables(db):
             stored.drop(db)
         cls.create_schema(db)
@@ -1354,6 +1381,14 @@ def carry_schema(cls, db: sqlite3.Connection) -> None:
         cls.derive_schema(db)
 
 
+class ProjectionProducerTables(SharedFileTables, DurableTables):
+    """The producer declaration last committed with each rebuilt projection."""
+
+
+class ProducedRow:
+    """Rows rebuilt whenever the installed Core build changes."""
+
+
 @lru_cache(maxsize=4)
 def _claimed_names(members: tuple[type[TypedTable], ...]) -> frozenset[str]:
     """Every object name any row declares; keyed by the registry, so a new member refreshes it."""
@@ -1860,6 +1895,25 @@ def update(
         )
 
 
+@dataclass(frozen=True)
+class ProjectionProducer(TypedTable, ProjectionProducerTables):
+    projection: Annotated[str, Column(primary_key=True)]
+    commit: str
+
+    @classmethod
+    def records_for(cls, rows: tuple[type[TypedTable], ...]) -> tuple[Self, ...]:
+        return tuple(cls(row.declared_name, STAMP.commit()) for row in rows)
+
+    @classmethod
+    def matches(cls, db: sqlite3.Connection, rows: tuple[type[TypedTable], ...]) -> bool:
+        if not rows:
+            return True
+        if not cls.present(db):
+            return False
+        stored = {record.projection: record for record in cls.select(db)}
+        return all(stored.get(record.projection) == record for record in cls.records_for(rows))
+
+
 class StoreWriter(PublishedSQLiteFile, FileStore):
     """A transaction capability shared by stores and their turn-owned writers.
 
diff --git a/tests/test_bus.py b/tests/test_bus.py
index 9fd075cb55..db63dc0118 100644
--- a/tests/test_bus.py
+++ b/tests/test_bus.py
@@ -18,8 +18,8 @@
     AnsweredSource,
     BusProjectionTables,
     BusRow,
+    BusSchema,
     BusTables,
-    DirectReplySource,
     MessageIdentitySource,
 )
 from agent_comms.coordination_tables.publications import PublicationIntents
@@ -36,12 +36,20 @@
     MembershipChange,
     Message,
     MessageRoleWire,
+    NoticeRole,
 )
 from agent_comms.message_reference import MessageReference
+from agent_comms.package_stamp import STAMP
 from agent_comms.response_acknowledgement import (
     canonical_publication_key,
 )
-from agent_comms.typed_table import RowNotFoundError, SQLiteTableName, TypedTable
+from agent_comms.typed_table import (
+    ProjectionProducer,
+    RowNotFoundError,
+    SQLiteTableName,
+    TypedTable,
+    WriterOwnedFile,
+)
 from bus_record_fixture import publication_processes, record_from
 
 
@@ -84,6 +92,41 @@ def test_install_refuses_existing_schema_changes_until_an_explicit_rebuild(bus):
     assert bus.page(3, 2) == (second, record)
 
 
+def test_projection_freshness_follows_the_installed_core_commit(bus):
+    record = bus.append(lambda seq, db: PublicRecord.from_message(
+        Message("author", "worker", "preserved source", InfoKind, seq=seq),
+    ))
+    current = STAMP.commit()
+    older = "0" * 40
+    assert older != current
+    with bus.writer_connection() as db:
+        before = db.total_changes
+        assert not BusSchema.schema_differs(db)
+        assert not BusSchema.reconcile_schema(db, WriterOwnedFile())
+        assert db.total_changes == before
+
+        ProjectionProducer.update(
+            db, where="{projection}=?", parameters=(MessageIdentitySource.declared_name,),
+            commit=older,
+        )
+        MessageIdentitySource.update(db, where="{seq}=?", parameters=(record.seq,), sender="stale")
+        assert BusSchema.schema_differs(db)
+        assert BusSchema.reconcile_schema(db, WriterOwnedFile())
+        assert MessageIdentitySource.require(db, seq=record.seq).sender == record.sender
+        assert all(row.commit == current for row in ProjectionProducer.select(db))
+
+        ProjectionProducer.drop(db)
+        MessageIdentitySource.update(db, where="{seq}=?", parameters=(record.seq,), sender="stale")
+        assert BusSchema.schema_differs(db)
+        assert BusSchema.reconcile_schema(db, WriterOwnedFile())
+        assert MessageIdentitySource.require(db, seq=record.seq).sender == record.sender
+        assert ProjectionProducer.select(db) == list(ProjectionProducer.records_for(
+            BusProjectionTables.projection_rows()))
+        assert db.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
+        assert db.execute("PRAGMA foreign_key_check").fetchall() == []
+    assert bus.lookup(record.seq) == record
+
+
 @pytest.mark.parametrize("declaration", BusRecord.members_with(BusRecord))
 @pytest.mark.parametrize(
     "role",
@@ -265,9 +308,19 @@ def test_scoped_and_frozen_source_queries_are_sequence_bounded(bus):
     assert bus.scoped_page("target=?", ("worker",), kinds=(InfoKind,), limit=2) == (reply,)
     assert bus.conversation("b" * 32, limit=2) == (reply, notice)
     assert bus.sent("a" * 32, limit=1) == (reply,)
-    assert bus.received("b" * 32, after=original.seq, limit=1, descending=False,
-                        sender_lookups=("a" * 32,), source=DirectReplySource) == (reply,)
+    assert bus.direct_replies("b" * 32, after=original.seq, limit=1, descending=False,
+                              sender_lookups=("a" * 32,)) == (reply,)
     assert bus.received("b" * 32, sender_lookups=("f" * 32,)) == ()
+    assert bus.latest_addressed(frozenset({"author", "old-author"}),
+                                frozenset({"worker", "old-worker"})) == (reply,)
+    renamed = bus.append(lambda seq, db: PublicRecord.from_message(
+        Message("old-author", "old-worker", "newest across aliases", InfoKind, seq=seq)))
+    bus.append(lambda seq, db: PublicRecord.from_message(
+        Message("worker", "author", "later reverse reply", InfoKind, seq=seq)))
+    assert bus.latest_addressed(frozenset({"author", "old-author"}),
+                                frozenset({"worker", "old-worker"})) == (renamed,)
+    assert bus.latest_addressed(frozenset({"absent", "old-absent"}),
+                                frozenset({"worker", "old-worker"})) == ()
     assert bus.replies((original.reference,)) == (reply,)
     assert bus.deliveries_for_references((reply.reference,)) == (reply,)
     tips = bus.latest_conversations("target=?", ("worker",), notice.seq)
@@ -276,6 +329,21 @@ def test_scoped_and_frozen_source_queries_are_sequence_bounded(bus):
     )
     selected = bus.selected_sequences("target=?", ("worker",), reply.seq)
     assert {row.seq for row in selected} == {notice.seq, reply.seq}
+    assert bus.conversation("a" * 32, limit=1, descending=False) == (notice,)
+    assert bus.conversation(
+        "a" * 32, limit=1, descending=False, sequences=(reply.seq,),
+    ) == (reply,)
+    after = bus.revision()
+    for index in range(101):
+        bus.append(lambda seq, db, index=index: record_from(PrivateRecord, Message(
+            "author", "worker", f"notice {index}", InfoKind, seq=seq, role=NoticeRole(),
+        )))
+    accepted = bus.append(lambda seq, db: record_from(PrivateRecord, Message(
+        "author", "worker", "first eligible reply after notices", InfoKind, seq=seq,
+    )))
+    assert bus.direct_replies(
+        "b" * 32, after=after, limit=1, descending=False, sender_lookups=("a" * 32,),
+    ) == (accepted,)
 
 
 def test_unanswered_reference_selection_requires_the_recorded_original_and_exact_owner_answer(bus):
diff --git a/tests/test_goal_input_review.py b/tests/test_goal_input_review.py
index cd9ae45123..6907c3ae05 100644
--- a/tests/test_goal_input_review.py
+++ b/tests/test_goal_input_review.py
@@ -11,7 +11,8 @@
 from agent_comms.child_process import ProcessIdentity
 from agent_comms.comms import wire
 from conftest import installed_comms
-from agent_comms.field_codec import MalformedRecord
+from agent_comms.field_codec import FieldCodec, MalformedRecord
+from agent_comms.goal_presentation import GoalExecution
 from agent_comms.goal_actions import (
     ActiveGoalAction,
     GoalPrecondition,
@@ -94,10 +95,8 @@ def test_inspected_canonical_dependencies_allow_standby_without_replay(tmp_path,
     assert list(scoped.reviewed_inputs) == keys and len(scoped.excluded_inputs) == 2
     with pytest.raises(ValueError):
         report.invoke(comms, {**args, "reviewed_inputs": [*keys, f"bus:{unrelated.seq}"]}, caller=ThreadCaller("worker"))
-    assert (
-        report.invoke(comms, {**args, "reviewed_inputs": keys}, caller=ThreadCaller("worker"))["goal_execution"]["state"]
-        == "standby"
-    )
+    result = report.invoke(comms, {**args, "reviewed_inputs": keys}, caller=ThreadCaller("worker"))
+    assert FieldCodec.decode(GoalExecution, result["goal_execution"]).standby()
     again = comms.goals.goal_input_review("worker", goal.id, ["parent"])
     assert again.reviewed_inputs == ()
     assert [row.input_id for row in again.already_reviewed_inputs] == keys
diff --git a/tests/test_goal_standby_liveness.py b/tests/test_goal_standby_liveness.py
index 02d18e83aa..42bb98a1ed 100644
--- a/tests/test_goal_standby_liveness.py
+++ b/tests/test_goal_standby_liveness.py
@@ -115,7 +115,7 @@ def test_last_target_finishes_silently_releases_once_and_reopens(tmp_path):
     assert continued.id == goal.id and continued.state.declared_name == "active"
     assert continued.revision == goal.revision + 2
     assert "without a qualifying direct reply" in continued.progress
-    assert reopened.goals.goal_execution("owner").state.value == "runnable"
+    assert reopened.goals.goal_execution("owner").runnable()
     assert isinstance(reopened.goals.goal_wait("owner"), NoGoalWait)
     assert reopened.registry.require("owner").goal.state.pause_source is None
     assert reopened.goals.release_waits_after_terminal_turn(child_fence) == ()
diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py
index 67bedee42c..5190de7eaa 100644
--- a/tests/test_holder_standing_rules.py
+++ b/tests/test_holder_standing_rules.py
@@ -23,6 +23,8 @@
 from agent_comms.commander.questions import UnansweredBurst, UnansweredUnitAsks
 from agent_comms.commander.standing import (
     HeartbeatSilent,
+    HeartbeatNotStarted,
+    MissedHeartbeat,
     AnswersTo,
     UnitAskUnanswered,
 )
@@ -376,6 +378,12 @@ def test_latest_answer_uses_the_complete_original_reference_and_time_cut(held_bu
     latest = publish(held_bus, "lead", "worker", since + 6, answers=(question.reference,))
     assert tuple(message.reference for message in AnswersTo(scope, since).read_opened(
         held_bus.bus.bus)) == (latest.reference,)
+    # Bus order and the producer's timestamp are independent answers.
+    backdated = publish(held_bus, "lead", "worker", since + 2.5, answers=(question.reference,))
+    unrelated = publish(held_bus, "worker", "outsider", since + 7)
+    publish(held_bus, "lead", "worker", since + 8, answers=(unrelated.reference,))
+    assert AnswersTo(scope, since).read_opened(held_bus.bus.bus) == (backdated,)
+    assert held_bus.bus.bus.answer_timestamp(scope, since=since) == latest.timestamp
 
 
 def test_unanswered_pages_match_the_complete_answer_reference(held_bus):
@@ -396,8 +404,11 @@ def test_answering_an_unrelated_peer_does_not_clear_the_commander_burst(held_bus
     now = time.time()
     owed = publish(held_bus, "worker", "lead", now - 30)
     unrelated = publish(held_bus, "worker", "outsider", now - 20)
-    publish(held_bus, "lead", "outsider", now - 10, answers=(unrelated.reference,))
     scope = RepliesDisplayScope(frozenset({"lead"}), frozenset({"worker"}))
+    assert held_bus.bus.bus.unanswered_page(
+        scope, since=now - 60, before=now, upper=unrelated.seq + 1, limit=1,
+    ) == (owed,)
+    publish(held_bus, "lead", "outsider", now - 10, answers=(unrelated.reference,))
     assert AnswersTo(scope, now - 60).read_opened(held_bus.bus.bus) == ()
     assert UnansweredBurst(scope, now - 60, now).read_opened(held_bus.bus.bus) == (owed,)
     reply = publish(held_bus, "lead", "worker", now - 5, answers=(owed.reference,))
@@ -424,11 +435,17 @@ def test_latest_answer_retains_the_original_response_eligibility(
     original = held_bus.bus.publisher.publish_ordinary(
         Message(sender, target, body, InfoKind, timestamp=now - 20, role=role), destination=CommandTarget.named(target), origin=origin,
     )
-    reply = publish(held_bus, "lead", sender, now - 10, answers=(original.reference,))
     scope = RepliesDisplayScope(frozenset({"lead"}), frozenset({sender}))
     assert scope.includes(original) is eligible
+    assert held_bus.bus.bus.unanswered_page(
+        scope, since=now - 60, before=now, upper=original.seq + 1, limit=1,
+    ) == ((original,) if eligible else ())
+    reply = publish(held_bus, "lead", sender, now - 10, answers=(original.reference,))
     found = AnswersTo(scope, now - 60).read_opened(held_bus.bus.bus)
     assert found == ((reply,) if eligible else ())
+    assert held_bus.bus.bus.unanswered_page(
+        scope, since=now - 60, before=now, upper=reply.seq + 1, limit=1,
+    ) == ()
 
 
 def test_latest_answer_retains_the_original_commander_alias(held_bus):
@@ -644,11 +661,57 @@ def test_every_pass_beats_and_a_stale_beat_is_heartbeat_silent(held_bus, monkeyp
     assert HeartbeatSilent.violations(held_bus, "lead") == ()
     later = unchanged.timestamp + HeartbeatSilent.seconds + 1
     monkeypatch.setattr(time, "time", lambda: later)
+    # Neither a commander reply nor a fresh beat sent to another commander proves
+    # this commander's heartbeat is alive.
+    held_bus.messaging.send("lead", CommanderHeartbeat.sender, "I read the last beat")
+    held_bus.messaging.send(CommanderHeartbeat.sender, "outsider", "Another commander is healthy")
     silent, = HeartbeatSilent.violations(held_bus, "lead")
     assert silent.beat == unchanged.seq
     assert silent.line().startswith(f"HEARTBEAT SILENT: no beat since seq {unchanged.seq}")
 
 
+def test_first_heartbeat_is_owed_from_the_oldest_open_request_then_survives_renames(
+        held_bus, monkeypatch):
+    commander = held_bus.registry.require("lead")
+    held_bus.registry.register(replace(commander, process_identity=ProcessIdentity.capture(os.getpid())))
+    request, = CommanderHeartbeat.requested_by(held_bus.registry.require("lead"), held_bus.goals)
+    assert HeartbeatSilent.violations(held_bus, "lead") == ()
+    due = request.created_at + HeartbeatSilent.seconds
+    monkeypatch.setattr(time, "time", lambda: due + 1)
+    never, = HeartbeatSilent.violations(held_bus, "lead")
+    assert isinstance(never, HeartbeatNotStarted)
+    assert never.request == request.id and never.since == due
+    assert "HEARTBEAT NOT STARTED" in never.line()
+    assert HeartbeatSilent.broken(held_bus, "lead", observed=()) == (never,)
+
+    # A newer request and mere service registration cannot reset older unpaid service.
+    held_bus.goals.todos.add_request(commander.incarnation, "A newer independent request")
+    CommanderHeartbeat.register(held_bus.registry)
+    assert HeartbeatSilent.violations(held_bus, "lead") == (never,)
+    fresh = held_bus.bus.publisher.publish_ordinary(Message(
+        sender=CommanderHeartbeat.sender, target="lead", body="The heartbeat is running now",
+        type=InfoKind, timestamp=due + 1,
+    ), destination=CommandTarget.named("lead"), origin=AgentOrigin())
+    assert HeartbeatSilent.violations(held_bus, "lead") == ()
+
+    held_bus.threads._rename_thread("lead", "renamed-lead")
+    held_bus.threads._rename_thread(CommanderHeartbeat.sender, "renamed-heartbeat")
+    monkeypatch.setattr(time, "time", lambda: fresh.timestamp + HeartbeatSilent.seconds + 1)
+    held_bus.messaging.send("renamed-lead", "renamed-heartbeat", "Read the earlier beat")
+    silent, = HeartbeatSilent.violations(held_bus, "renamed-lead")
+    assert isinstance(silent, MissedHeartbeat) and silent.beat == fresh.seq
+    assert silent.since == fresh.timestamp + HeartbeatSilent.seconds
+    newer = held_bus.bus.publisher.publish_ordinary(Message(
+        sender="renamed-heartbeat", target="renamed-lead", body="Heartbeat after both renames",
+        type=InfoKind, timestamp=time.time(),
+    ), destination=CommandTarget.named("renamed-lead"), origin=AgentOrigin())
+    assert HeartbeatSilent.violations(held_bus, "renamed-lead") == ()
+    monkeypatch.setattr(time, "time", lambda: newer.timestamp + HeartbeatSilent.seconds + 1)
+    held_bus.messaging.send("renamed-lead", "renamed-heartbeat", "Read the new beat too")
+    latest, = HeartbeatSilent.violations(held_bus, "renamed-lead")
+    assert isinstance(latest, MissedHeartbeat) and latest.beat == newer.seq
+
+
 def test_one_answer_covers_an_escalation_burst_sent_before_it(held_bus):
     from agent_comms.commander.vibe_check import VibeCheckSender
 
diff --git a/tests/test_private_runtime_bootstrap.py b/tests/test_private_runtime_bootstrap.py
index b07dc5a452..0500064228 100644
--- a/tests/test_private_runtime_bootstrap.py
+++ b/tests/test_private_runtime_bootstrap.py
@@ -14,6 +14,7 @@
 
 def test_drifted_private_owner_refused_and_readers_leave_missing_schema_alone(tmp_path):
     comms = Comms(tmp_path / "wire")
+    comms.registry.prepare_native_root()
     Bus.install(Bus.path_in(comms.root))
     comms.registry.declare(comms.registry.new_thread('owner', frozenset(), str(tmp_path)))
     database = comms.root / "coordination.sqlite3"
diff --git a/tests/test_response_policy.py b/tests/test_response_policy.py
index 049fb0ba30..fa657f329e 100644
--- a/tests/test_response_policy.py
+++ b/tests/test_response_policy.py
@@ -19,10 +19,6 @@ def selected(audience):
         def recipients(self, message, audience, *, aliases=None):
             return self.selected(audience)
 
-        @classmethod
-        def recipients_sql(cls, mentions, audience):
-            return ("1" if cls.selected(audience) else "0"), ()
-
         def guidance(self, message, *, aliases=None):
             return "review; only reviewers respond"