diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md
index f9fcec6bd9..b9a47e1ac9 100644
--- a/plans/commander-native-plan.md
+++ b/plans/commander-native-plan.md
@@ -211,6 +211,112 @@ 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.
+
+Draft #2624 is stacked on the published owner-request port #2621. Eighteen
+focused copied-bus/request/page/channel/source checks pass (7.54 seconds),
+including the newest beat across both participants' renames and later reverse
+replies. On the retained 19,189-record bus copy, an absent service with aliases
+takes 76 SQLite instructions after index reconciliation versus 45,088 before;
+both return no message. All exact-pair queries use the declared covering index.
+Reconciliation preserves every bus sequence and both bus/address row counts;
+integrity and foreign-key checks pass, and a repeated reconciliation is a no-op.
+Its current unconditional projection rebuild costs 39.83 seconds even for this
+index-only change; that installer/schema defect remains open with the commander.
+Installed full-session/parity acceptance and the complete control-plane phase
+remain open. The source census has zero parse omissions; unprojected declarations
+remain visible.
+
+### Reply eligibility work: the message owns the answer (2026-10-23)
+
+`MessageIdentitySource.response_policy` copied the result of
+`Message.role.response_policy`, and `RepliesDisplayScope.response_predicate`
+and `ResponsePolicy.recipients_sql` supplied a second implementation of the
+existing Python eligibility contract. This is IDEN-5/IMPL-5: the message role
+and `RepliesDisplayScope.includes` already determine the answer. The current
+checkpoint deletes that column and SQL family. `AnsweredSource` first joins
+the recorded exact question/reply references, groups by original, and visits
+actual originals lazily in the requested answer order. Greatest reply sequence
+and greatest timestamp are separate selections. Unanswered pages apply the
+same real eligibility owner before counting their page limit. No callback,
+callback cache, fake message, policy switch or new index is added.
+
+Focused copied-bus checks cover aliases, wrong reference hashes, expired
+questions, newer ineligible candidates, non-monotonic reply timestamps and
+eligibility before a one-message page limit. The retained 19,189-record replay
+and source census are recorded in #2630, which is rebased onto the current
+integration branch and carries the accepted heartbeat repair. It excludes
+rejected #2629; removing these mirrors does not establish generic projection
+freshness.
+
+The change also deletes `DirectReplySource`, the other stored answer
+computed by `Delivered.direct_for`. `Goals._goal_input_review` and
+`RecordedGoalWait.has_reply` now consume `BusReader.direct_replies`, which
+checks actual delivery policy before counting the page limit. Frozen
+sender/recipient incarnations, UNKNOWN input selection and review cursors are
+preserved. The shared participant reader returns actual `Delivered` values
+through its existing checked decoder, closing the inaccurate `BusRecord`
+return contract used by `cohort_foreground`.
+
+The original streaming candidate needed 170–601 full delivery decodes for a
+100-reply page. Frozen capability decoding dominated. The implemented query
+reads actual complete `Message` values first and loads frozen deliveries only
+for accepted messages.
+`ResponsePolicy.starts_direct_turn` owns the shared predicate; `ReceivedSource`
+already establishes the frozen recipient incarnation. All sixteen counted
+replays match: a 100-reply page loads exactly 100 full deliveries, with 170–601
+message reads and 148–363 milliseconds in the completed counted replay.
+An earlier cold cell hit its one-second cutoff; that failed run is retained.
+That comparison was against the rejected full-manifest replacement, not the
+original cache. A paired source replay of the actual pre-change
+`received(source=DirectReplySource)` query on the same four saved-range
+100-reply pages costs 52–105 milliseconds with 100 delivery decodes. The new
+query costs 101–218 milliseconds with 170–601 message reads plus 100 delivery
+decodes: 1.3–4.2 times slower in those cases. Removing the stored answer closes
+the maintenance defect; this direct-query runtime regression remains owned by
+the control-plane worker. Neither replay establishes constant cost.
+Ordinary participant pages retain
+their SQL limit. Conversation pages query each declared `ParticipantSource`
+member up to that limit, then combine the bounded candidates before decoding.
+The retained one-message case takes 50 SQLite instructions on both short and
+long sources; the previous UNION materialized 6,050 versus 256,859 instructions.
+
+`tools/cutover/carry_reply_policy.py` retires the exact source-proven old table
+and applies the existing `BusSchema` family carry to the live root and its
+catalog-declared snapshots. The main installation witness and archive catalog
+lock own the work; backup and rewritten paths include every affected copy.
+An actual worker-process carry of the retained bus removes 8,290 cached rows,
+keeps all six durable source-table counts and hashes unchanged, removes the
+old policy column, passes integrity/foreign-key checks and repeats as a no-op.
+It costs 44.02 seconds; no projection skip is restored. A real attached archive
+still validates and browses with identical catalog and read results afterward.
+The cutover is deleted after all buses have run it.
+
+Whole-file ownership remains open with the control-plane owner. The attempted
+complete-catalog check was removed: the converter's `HistoryOffset` must survive
+retry until `BusHistory.after_acceptance`, and private optional awareness owns
+a separate schema view. The repair retains existing partial schema-family
+scopes and drops only the exact proven retired input. Other unclaimed tables
+are preserved; general detection is not closed by this change. The actual
+`OwnerRequests`, `ReplyPolicy`, `BusHistory` order is checked with unconverted
+JSONL and retained conversion maps, through normal after-acceptance retirement.
+The current installed pair remains independent of this draft; installation
+and live commander qualification belong to integration acceptance.
+
 ### Phase 8: cutover
 
 **Changes.**
diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py
index 25f0c9b891..eb513a9136 100644
--- a/src/agent_comms/bus.py
+++ b/src/agent_comms/bus.py
@@ -5,9 +5,10 @@
 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 itertools import islice, product
 from pathlib import Path
 from typing import TYPE_CHECKING, Self
 
@@ -23,6 +24,7 @@
     ClaimOperations,
     HeldResources,
     MessageIdentitySource,
+    ParticipantSource,
     ReceivedSource,
     SentSource,
 )
@@ -231,6 +233,23 @@ def latest_conversations(
                 (*parameters, through),
             )
 
+    def latest_addressed(
+        self, senders: frozenset[str], targets: frozenset[str],
+    ) -> tuple[Message, ...]:
+        """Newest directed message across aliases, with one indexed candidate per pair."""
+        with self.observing() as db:
+            newest = max((
+                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,
+                )
+            ), default=None)
+            if newest is None:
+                return ()
+            header = BusRow.require(db, seq=newest)
+            return (header.kind.read_from(db, header.seq),)
+
     def messages_for_references(
         self, references: Iterable[MessageReference],
     ) -> tuple[Message, ...]:
@@ -248,61 +267,96 @@ 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, ...]:
+        before: int | None = None, descending: bool = True,
+        sequences: tuple[int, ...] = (), limit: int | None = None,
+    ) -> Generator[int, None, None]:
+        """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)
+                f"ORDER BY seq {order}" + (" LIMIT ?" if limit is not None else ""),
+                (*parameters, after, before, *sequences) + ((limit,) if limit is not None else ()),
+            )) 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(
+                    f'SELECT seq FROM "{participant.declared_name}" WHERE lookup=?',
+                    (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(
+    ) -> tuple[Delivered, ...]:
+        with self.reading() as source, closing(source._source_sequences(
             f'SELECT seq FROM "{SentSource.declared_name}" WHERE lookup=?', (lookup,),
-            after=after, before=before, limit=limit, descending=descending,
-        )
+            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,
+    def _received_sequences(
+        self, lookup: str, *, after: int = 0, before: int | None = None,
         sender_lookups: tuple[str, ...] = (), descending: bool = True,
-        source: type[ReceivedSource] = ReceivedSource, sequences: tuple[int, ...] = (),
-    ) -> tuple[BusRecord, ...]:
+        sequences: tuple[int, ...] = (), limit: int | None = None,
+    ) -> Generator[int, None, None]:
         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,
+        return self._source_sequences(
+            f'SELECT seq FROM "{ReceivedSource.declared_name}" WHERE lookup=?{sender}',
+            (lookup, *sender_lookups), after=after, before=before,
+            descending=descending, sequences=sequences, limit=limit,
         )
 
+    def received(
+        self, lookup: str, *, after: int = 0, before: int | None = None, limit: int = 100,
+        sender_lookups: tuple[str, ...] = (), descending: bool = True,
+        sequences: tuple[int, ...] = (),
+    ) -> tuple[Delivered, ...]:
+        with self.reading() as source, closing(source._received_sequences(
+            lookup, after=after, before=before, sender_lookups=sender_lookups,
+            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, before: int | None = None, limit: int = 100,
+        sender_lookups: tuple[str, ...] = (), descending: bool = True,
+        sequences: tuple[int, ...] = (),
+    ) -> tuple[Delivered, ...]:
+        """Read actual message policy before loading accepted frozen deliveries."""
+        with self.reading() as source, source.observing() as db, closing(source._received_sequences(
+            lookup, after=after, before=before, sender_lookups=sender_lookups,
+            descending=descending, sequences=sequences,
+        )) as positions:
+            messages = (Message.read_from(db, seq) for seq in positions)
+            # ReceivedSource already establishes the frozen recipient incarnation.
+            eligible = (message for message in messages
+                        if message.role.response_policy(message).starts_direct_turn)
+            return tuple(source.delivery(message.seq) for message in islice(eligible, limit))
+
     def unanswered_references(
         self, sequences: tuple[int, ...], responders: frozenset[str],
     ) -> tuple[MessageReference, ...]:
diff --git a/src/agent_comms/bus_publication.py b/src/agent_comms/bus_publication.py
index 7b31db09b5..01f80bb5ab 100644
--- a/src/agent_comms/bus_publication.py
+++ b/src/agent_comms/bus_publication.py
@@ -42,14 +42,12 @@ def sender_lookup(self) -> str:
         return self.audience.sender_lookup
 
     def projections(self):
-        from .bus_tables import DirectReplySource, ReceivedSource, SentSource
+        from .bus_tables import 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)
 
     def deliveries(self) -> tuple[Self, ...]:
         return (self,)
@@ -57,7 +55,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..7eea3052ae 100644
--- a/src/agent_comms/bus_tables.py
+++ b/src/agent_comms/bus_tables.py
@@ -3,11 +3,12 @@
 from __future__ import annotations
 
 import sqlite3
-from collections.abc import Iterator
+from collections.abc import Generator, Iterator
 from contextlib import closing
 from dataclasses import Field, dataclass, fields, make_dataclass
 from functools import cache
 from inspect import get_annotations
+from itertools import islice
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self, get_origin, get_type_hints
 
 from agent_comms import messages
@@ -22,7 +23,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,
@@ -61,8 +61,11 @@ def create_schema(cls, db: sqlite3.Connection) -> None:
 class BusSchema(ComposedSchemas, BusTables, 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)]
 
@@ -75,16 +78,11 @@ class ReceivedSource(ParticipantSource):
     """The participant belongs to the frozen delivery audience."""
 
 
-class DirectReplySource(ReceivedSource):
-    """The record's declared reply policy starts a separate direct turn."""
-
-
 @dataclass(frozen=True)
 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 +97,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,9 +137,6 @@ 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",
