From d59c4942bc7237455cd3884a3c7d31002608251d Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 22:49:01 -0400 Subject: [PATCH 01/10] 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 94796dd7235f682e11aa9a58806545e3e5335fc6 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 22:55:43 -0400 Subject: [PATCH 02/10] 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 8b2b26dcd4..030b7c7b56 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -883,14 +883,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: @@ -903,25 +928,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 67bedee42c..3a2bac67e8 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -644,6 +644,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 a2e1d342aa14899115fd34e1ad61786a8ac684e8 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 23:09:48 -0400 Subject: [PATCH 03/10] 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 030b7c7b56..ccc56f487a 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -913,6 +913,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.""" @@ -929,8 +930,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 @@ -938,7 +939,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 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_holder_standing_rules.py b/tests/test_holder_standing_rules.py index 3a2bac67e8..ea6bd988c7 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, ) @@ -653,6 +655,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 54692416618c8b3ec020ce7ce8304645542956f5 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Thu, 22 Oct 2026 23:29:56 -0400 Subject: [PATCH 04/10] 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 d1b1968402..8b92ab6eb4 100644 --- a/src/agent_comms/bus_tables.py +++ b/src/agent_comms/bus_tables.py @@ -420,3 +420,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 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 ccc56f487a..8b2b26dcd4 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -883,40 +883,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: @@ -929,18 +903,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 ea6bd988c7..5195f7df8c 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -686,6 +686,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): From 5831c901d2ef7494694ad2f328a3f166b8b5c2ca Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 00:23:09 -0400 Subject: [PATCH 05/10] Read reply eligibility from linked original messages --- plans/commander-native-plan.md | 30 ++++++++++ src/agent_comms/bus_tables.py | 92 +++++++++++------------------ src/agent_comms/messages.py | 4 +- src/agent_comms/read_basis.py | 19 ------ src/agent_comms/response_policy.py | 25 -------- tests/test_holder_standing_rules.py | 11 +++- tests/test_response_policy.py | 4 -- 7 files changed, 77 insertions(+), 108 deletions(-) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 67e1c07fa8..79364df588 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -241,6 +241,36 @@ 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. + +Eleven focused copied-bus checks pass, including aliases, wrong reference +hashes, an older question outside the time cut, newer ineligible candidates, +non-monotonic reply timestamps and an unanswered page limited to one eligible +original. The retained 19,189-record replay and source census are recorded in +the PR. This branch starts at accepted #2624 and excludes rejected #2629; +removing this policy mirror does not establish generic projection freshness. + +The computed-policy family is still open: `Delivered.projections` also stores +`DirectReplySource` membership computed by `Delivered.direct_for`. +`Goals._goal_input_review` and `RecordedGoalWait.has_reply` consume +that membership through `BusReader.received`. Their migration must preserve +the recorded sender/recipient incarnations, UNKNOWN inputs and review cursor, +and apply actual `direct_for` before the page limit. The control-plane owner +keeps this repair; the MI worker owns the overlapping received/decode contract. +The current installed pair remains independent of this draft. + ### Phase 8: cutover **Changes.** diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index 8b92ab6eb4..a9a743500e 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, @@ -84,7 +84,6 @@ 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 +98,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 +138,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 +148,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= ? ') - + 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= ? 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) 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/read_basis.py b/src/agent_comms/read_basis.py index 8b2b26dcd4..a32e8cb9c3 100644 --- a/src/agent_comms/read_basis.py +++ b/src/agent_comms/read_basis.py @@ -939,25 +939,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..28b97dacf9 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 @@ -36,11 +34,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 +54,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 +79,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 +89,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/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index 5195f7df8c..beb9a06f26 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -378,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): @@ -398,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,)) 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" From 6ab97ff4ba18029e950216f3f9bf19f29d8846cb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 00:32:32 -0400 Subject: [PATCH 06/10] Stream direct replies from actual frozen deliveries --- plans/commander-native-plan.md | 25 ++++++--- src/agent_comms/bus.py | 80 +++++++++++++++++++---------- src/agent_comms/bus_publication.py | 4 +- src/agent_comms/bus_tables.py | 4 -- src/agent_comms/goal_management.py | 11 ++-- src/agent_comms/goal_waits.py | 5 +- tests/test_bus.py | 5 +- tests/test_goal_input_review.py | 9 ++-- tests/test_goal_standby_liveness.py | 2 +- 9 files changed, 84 insertions(+), 61 deletions(-) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 79364df588..3386ee077f 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -262,14 +262,23 @@ original. The retained 19,189-record replay and source census are recorded in the PR. This branch starts at accepted #2624 and excludes rejected #2629; removing this policy mirror does not establish generic projection freshness. -The computed-policy family is still open: `Delivered.projections` also stores -`DirectReplySource` membership computed by `Delivered.direct_for`. -`Goals._goal_input_review` and `RecordedGoalWait.has_reply` consume -that membership through `BusReader.received`. Their migration must preserve -the recorded sender/recipient incarnations, UNKNOWN inputs and review cursor, -and apply actual `direct_for` before the page limit. The control-plane owner -keeps this repair; the MI worker owns the overlapping received/decode contract. -The current installed pair remains independent of this draft. +The next checkpoint 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`. + +This streaming candidate is not performance-qualified. All sixteen retained +four-pair direct-reply selections match the stored table, but a 100-reply +page requires 170–601 full delivery decodes and costs 101–988 milliseconds. +The same 601 records cost 1,106 milliseconds as full deliveries versus 133 +milliseconds as complete `Message` values: frozen delivery capability decoding +is the dominant term. The control-plane owner is repairing that query/decode +boundary without another policy copy. The current installed pair remains +independent of this draft. ### Phase 8: cutover diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 59d6ca2e17..1c203a0e36 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -5,10 +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 product +from itertools import islice, product from pathlib import Path from typing import TYPE_CHECKING, Self @@ -266,61 +266,87 @@ 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_records( self, query: str, parameters: tuple[object, ...], *, after: int = 0, - before: int | None = None, limit: int = 100, descending: bool = True, + before: int | None = None, descending: bool = True, sequences: tuple[int, ...] = (), - ) -> tuple[BusRecord, ...]: - with self.observing() as db: + ) -> Generator[Delivered, None, None]: + """Stream frozen participant rows as actual deliveries on one held snapshot.""" + with self.reading() as source, source.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}", + (*parameters, after, before, *sequences), + )) as rows: + for row in rows: + yield source.delivery(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( + ) -> tuple[Delivered, ...]: + with closing(self._source_records( 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, + (lookup, lookup), after=after, before=before, descending=descending, sequences=sequences, - ) + )) as records: + return tuple(islice(records, 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 closing(self._source_records( f'SELECT seq FROM "{SentSource.declared_name}" WHERE lookup=?', (lookup,), - after=after, before=before, limit=limit, descending=descending, - ) + after=after, before=before, descending=descending, + )) as records: + return tuple(islice(records, limit)) - def received( - self, lookup: str, *, after: int = 0, before: int | None = None, limit: int = 100, + def _received_records( + 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, ...] = (), + ) -> Generator[Delivered, 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, + return self._source_records( + f'SELECT seq FROM "{ReceivedSource.declared_name}" WHERE lookup=?{sender}', + (lookup, *sender_lookups), after=after, before=before, descending=descending, sequences=sequences, ) + 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 closing(self._received_records( + lookup, after=after, before=before, sender_lookups=sender_lookups, + descending=descending, sequences=sequences, + )) as records: + return tuple(islice(records, limit)) + + 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, ...]: + """Count actual direct replies after checking each delivery's own policy.""" + with closing(self._received_records( + lookup, after=after, before=before, sender_lookups=sender_lookups, + descending=descending, sequences=sequences, + )) as records: + return tuple(islice((record for record in records if record.direct_for(lookup)), 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..22fbb9b0ea 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,) diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index a9a743500e..a76a529978 100644 --- a/src/agent_comms/bus_tables.py +++ b/src/agent_comms/bus_tables.py @@ -75,10 +75,6 @@ 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.""" 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/tests/test_bus.py b/tests/test_bus.py index 6264c2d67e..db45a47b2b 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,8 +264,8 @@ 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,) 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) == () From 298050738c0d009523be9582109a259944cdcb79 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 00:40:02 -0400 Subject: [PATCH 07/10] Read actual message policy before frozen delivery manifests --- plans/commander-native-plan.md | 13 +++++++-- src/agent_comms/bus.py | 46 ++++++++++++++++-------------- src/agent_comms/bus_publication.py | 2 +- src/agent_comms/response_policy.py | 4 +++ 4 files changed, 40 insertions(+), 25 deletions(-) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 3386ee077f..88e72ab489 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -276,9 +276,16 @@ four-pair direct-reply selections match the stored table, but a 100-reply page requires 170–601 full delivery decodes and costs 101–988 milliseconds. The same 601 records cost 1,106 milliseconds as full deliveries versus 133 milliseconds as complete `Message` values: frozen delivery capability decoding -is the dominant term. The control-plane owner is repairing that query/decode -boundary without another policy copy. The current installed pair remains -independent of this draft. +is the dominant term. The next candidate reads actual complete `Message` +values first and loads the frozen delivery 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. +Neither replay establishes constant cost. Physical retirement of the removed +table remains open through the existing one-shot cutover owner. The current +installed pair remains independent of this draft. ### Phase 8: cutover diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 1c203a0e36..9ffddf83de 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -266,13 +266,13 @@ def deliveries_for_references( return tuple(delivery for record in self.messages_for_references(references) for delivery in record.deliveries()) - def _source_records( + def _source_sequences( self, query: str, parameters: tuple[object, ...], *, after: int = 0, before: int | None = None, descending: bool = True, sequences: tuple[int, ...] = (), - ) -> Generator[Delivered, None, None]: - """Stream frozen participant rows as actual deliveries on one held snapshot.""" - with self.reading() as source, source.observing() as db: + ) -> 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: @@ -284,41 +284,41 @@ def _source_records( (*parameters, after, before, *sequences), )) as rows: for row in rows: - yield source.delivery(FieldCodec.decode(int, row[0])) + 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[Delivered, ...]: - with closing(self._source_records( + with self.reading() as source, closing(source._source_sequences( 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, descending=descending, sequences=sequences, - )) as records: - return tuple(islice(records, limit)) + )) as positions: + return tuple(source.delivery(seq) for seq in islice(positions, limit)) def sent( self, lookup: str, *, after: int = 0, before: int | None = None, limit: int = 100, descending: bool = True, ) -> tuple[Delivered, ...]: - with closing(self._source_records( + with self.reading() as source, closing(source._source_sequences( f'SELECT seq FROM "{SentSource.declared_name}" WHERE lookup=?', (lookup,), after=after, before=before, descending=descending, - )) as records: - return tuple(islice(records, limit)) + )) as positions: + return tuple(source.delivery(seq) for seq in islice(positions, limit)) - def _received_records( + def _received_sequences( self, lookup: str, *, after: int = 0, before: int | None = None, sender_lookups: tuple[str, ...] = (), descending: bool = True, sequences: tuple[int, ...] = (), - ) -> Generator[Delivered, 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_records( + 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, @@ -329,23 +329,27 @@ def received( sender_lookups: tuple[str, ...] = (), descending: bool = True, sequences: tuple[int, ...] = (), ) -> tuple[Delivered, ...]: - with closing(self._received_records( + with self.reading() as source, closing(source._received_sequences( lookup, after=after, before=before, sender_lookups=sender_lookups, descending=descending, sequences=sequences, - )) as records: - return tuple(islice(records, limit)) + )) as positions: + return tuple(source.delivery(seq) for seq in islice(positions, limit)) 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, ...]: - """Count actual direct replies after checking each delivery's own policy.""" - with closing(self._received_records( + """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 records: - return tuple(islice((record for record in records if record.direct_for(lookup)), limit)) + )) 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], diff --git a/src/agent_comms/bus_publication.py b/src/agent_comms/bus_publication.py index 22fbb9b0ea..01f80bb5ab 100644 --- a/src/agent_comms/bus_publication.py +++ b/src/agent_comms/bus_publication.py @@ -55,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/response_policy.py b/src/agent_comms/response_policy.py index 28b97dacf9..82ef826e5b 100644 --- a/src/agent_comms/response_policy.py +++ b/src/agent_comms/response_policy.py @@ -20,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: From 4cc95afcabf43dd4dcf9d223de47102325ea7447 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 00:58:44 -0400 Subject: [PATCH 08/10] Retire policy caches through the existing bus carry and bound participant pages --- plans/commander-native-plan.md | 54 ++++++--- src/agent_comms/bus.py | 38 ++++--- src/agent_comms/bus_tables.py | 19 ++++ src/agent_comms/private_runtime_schema.py | 4 +- src/agent_comms/read_basis.py | 1 - tests/test_bus.py | 4 + tests/test_carry_reply_policy.py | 128 ++++++++++++++++++++++ tests/test_holder_standing_rules.py | 4 +- tests/test_private_runtime_bootstrap.py | 1 + tools/cutover/carry_reply_policy.py | 72 ++++++++++++ 10 files changed, 286 insertions(+), 39 deletions(-) create mode 100644 tests/test_carry_reply_policy.py create mode 100644 tools/cutover/carry_reply_policy.py diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 88e72ab489..58c3822462 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -255,14 +255,15 @@ 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. -Eleven focused copied-bus checks pass, including aliases, wrong reference -hashes, an older question outside the time cut, newer ineligible candidates, -non-monotonic reply timestamps and an unanswered page limited to one eligible -original. The retained 19,189-record replay and source census are recorded in -the PR. This branch starts at accepted #2624 and excludes rejected #2629; -removing this policy mirror does not establish generic projection freshness. - -The next checkpoint deletes `DirectReplySource`, the other stored answer +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 @@ -271,21 +272,38 @@ preserved. The shared participant reader returns actual `Delivered` values through its existing checked decoder, closing the inaccurate `BusRecord` return contract used by `cohort_foreground`. -This streaming candidate is not performance-qualified. All sixteen retained -four-pair direct-reply selections match the stored table, but a 100-reply -page requires 170–601 full delivery decodes and costs 101–988 milliseconds. -The same 601 records cost 1,106 milliseconds as full deliveries versus 133 -milliseconds as complete `Message` values: frozen delivery capability decoding -is the dominant term. The next candidate reads actual complete `Message` -values first and loads the frozen delivery only for accepted messages. +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. -Neither replay establishes constant cost. Physical retirement of the removed -table remains open through the existing one-shot cutover owner. The current -installed pair remains independent of this draft. +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. + +`BusSchema` now compares the complete catalog and rejects unknown tables without +deleting them. Private coordinator composition remains its original mandatory +view: optional awareness has a separate owner and absence/drift contract. +Configured optional awareness is covered by a real coordinator check. The +current installed pair remains independent of this draft; installation and +live commander qualification belong to integration acceptance. ### Phase 8: cutover diff --git a/src/agent_comms/bus.py b/src/agent_comms/bus.py index 9ffddf83de..eb513a9136 100644 --- a/src/agent_comms/bus.py +++ b/src/agent_comms/bus.py @@ -24,6 +24,7 @@ ClaimOperations, HeldResources, MessageIdentitySource, + ParticipantSource, ReceivedSource, SentSource, ) @@ -269,7 +270,7 @@ def deliveries_for_references( def _source_sequences( self, query: str, parameters: tuple[object, ...], *, after: int = 0, before: int | None = None, descending: bool = True, - sequences: tuple[int, ...] = (), + 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: @@ -280,8 +281,8 @@ def _source_sequences( order = "DESC" if descending else "ASC" with closing(db.execute( f"SELECT seq FROM ({query}) WHERE seq > ? AND seq < ?{selected} " - f"ORDER BY seq {order}", - (*parameters, after, before, *sequences), + 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]) @@ -290,13 +291,18 @@ def conversation( self, lookup: str, *, after: int = 0, before: int | None = None, limit: int = 100, descending: bool = True, sequences: tuple[int, ...] = (), ) -> tuple[Delivered, ...]: - with self.reading() as source, closing(source._source_sequences( - 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, - descending=descending, sequences=sequences, - )) as positions: - return tuple(source.delivery(seq) for seq in islice(positions, limit)) + 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, @@ -304,14 +310,14 @@ def sent( ) -> 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, descending=descending, + after=after, before=before, descending=descending, limit=limit, )) as positions: - return tuple(source.delivery(seq) for seq in islice(positions, limit)) + return tuple(source.delivery(seq) for seq in positions) def _received_sequences( self, lookup: str, *, after: int = 0, before: int | None = None, sender_lookups: tuple[str, ...] = (), descending: bool = True, - sequences: tuple[int, ...] = (), + sequences: tuple[int, ...] = (), limit: int | None = None, ) -> Generator[int, None, None]: sender = "" if sender_lookups: @@ -321,7 +327,7 @@ def _received_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, + descending=descending, sequences=sequences, limit=limit, ) def received( @@ -331,9 +337,9 @@ def received( ) -> tuple[Delivered, ...]: with self.reading() as source, closing(source._received_sequences( lookup, after=after, before=before, sender_lookups=sender_lookups, - descending=descending, sequences=sequences, + descending=descending, sequences=sequences, limit=limit, )) as positions: - return tuple(source.delivery(seq) for seq in islice(positions, limit)) + 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, diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index a76a529978..bd873a01b0 100644 --- a/src/agent_comms/bus_tables.py +++ b/src/agent_comms/bus_tables.py @@ -32,6 +32,9 @@ ForeignKey, Index, SharedFileTables, + SQLiteSchemaObject, + SQLiteTableName, + TableSchema, TypedRow, TypedTable, sql_literal, @@ -61,8 +64,24 @@ 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.""" + @classmethod + def stored_schema(cls, db: sqlite3.Connection) -> list[SQLiteSchemaObject]: + """This complete file sees objects its partial family views cannot name.""" + return TableSchema.stored_schema(db) + + @classmethod + def stored_tables(cls, db: sqlite3.Connection) -> list[SQLiteTableName]: + return TableSchema.stored_tables(db) + + @classmethod + def carry_schema(cls, db: sqlite3.Connection) -> None: + super().carry_schema(db) + cls.require_schema(db) + @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)] 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 a32e8cb9c3..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"): diff --git a/tests/test_bus.py b/tests/test_bus.py index db45a47b2b..09e121e15a 100644 --- a/tests/test_bus.py +++ b/tests/test_bus.py @@ -285,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_reply_policy.py b/tests/test_carry_reply_policy.py new file mode 100644 index 0000000000..3467e48225 --- /dev/null +++ b/tests/test_carry_reply_policy.py @@ -0,0 +1,128 @@ +"""One-time cache retirement preserves source rows and refuses unknown ownership.""" + +import sqlite3 +from pathlib import Path + +import pytest + +from agent_comms.bus import Bus, PrivateRecord +from agent_comms.bus_tables import BusSchema +from agent_comms.cohort_schema import AwarenessTable +from agent_comms.coordinator import Coordination +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 agent_comms.private_runtime_schema import PrivateCoordinatorSchema, PrivateCoordinatorTable +from agent_comms.typed_table import SQLiteTableName, WriterOwnedFile +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_complete_file_reconcile_preserves_and_rejects_unclaimed_tables(): + schema = BusSchema + with sqlite3.connect(":memory:") as db: + schema.create_schema(db) + assert not schema.schema_differs(db) + db.execute("CREATE TABLE unknown_owner (payload TEXT)") + db.execute("INSERT INTO unknown_owner VALUES ('retain this input')") + db.commit() + assert schema.schema_differs(db) + with pytest.raises(schema.schema_error): + schema.reconcile_schema(db, WriterOwnedFile()) + assert db.execute("SELECT payload FROM unknown_owner").fetchall() == [("retain this input",)] + + +def test_mandatory_private_view_preserves_optional_awareness_ownership(tmp_path): + with Coordination(str(tmp_path / "private.sqlite3")) as store: + PrivateCoordinatorTable.install(store) + db = store.session._connection + assert AwarenessTable.stored_schema(db) + PrivateCoordinatorSchema.require_schema(db) + optional, = AwarenessTable.schema_rows() + with store.session.writing() as writer: + SQLiteTableName(optional.declared_name).drop(writer) + assert AwarenessTable.schema_differs(db) + PrivateCoordinatorSchema.require_schema(db) + + +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): + with pytest.raises(RelationViolationError): + step.carry(apply=True) + with bus.observing() as db: + assert StoredDirectReply.present(db) + value, = db.execute("SELECT payload FROM unknown_owner").fetchone() + assert value == "unclassified" + with pytest.raises(RelationViolationError): + BusSchema.require_schema(db) + assert bus.delivery(reply.seq) == reply + with bus.writing() as db: + SQLiteTableName("unknown_owner").drop(db) + 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) + BusSchema.require_schema(db) + + +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_holder_standing_rules.py b/tests/test_holder_standing_rules.py index beb9a06f26..e3ac408bcf 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -685,7 +685,7 @@ def test_first_heartbeat_is_owed_from_the_oldest_open_request_then_survives_rena 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()) + ), destination=CommandTarget.named("lead"), origin=AgentOrigin()) assert HeartbeatSilent.violations(held_bus, "lead") == () held_bus.threads._rename_thread("lead", "renamed-lead") @@ -698,7 +698,7 @@ def test_first_heartbeat_is_owed_from_the_oldest_open_request_then_survives_rena 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()) + ), 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") 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/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) From 23e9e5bcfc888627daae4b6ed80cadf117a2649f Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 01:09:25 -0400 Subject: [PATCH 09/10] Preserve converter offset ownership through policy retirement --- plans/commander-native-plan.md | 16 +++++++---- src/agent_comms/bus_tables.py | 16 ----------- tests/test_carry_bus_history.py | 32 +++++++++++++++++++++ tests/test_carry_reply_policy.py | 49 +++----------------------------- 4 files changed, 46 insertions(+), 67 deletions(-) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 58c3822462..04aca07c64 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -298,12 +298,16 @@ 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. -`BusSchema` now compares the complete catalog and rejects unknown tables without -deleting them. Private coordinator composition remains its original mandatory -view: optional awareness has a separate owner and absence/drift contract. -Configured optional awareness is covered by a real coordinator check. The -current installed pair remains independent of this draft; installation and -live commander qualification belong to integration acceptance. +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 diff --git a/src/agent_comms/bus_tables.py b/src/agent_comms/bus_tables.py index bd873a01b0..7eea3052ae 100644 --- a/src/agent_comms/bus_tables.py +++ b/src/agent_comms/bus_tables.py @@ -32,9 +32,6 @@ ForeignKey, Index, SharedFileTables, - SQLiteSchemaObject, - SQLiteTableName, - TableSchema, TypedRow, TypedTable, sql_literal, @@ -64,19 +61,6 @@ 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.""" - @classmethod - def stored_schema(cls, db: sqlite3.Connection) -> list[SQLiteSchemaObject]: - """This complete file sees objects its partial family views cannot name.""" - return TableSchema.stored_schema(db) - - @classmethod - def stored_tables(cls, db: sqlite3.Connection) -> list[SQLiteTableName]: - return TableSchema.stored_tables(db) - - @classmethod - def carry_schema(cls, db: sqlite3.Connection) -> None: - super().carry_schema(db) - cls.require_schema(db) @dataclass(frozen=True) class ParticipantSource(TypedTable, BusProjectionTables, declared_name=None): 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 index 3467e48225..3d17aa4b51 100644 --- a/tests/test_carry_reply_policy.py +++ b/tests/test_carry_reply_policy.py @@ -1,54 +1,22 @@ -"""One-time cache retirement preserves source rows and refuses unknown ownership.""" +"""One-time cache retirement preserves unrelated state and rejects unrecognized input.""" -import sqlite3 from pathlib import Path import pytest from agent_comms.bus import Bus, PrivateRecord from agent_comms.bus_tables import BusSchema -from agent_comms.cohort_schema import AwarenessTable -from agent_comms.coordinator import Coordination 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 agent_comms.private_runtime_schema import PrivateCoordinatorSchema, PrivateCoordinatorTable -from agent_comms.typed_table import SQLiteTableName, WriterOwnedFile 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_complete_file_reconcile_preserves_and_rejects_unclaimed_tables(): - schema = BusSchema - with sqlite3.connect(":memory:") as db: - schema.create_schema(db) - assert not schema.schema_differs(db) - db.execute("CREATE TABLE unknown_owner (payload TEXT)") - db.execute("INSERT INTO unknown_owner VALUES ('retain this input')") - db.commit() - assert schema.schema_differs(db) - with pytest.raises(schema.schema_error): - schema.reconcile_schema(db, WriterOwnedFile()) - assert db.execute("SELECT payload FROM unknown_owner").fetchall() == [("retain this input",)] - - -def test_mandatory_private_view_preserves_optional_awareness_ownership(tmp_path): - with Coordination(str(tmp_path / "private.sqlite3")) as store: - PrivateCoordinatorTable.install(store) - db = store.session._connection - assert AwarenessTable.stored_schema(db) - PrivateCoordinatorSchema.require_schema(db) - optional, = AwarenessTable.schema_rows() - with store.session.writing() as writer: - SQLiteTableName(optional.declared_name).drop(writer) - assert AwarenessTable.schema_differs(db) - PrivateCoordinatorSchema.require_schema(db) - - 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), @@ -66,24 +34,15 @@ def test_retired_reply_policy_carry_is_dry_idempotent_and_source_preserving(bus, assert StoredDirectReply.count(db) == 1 with pytest.raises(MaintenanceNotHeldError): step.carry(apply=True) - with maintenance_held(bus.path.parent): - with pytest.raises(RelationViolationError): - step.carry(apply=True) - with bus.observing() as db: - assert StoredDirectReply.present(db) - value, = db.execute("SELECT payload FROM unknown_owner").fetchone() - assert value == "unclassified" - with pytest.raises(RelationViolationError): - BusSchema.require_schema(db) - assert bus.delivery(reply.seq) == reply - with bus.writing() as db: - SQLiteTableName("unknown_owner").drop(db) 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): From 3f6ae92e21d3f7d5e57b0625d63f830daa9c63c7 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Fri, 23 Oct 2026 01:14:33 -0400 Subject: [PATCH 10/10] Record the actual cached-query baseline and direct scan regression --- plans/commander-native-plan.md | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 04aca07c64..b9a47e1ac9 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -281,7 +281,15 @@ 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. -Neither replay establishes constant cost. Ordinary participant pages retain +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