From 165022a9b0ff85bb03bb8a973ff4ea07e31151f6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 22:49:01 -0400 Subject: [PATCH 1/4] Record native heartbeat liveness ownership repair --- plans/commander-native-plan.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index f9fcec6bd9..464e2f778b 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -211,6 +211,16 @@ 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) + +`HeartbeatSilent.violations` currently reads the newest row in the two-way DM +conversation and then checks its sender. A commander reply after the last beat +therefore hides a silent heartbeat. The existing display-scope family will own +the one-way service-to-commander predicate, applied before the bounded SQL page. +The bus remains the liveness authority; no parallel timestamp, ping store or +scheduler is added. The complete phase and its installed-session/parity evidence +remain open. Source census and the copied-bus check accompany the delivery PR. + ### Phase 8: cutover **Changes.** From 003fd03f5e748f2dbc8ba090953ad9e921fe6b75 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 22:55:43 -0400 Subject: [PATCH 2/4] Select actual heartbeat messages before bounding the liveness read --- src/agent_comms/commander/standing.py | 15 ++++----- src/agent_comms/read_basis.py | 47 ++++++++++++++++++--------- tests/test_holder_standing_rules.py | 4 +++ 3 files changed, 43 insertions(+), 23 deletions(-) diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index bb631e6b25..0a73e5d53b 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -9,18 +9,17 @@ 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.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 DirectedDMDisplayScope, RepliesDisplayScope from agent_comms.recorded_time import RecordedEventTime from agent_comms.registry_document import RegistrySnapshot from agent_comms.standing_rules import ( @@ -355,11 +354,11 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: if CommanderHeartbeat.sender not in comms.registry or ( comms.registry.require(commander).incarnation not in served): 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()) + snapshot = comms.registry.snapshot() + scope = DirectedDMDisplayScope.capture(CommanderHeartbeat.sender, commander, snapshot) + beats = scope.records(comms.bus.bus, descending=True, limit=1) + return tuple(cls(beat.seq, beat.timestamp + cls.seconds) for beat in beats + if beat.timestamp + cls.seconds < time.time()) def began(self) -> float: return self.since diff --git a/src/agent_comms/read_basis.py b/src/agent_comms/read_basis.py index 4faa9e493b..634f79f732 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -880,14 +880,39 @@ def lower_bound(self, cursor: ReadCursor, snapshot: RegistryProvenance) -> int: @dataclass(frozen=True, slots=True) -class DMDisplayScope(MessageDisplayScope): +class DirectedDMDisplayScope(MessageDisplayScope): + """Messages authored by the first participant and addressed to the second.""" + first_names: frozenset[str] second_names: frozenset[str] @classmethod - def capture(cls, a: str, b: str, snapshot: RegistryProvenance) -> DMDisplayScope: + def capture(cls, a: str, b: str, snapshot: RegistryProvenance) -> Self: return cls(snapshot.names_for(a), snapshot.names_for(b)) + @property + def index_predicate(self) -> tuple[str, tuple]: + return self.address_predicate("sender", "target") + + @property + def index_targets(self) -> frozenset[str]: + return self.second_names + + def address_predicate(self, sender: str, target: str) -> tuple[str, tuple]: + first, second = tuple(self.first_names), tuple(self.second_names) + a, b = ",".join("?" for _ in first), ",".join("?" for _ in second) + return f"({sender} IN ({a}) AND {target} IN ({b}))", (*first, *second) + + def addresses(self, sender: str, target: str) -> bool: + return sender in self.first_names and target in self.second_names + + def includes(self, message: Message) -> bool: + return self.addresses(message.sender, message.target) + + +class DMDisplayScope(DirectedDMDisplayScope): + """A conversation includes both directions of its participants' direct messages.""" + @classmethod def capture_human(cls, viewer: str, peer: str, snapshot: RegistrySnapshot) -> DMDisplayScope: try: @@ -900,25 +925,17 @@ def capture_human(cls, viewer: str, peer: str, snapshot: RegistrySnapshot) -> DM raise DMScopeChanged("DM display identity changed; refresh the page.") return cls.capture(first.name, second.name, snapshot) - @property - def index_predicate(self) -> tuple[str, tuple]: - return self.address_predicate("sender", "target") - def address_predicate(self, sender: str, target: str) -> tuple[str, tuple]: - first, second = tuple(self.first_names), tuple(self.second_names) - a, b = ",".join("?" for _ in first), ",".join("?" for _ in second) - return (f"(({sender} IN ({a}) AND {target} IN ({b})) OR " - f"({sender} IN ({b}) AND {target} IN ({a})))", - (*first, *second, *second, *first)) + forward, first = super().address_predicate(sender, target) + reverse, second = super().address_predicate(target, sender) + return f"({forward} OR {reverse})", (*first, *second) @property def index_targets(self) -> frozenset[str]: return self.first_names | self.second_names - def includes(self, message: Message) -> bool: - return (message.sender in self.first_names and message.target in self.second_names) or ( - message.sender in self.second_names and message.target in self.first_names - ) + def addresses(self, sender: str, target: str) -> bool: + return super().addresses(sender, target) or super().addresses(target, sender) @dataclass(frozen=True) diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index a7b7491e5a..e87d3f7ee1 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -570,6 +570,10 @@ 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}") From 83071a0a89d80b48ca42a856ebbe77a6010c89f5 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 23:09:48 -0400 Subject: [PATCH 3/4] Derive missing first heartbeat from the oldest open request --- plans/commander-native-plan.md | 16 +++++++-- src/agent_comms/commander/standing.py | 50 +++++++++++++++++++------- src/agent_comms/commander_heartbeat.py | 11 ++++-- src/agent_comms/read_basis.py | 8 +++-- src/agent_comms/todos.py | 4 +-- tests/test_holder_standing_rules.py | 35 ++++++++++++++++++ 6 files changed, 101 insertions(+), 23 deletions(-) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 464e2f778b..c86f7af8bd 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -217,9 +217,19 @@ The fix is one derived model with four renderers. **Core is the engine, Toad is conversation and then checks its sender. A commander reply after the last beat therefore hides a silent heartbeat. The existing display-scope family will own the one-way service-to-commander predicate, applied before the bounded SQL page. -The bus remains the liveness authority; no parallel timestamp, ping store or -scheduler is added. The complete phase and its installed-session/parity evidence -remain open. Source census and the copied-bus check accompany the delivery PR. +The bus remains the liveness authority. `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. The directional +scope has its own commit; never-started handling follows in a separate commit. +Seventeen focused copied-bus/request/page/channel checks pass (5.73 seconds), +including the real rename and frozen scope family. Installed full-session/parity +acceptance and the complete control-plane phase remain open. The source and +consumer census has zero parse omissions; unprojected declarations remain visible. ### Phase 8: cutover diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 0a73e5d53b..11e60ab55c 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -17,6 +17,7 @@ from agent_comms.message_bus import WatchPass 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 DirectedDMDisplayScope, RepliesDisplayScope @@ -336,38 +337,63 @@ 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): - return () snapshot = comms.registry.snapshot() - scope = DirectedDMDisplayScope.capture(CommanderHeartbeat.sender, commander, snapshot) - beats = scope.records(comms.bus.bus, descending=True, limit=1) - return tuple(cls(beat.seq, beat.timestamp + cls.seconds) for beat in beats - if beat.timestamp + cls.seconds < time.time()) + thread = snapshot.require(commander) + requests = CommanderHeartbeat.requested_by(thread, comms.goals) + if not requests: + return () + request, = requests + beats = () + if snapshot.canonical_name(CommanderHeartbeat.sender) in snapshot.threads: + scope = DirectedDMDisplayScope.capture(CommanderHeartbeat.sender, thread.name, snapshot) + beats = scope.records(comms.bus.bus, descending=True, limit=1) + 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/read_basis.py b/src/agent_comms/read_basis.py index 634f79f732..c5b241dfb6 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -910,6 +910,7 @@ def includes(self, message: Message) -> bool: return self.addresses(message.sender, message.target) +@dataclass(frozen=True, slots=True) class DMDisplayScope(DirectedDMDisplayScope): """A conversation includes both directions of its participants' direct messages.""" @@ -926,8 +927,8 @@ def capture_human(cls, viewer: str, peer: str, snapshot: RegistrySnapshot) -> DM return cls.capture(first.name, second.name, snapshot) def address_predicate(self, sender: str, target: str) -> tuple[str, tuple]: - forward, first = super().address_predicate(sender, target) - reverse, second = super().address_predicate(target, sender) + forward, first = super(DMDisplayScope, self).address_predicate(sender, target) + reverse, second = super(DMDisplayScope, self).address_predicate(target, sender) return f"({forward} OR {reverse})", (*first, *second) @property @@ -935,7 +936,8 @@ def index_targets(self) -> frozenset[str]: return self.first_names | self.second_names def addresses(self, sender: str, target: str) -> bool: - return super().addresses(sender, target) or super().addresses(target, sender) + return (super(DMDisplayScope, self).addresses(sender, target) + or super(DMDisplayScope, self).addresses(target, sender)) @dataclass(frozen=True) 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_holder_standing_rules.py b/tests/test_holder_standing_rules.py index e87d3f7ee1..f2afd46ab6 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, ) @@ -579,6 +581,39 @@ def test_every_pass_beats_and_a_stale_beat_is_heartbeat_silent(held_bus, monkeyp 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 + + def test_one_answer_covers_an_escalation_burst_sent_before_it(held_bus): from agent_comms.commander.vibe_check import VibeCheckSender From 477a44afb3aa92f2886cf023efbd51a900dd5fa1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 23:29:56 -0400 Subject: [PATCH 4/4] Bound newest directed message reads across participant aliases --- plans/commander-native-plan.md | 32 +++++++++++------ src/agent_comms/bus.py | 18 ++++++++++ src/agent_comms/bus_tables.py | 1 + src/agent_comms/commander/standing.py | 7 ++-- src/agent_comms/read_basis.py | 49 ++++++++------------------- tests/test_bus.py | 10 ++++++ tests/test_holder_standing_rules.py | 9 +++++ 7 files changed, 78 insertions(+), 48 deletions(-) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index c86f7af8bd..67e1c07fa8 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -213,23 +213,33 @@ The fix is one derived model with four renderers. **Core is the engine, Toad is ### Current control-plane work: heartbeat source identity (2026-10-23) -`HeartbeatSilent.violations` currently reads the newest row in the two-way DM -conversation and then checks its sender. A commander reply after the last beat -therefore hides a silent heartbeat. The existing display-scope family will own -the one-way service-to-commander predicate, applied before the bounded SQL page. -The bus remains the liveness authority. `CommanderHeartbeat.requested_by` derives +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. The directional -scope has its own commit; never-started handling follows in a separate commit. -Seventeen focused copied-bus/request/page/channel checks pass (5.73 seconds), -including the real rename and frozen scope family. Installed full-session/parity -acceptance and the complete control-plane phase remain open. The source and -consumer census has zero parse omissions; unprojected declarations remain visible. +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 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 11e60ab55c..b0bdb03d46 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -20,7 +20,7 @@ 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 DirectedDMDisplayScope, 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 ( @@ -361,8 +361,9 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: request, = requests beats = () if snapshot.canonical_name(CommanderHeartbeat.sender) in snapshot.threads: - scope = DirectedDMDisplayScope.capture(CommanderHeartbeat.sender, thread.name, snapshot) - beats = scope.records(comms.bus.bus, descending=True, limit=1) + 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 ( diff --git a/src/agent_comms/read_basis.py b/src/agent_comms/read_basis.py index c5b241dfb6..4faa9e493b 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -880,40 +880,14 @@ def lower_bound(self, cursor: ReadCursor, snapshot: RegistryProvenance) -> int: @dataclass(frozen=True, slots=True) -class DirectedDMDisplayScope(MessageDisplayScope): - """Messages authored by the first participant and addressed to the second.""" - +class DMDisplayScope(MessageDisplayScope): first_names: frozenset[str] second_names: frozenset[str] @classmethod - def capture(cls, a: str, b: str, snapshot: RegistryProvenance) -> Self: + def capture(cls, a: str, b: str, snapshot: RegistryProvenance) -> DMDisplayScope: return cls(snapshot.names_for(a), snapshot.names_for(b)) - @property - def index_predicate(self) -> tuple[str, tuple]: - return self.address_predicate("sender", "target") - - @property - def index_targets(self) -> frozenset[str]: - return self.second_names - - def address_predicate(self, sender: str, target: str) -> tuple[str, tuple]: - first, second = tuple(self.first_names), tuple(self.second_names) - a, b = ",".join("?" for _ in first), ",".join("?" for _ in second) - return f"({sender} IN ({a}) AND {target} IN ({b}))", (*first, *second) - - def addresses(self, sender: str, target: str) -> bool: - return sender in self.first_names and target in self.second_names - - def includes(self, message: Message) -> bool: - return self.addresses(message.sender, message.target) - - -@dataclass(frozen=True, slots=True) -class DMDisplayScope(DirectedDMDisplayScope): - """A conversation includes both directions of its participants' direct messages.""" - @classmethod def capture_human(cls, viewer: str, peer: str, snapshot: RegistrySnapshot) -> DMDisplayScope: try: @@ -926,18 +900,25 @@ def capture_human(cls, viewer: str, peer: str, snapshot: RegistrySnapshot) -> DM raise DMScopeChanged("DM display identity changed; refresh the page.") return cls.capture(first.name, second.name, snapshot) + @property + def index_predicate(self) -> tuple[str, tuple]: + return self.address_predicate("sender", "target") + def address_predicate(self, sender: str, target: str) -> tuple[str, tuple]: - forward, first = super(DMDisplayScope, self).address_predicate(sender, target) - reverse, second = super(DMDisplayScope, self).address_predicate(target, sender) - return f"({forward} OR {reverse})", (*first, *second) + first, second = tuple(self.first_names), tuple(self.second_names) + a, b = ",".join("?" for _ in first), ",".join("?" for _ in second) + return (f"(({sender} IN ({a}) AND {target} IN ({b})) OR " + f"({sender} IN ({b}) AND {target} IN ({a})))", + (*first, *second, *second, *first)) @property def index_targets(self) -> frozenset[str]: return self.first_names | self.second_names - def addresses(self, sender: str, target: str) -> bool: - return (super(DMDisplayScope, self).addresses(sender, target) - or super(DMDisplayScope, self).addresses(target, sender)) + def includes(self, message: Message) -> bool: + return (message.sender in self.first_names and message.target in self.second_names) or ( + message.sender in self.second_names and message.target in self.first_names + ) @dataclass(frozen=True) 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 f2afd46ab6..6079509ec7 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -612,6 +612,15 @@ def test_first_heartbeat_is_owed_from_the_oldest_open_request_then_survives_rena 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):