Use message owners for heartbeat liveness and reply eligibility

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.**

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

src/agent_comms/bus_publication.py

@@ -42,22 +42,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 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,)
 
     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, ...]: