Detect missing commander heartbeats from request and message owners

plans/commander-native-plan.md

@@ -211,6 +211,36 @@ 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.
+
 ### Phase 8: cutover
 
 **Changes.**

src/agent_comms/bus.py

@@ -8,6 +8,7 @@
 from collections.abc import Callable, Iterable, Iterator
 from contextlib import AbstractContextManager, ExitStack, closing, contextmanager
 from dataclasses import dataclass, replace
+from itertools import product
 from pathlib import Path
 from typing import TYPE_CHECKING, Self
 
@@ -231,6 +232,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, ...]:

src/agent_comms/bus_tables.py

@@ -425,3 +425,4 @@ def append_to(self, db: sqlite3.Connection, seq: int) -> int:
 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")),)

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

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

src/agent_comms/todos.py

@@ -3544,13 +3544,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:

tests/test_bus.py

@@ -268,6 +268,16 @@ def test_scoped_and_frozen_source_queries_are_sequence_bounded(bus):
     assert bus.received("b" * 32, after=original.seq, limit=1, descending=False,
                         sender_lookups=("a" * 32,), source=DirectReplySource) == (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)

tests/test_holder_standing_rules.py

@@ -20,6 +20,8 @@
 from agent_comms.commander.questions import UnansweredBurst, UnansweredUnitAsks
 from agent_comms.commander.standing import (
     HeartbeatSilent,
+    HeartbeatNotStarted,
+    MissedHeartbeat,
     AnswersTo,
     UnitAskUnanswered,
 )
@@ -570,11 +572,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,
+    ), 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(),
+    ), 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