-        )
         asking = ",".join("?" for _ in requesters)
         replying = ",".join("?" for _ in responders)
         kind_filter = (
@@ -162,76 +147,70 @@ 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 ('
+            + f'AND asked.seq < ? {kind_filter}AND NOT EXISTS ('
             f'SELECT 1 FROM "{cls.declared_name}" AS answered '
             f'JOIN "{MessageIdentitySource.declared_name}" AS author ON author.seq=answered.seq '
             + cls._sql(f'WHERE answered.{{original}}={reference} ')
-            + f'AND author.sender IN ({replying})) ORDER BY asked.seq DESC LIMIT ?'
+            + f'AND author.sender IN ({replying})) ORDER BY asked.seq DESC'
         )
-        sequences = db.execute(query, (
+        with closing(db.execute(query, (
             *requesters, since, before, upper, *(kind.declared_name for kind in kinds),
-            *eligibility, *responders, limit,
-        ))
-        headers = (BusRow.require(db, seq=FieldCodec.decode(int, row[0])) for row in sequences)
-        return tuple(header.kind.read_from(db, header.seq) for header in headers)
+            *responders,
+        ))) as sequences:
+            headers = (BusRow.require(db, seq=FieldCodec.decode(int, row[0])) for row in sequences)
+            originals = (header.kind.read_from(db, header.seq) for header in headers)
+            return tuple(islice((original for original in originals if scope.includes(original)), limit))
 
     @classmethod
-    def _answer_rows(
+    def _answer_values[V](
         cls, db: sqlite3.Connection, scope: RepliesDisplayScope,
-        since: float, selection: str,
-    ) -> sqlite3.Cursor:
+        since: float, selection: str, value_type: type[V],
+    ) -> Generator[V, None, None]:
+        """Visit each linked original once, in the requested answer order."""
         requesters, responders = scope.second_names, scope.first_names
-        eligible, eligibility = MessageIdentitySource.eligible_sql(
-            scope, "asked", "question", "asked_route",
-        )
         asking = ",".join("?" for _ in requesters)
         replying = ",".join("?" for _ in responders)
         identity = MessageIdentitySource.declared_name
         content = messages.Message.table
         query = (
-            'WITH matches AS (SELECT reply.seq, '
-            + content._sql('reply_message.{timestamp} AS timestamp ')
-            + f'FROM "{identity}" AS reply '
-            f'JOIN "{content.declared_name}" AS reply_message USING(seq) '
-            f'WHERE reply.sender IN ({replying}) AND '
-            + content._sql('reply_message.{timestamp} >= ? AND EXISTS (')
-            + f'SELECT 1 FROM "{cls.declared_name}" AS answered '
-            f'JOIN "{identity}" AS asked ON '
+            f'SELECT asked.seq, MAX({selection}) AS value '
+            f'FROM "{identity}" AS asked '
+            f'JOIN "{content.declared_name}" AS question ON question.seq=asked.seq '
+            f'JOIN "{cls.declared_name}" AS answered 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} >= ? ')
-            + f'AND ({eligible}))) ' + selection
+            + f'JOIN "{identity}" AS reply ON reply.seq=answered.seq '
+            f'JOIN "{content.declared_name}" AS reply_message ON reply_message.seq=reply.seq '
+            f'WHERE asked.sender IN ({asking}) AND reply.sender IN ({replying}) '
+            f'AND asked.seq<reply.seq AND '
+            + content._sql('question.{timestamp} >= ? AND reply_message.{timestamp} >= ? ')
+            + 'GROUP BY asked.seq ORDER BY value DESC'
         )
-        return db.execute(query, (*responders, since, *requesters, since, *eligibility))
+        with closing(db.execute(query, (*requesters, *responders, since, since))) as rows:
+            for original, value in rows:
+                header = BusRow.require(db, seq=FieldCodec.decode(int, original))
+                if scope.includes(header.kind.read_from(db, header.seq)):
+                    yield FieldCodec.decode(value_type, value)
 
     @classmethod
     def latest_answer(
         cls, db: sqlite3.Connection, scope: RepliesDisplayScope,
         since: float,
     ) -> tuple[messages.BusRecord, ...]:
-        sequences = cls._answer_rows(
-            db, scope, since,
-            'SELECT seq FROM matches ORDER BY seq DESC LIMIT 1',
-        )
-        headers = (BusRow.require(db, seq=FieldCodec.decode(int, row[0])) for row in sequences)
-        return tuple(header.kind.read_from(db, header.seq) for header in headers)
+        with closing(cls._answer_values(db, scope, since, 'reply.seq', int)) as sequences:
+            headers = (BusRow.require(db, seq=seq) for seq in islice(sequences, 1))
+            return tuple(header.kind.read_from(db, header.seq) for header in headers)
 
     @classmethod
     def answer_timestamp(
         cls, db: sqlite3.Connection, scope: RepliesDisplayScope,
         since: float,
     ) -> float:
-        row = cls._answer_rows(
-            db, scope, since,
-            'SELECT COALESCE(MAX(timestamp),0.0) FROM matches',
-        ).fetchone()
-        return FieldCodec.decode(float, row[0])
+        with closing(cls._answer_values(
+            db, scope, since, messages.Message.table._sql('reply_message.{timestamp}'), float,
+        )) as timestamps:
+            return next(timestamps, 0.0)
 
 
 @dataclass(frozen=True)
@@ -420,3 +399,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..b0bdb03d46 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,64 @@ 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 scan_owner(cls) -> type[StandingRule]:
+        return HeartbeatSilent
+
     @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):
+        snapshot = comms.registry.snapshot()
+        thread = snapshot.require(commander)
+        requests = CommanderHeartbeat.requested_by(thread, comms.goals)
+        if not requests:
             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())
