diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index f9fcec6bd9..67e1c07fa8 100644 --- a/plans/commander-native-plan.md +++ b/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.** diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 25f0c9b891..59d6ca2e17 100644 --- a/src/agent_comms/bus.py +++ b/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, ...]: diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index fa211330fe..954f91b6cd 100644 --- a/src/agent_comms/bus_tables.py +++ b/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")),) 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/todos.py b/src/agent_comms/todos.py index aa9e5660ac..cc0e68b088 100644 --- a/src/agent_comms/todos.py +++ b/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: diff --git a/tests/test_bus.py b/tests/test_bus.py index 9fd075cb55..6264c2d67e 100644 --- a/tests/test_bus.py +++ b/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) diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index a7b7491e5a..6079509ec7 100644 --- a/tests/test_holder_standing_rules.py +++ b/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