+        request, = requests
+        beats = ()
+        if snapshot.canonical_name(CommanderHeartbeat.sender) in snapshot.threads:
+            beats = comms.bus.bus.latest_addressed(
+                snapshot.names_for(CommanderHeartbeat.sender), snapshot.names_for(thread.name),
+            )
+        found = (tuple(MissedHeartbeat(
+            since=max(request.created_at, beat.timestamp) + cls.seconds, beat=beat.seq,
+        ) for beat in beats) if beats else (
+            HeartbeatNotStarted(since=request.created_at + cls.seconds, request=request.id),
+        ))
+        now = time.time()
+        return tuple(rule for rule in found if isinstance(rule, cls) and rule.since < now)
 
     def began(self) -> float:
         return self.since
 
+
+@dataclass(frozen=True)
+class MissedHeartbeat(HeartbeatSilent):
+    beat: int
+
     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
+
+    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/tests/test_bus.py b/tests/test_bus.py
index 9fd075cb55..09e121e15a 100644
--- a/tests/test_bus.py
+++ b/tests/test_bus.py
@@ -19,7 +19,6 @@
     BusProjectionTables,
     BusRow,
     BusTables,
-    DirectReplySource,
     MessageIdentitySource,
 )
 from agent_comms.coordination_tables.publications import PublicationIntents
@@ -265,9 +264,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 +285,10 @@ 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,)
 
 
 def test_unanswered_reference_selection_requires_the_recorded_original_and_exact_owner_answer(bus):
diff --git a/tests/test_carry_bus_history.py b/tests/test_carry_bus_history.py
index 6ba41228e8..d6c37b02ac 100644
--- a/tests/test_carry_bus_history.py
+++ b/tests/test_carry_bus_history.py
@@ -151,9 +151,13 @@ def test_predecessor_publisher_history_round_trips_into_bus(tmp_path):
     from agent_comms.registry_store import RegistryStore
     from agent_comms.todos import TodoStore
     from cutover.carry_owner_requests import OwnerRequests
+    from cutover.carry_reply_policy import ReplyPolicy, StoredDirectReply
+    from agent_comms.bus_tables import MessageIdentitySource
+    from conftest import maintenance_held
 
     chain = CarryChain.at(REPOSITORY)
     assert chain.steps.index(OwnerRequests) < chain.steps.index(BusHistory)
+    assert chain.steps.index(OwnerRequests) < chain.steps.index(ReplyPolicy) < chain.steps.index(BusHistory)
 
     generated = predecessor_history(tmp_path)
     original = generated / "destination"
@@ -171,6 +175,9 @@ def test_predecessor_publisher_history_round_trips_into_bus(tmp_path):
     modes = {relative: (root / relative / BusHistory.filename).stat().st_mode
              for relative in relatives}
     request = BusHistory(root, original, spill)
+    policy = ReplyPolicy(root, original, spill)
+    # No SQLite destination exists yet; the later history writer uses current declarations.
+    assert policy.carry(apply=False) == Carried(0)
     pending = request.carry(apply=False)
     assert pending.changed > 0
     assert pending.detail == dict(rows=8, continuous=8, round_tripped=8)
@@ -187,6 +194,7 @@ def test_predecessor_publisher_history_round_trips_into_bus(tmp_path):
         request_result = requests.carry(apply=True)
         assert request_result.detail == request_dry_run.detail
         assert requests.carry(apply=False).changed == 0
+        assert policy.carry(apply=True) == Carried(0)
     result_path = tmp_path / "carried.json"
     with hold.holding():
         completed = subprocess.run(
@@ -250,6 +258,30 @@ def test_predecessor_publisher_history_round_trips_into_bus(tmp_path):
         assert ReadLedger.at(directory).read() == replace(
             before[relative], bus_identity=RecordedBusIdentity.capture(bus.path))
         assert BusHistory.rebind_reads(directory, before[relative].bus_identity).changed == 0
+    # Exercise a retained conversion map together with the actual retired policy shape.
+    for relative in relatives:
+        with Bus(Bus.path_in(root / relative)).writing() as db:
+            StoredDirectReply.create(db)
+            StoredDirectReply("retained-recipient", 1).insert(db)
+            db.execute('ALTER TABLE "message_identity_source" ADD COLUMN response_policy TEXT')
+    policy_pending = policy.carry(apply=False)
+    assert policy_pending.changed == len(relatives)
+    with maintenance_held(root):
+        assert policy.carry(apply=True) == policy_pending
+        assert policy.carry(apply=True) == Carried(0)
+    for relative in relatives:
+        directory = root / relative
+        bus = Bus(Bus.path_in(directory))
+        source = directory / BusHistory.filename
+        metadata = HistoryMetadata.read(directory)
+        with bus.observing() as db:
+            assert not StoredDirectReply.present(db)
+            assert MessageIdentitySource.stored_columns(db) == ("seq", "message_id", "sender")
+            assert HistoryOffset.count(db) == 4
+            assert all(HistoryOffset.select(db, offset=line.offset) == [
+                HistoryOffset(line.offset, line.row.seq)]
+                for line in HistoryLine.rows(source, metadata))
+        assert BusHistory.transfer(source, bus, metadata).changed == 0
     converted, = HistoryArchive(Bus.path_in(root)).sources()
     converted.validate()
     assert converted.root == str(root / archived.relative_to(original))
diff --git a/tests/test_carry_reply_policy.py b/tests/test_carry_reply_policy.py
new file mode 100644
index 0000000000..3d17aa4b51
--- /dev/null
+++ b/tests/test_carry_reply_policy.py
@@ -0,0 +1,87 @@
+"""One-time cache retirement preserves unrelated state and rejects unrecognized input."""
+
+from pathlib import Path
+
+import pytest
+
+from agent_comms.bus import Bus, PrivateRecord
+from agent_comms.bus_tables import BusSchema
+from agent_comms.errors import RelationViolationError
+from agent_comms.installer.carry import Carried
+from agent_comms.historical_views import HistoryArchive
+from agent_comms.maintenance_barrier import MaintenanceNotHeldError
+from agent_comms.messages import InfoKind, Message
+from bus_record_fixture import record_from
+from conftest import maintenance_held
+from test_historical_views import collect, migrated
+from cutover.carry_reply_policy import ReplyPolicy, StoredDirectReply
+
+
+def test_retired_reply_policy_carry_is_dry_idempotent_and_source_preserving(bus, tmp_path):
+    reply = bus.append(lambda seq, db: record_from(
+        PrivateRecord, Message("author", "worker", "Retain the actual delivery", InfoKind, seq=seq),
+    ))
+    with bus.writing() as db:
+        StoredDirectReply.create(db)
+        StoredDirectReply("b" * 32, reply.seq).insert(db)
+        db.execute("CREATE TABLE unknown_owner (payload TEXT)")
+        db.execute("INSERT INTO unknown_owner VALUES ('unclassified')")
+    step = ReplyPolicy(bus.path.parent, bus.path.parent, tmp_path / "spill")
+    assert step.backup_files() == (bus.path,)
+    pending = step.carry(apply=False)
+    assert pending == Carried(1, {"retired_direct_replies": 1}, (bus.path.name,))
+    with bus.observing() as db:
+        assert StoredDirectReply.count(db) == 1
+    with pytest.raises(MaintenanceNotHeldError):
+        step.carry(apply=True)
+    with maintenance_held(bus.path.parent):
+        assert step.carry(apply=True) == pending
+        assert step.carry(apply=True) == Carried(0)
+    with bus.observing() as db:
+        assert not StoredDirectReply.present(db)
+        value, = db.execute("SELECT payload FROM unknown_owner").fetchone()
+        assert value == "unclassified"
+        BusSchema.require_schema(db)
+    assert bus.delivery(reply.seq) == reply
+
+
+def test_retired_reply_policy_carry_rejects_an_unrecognized_input_shape(bus, tmp_path):
+    with bus.writing() as db:
+        StoredDirectReply.create(db)
+        db.execute('ALTER TABLE "direct_reply_source" ADD COLUMN unrelated TEXT')
+    step = ReplyPolicy(bus.path.parent, bus.path.parent, tmp_path / "spill")
+    for apply in (False, True):
+        with pytest.raises(RelationViolationError):
+            step.carry(apply=apply)
+    with bus.observing() as db:
+        assert StoredDirectReply.present(db)
+
+
+def test_retired_reply_policy_carries_catalog_snapshots_without_resealing(migrated, tmp_path):
+    _, _, live, snapshot = migrated
+    archived = Bus(Bus.path_in(Path(snapshot.root)))
+    selected = (live.bus.bus, archived)
+    for source in selected:
+        original = source.delivery(source.revision())
+        with source.writing() as db:
+            StoredDirectReply.create(db)
+            StoredDirectReply(original.audience.sender_lookup, original.seq).insert(db)
+            db.execute('ALTER TABLE "message_identity_source" ADD COLUMN response_policy TEXT')
+    catalog = HistoryArchive.path_in(live.root)
+    before_catalog = catalog.read_bytes()
+    viewer = live.views.display_viewer(str(live.root))
+    before_rows = collect(live, "channel_display_page", "#team", viewer=viewer)
+    step = ReplyPolicy(live.root, live.root, tmp_path / "spill")
+    assert {source.path for source in selected} <= set(step.backup_files())
+    pending = step.carry(apply=False)
+    assert pending.changed == 2
+    assert set(pending.rewritten) == {str(source.path.relative_to(live.root)) for source in selected}
+    with maintenance_held(live.root):
+        assert step.carry(apply=True) == pending
+    for source in selected:
+        with source.observing() as db:
+            assert not StoredDirectReply.present(db)
+            BusSchema.require_schema(db)
+    snapshot.validate()
+    assert catalog.read_bytes() == before_catalog
+    assert collect(live, "channel_display_page", "#team", viewer=viewer) == before_rows
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..e3ac408bcf 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,))
@@ -644,11 +655,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"
 
diff --git a/tools/cutover/carry_reply_policy.py b/tools/cutover/carry_reply_policy.py
new file mode 100644
index 0000000000..6ee2e0d3e3
--- /dev/null
+++ b/tools/cutover/carry_reply_policy.py
@@ -0,0 +1,72 @@
+"""Retire the stored direct-reply answer before reconciling the message-owned policy.
+
+Delete this step after every bus has run it. The exact input declaration comes
+from DirectReplySource/ParticipantSource at 477a44afb; it is never part of the
+incoming BusProjectionTables family.
+"""
+
+from __future__ import annotations
+
+from contextlib import closing
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Annotated
+
+from agent_comms.bus import Bus
+from agent_comms.bus_tables import BusSchema
+from agent_comms.historical_views import HistoryArchive
+from agent_comms.installer.carry import Carried, Carry
+from agent_comms.locked_store import SQLiteFile
+from agent_comms.store_files import _store_lock
+from agent_comms.typed_table import (
+    Column,
+    DerivedTables,
+    SharedFileTables,
+    SQLiteTableName,
+    TypedTable,
+)
+from .carry_bus_history import CarryCatalogSource
+
+
+@dataclass(frozen=True)
+class StoredDirectReply(TypedTable, SharedFileTables, DerivedTables, declared_name="direct_reply_source"):
+    """Only the removed input shape; no incoming schema family owns this row."""
+
+    lookup: Annotated[str, Column(primary_key=True)]
+    seq: Annotated[int, Column(primary_key=True)]
+
+
+class ReplyPolicy(Carry):
+    def backup_files(self) -> tuple[Path, ...]:
+        sources = CarryCatalogSource.read(self.root, self.original_root)
+        roots = (self.root, *(source.mapped_root(self.root, self.original_root) for source in sources))
+        paths = tuple(dict.fromkeys(Bus.path_in(root) for root in roots))
+        for path in paths:
+            self.require_confined(self.root, path)
+        return tuple(path for path in paths if path.exists())
+
+    def carry(self, *, apply: bool) -> Carried:
+        changed = rows = 0
+        rewritten = []
+        with _store_lock(HistoryArchive.path_in(self.root)):
+            for path in self.backup_files():
+                with closing(SQLiteFile.read_snapshot(path)) as db:
+                    present = StoredDirectReply.present(db)
+                    if not present and not BusSchema.schema_differs(db):
+                        continue
+                    if present:
+                        StoredDirectReply.require_schema(db)
+                        rows += StoredDirectReply.count(db)
+                if apply:
+                    # The installer's main-root hold also owns its retained sources.
+                    Bus.schema_change(Bus.path_in(self.root)).require()
+                    with closing(SQLiteFile.writer(path, isolation_level=None)) as db:
+                        with SQLiteFile.write_transaction(db):
+                            if present:
+                                StoredDirectReply.require_schema(db)
+                                SQLiteTableName(StoredDirectReply.declared_name).drop(db)
+                            # Catalog snapshots use the same current family carry as main.
+                            BusSchema.carry_schema(db)
+                changed += 1
+                rewritten.append(str(path.relative_to(self.root)))
+        return Carried(changed, {"retired_direct_replies": rows}, tuple(rewritten)) if changed else Carried(0)