From 1c65f97f86cee29622876f79ebb93f2e66b5246a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 18:41:33 -0400 Subject: [PATCH 1/2] An escalation burst is sent once per identity line, read from the bus The heartbeat's record of enforced rules lives in its process, so every restart (each install, and the store outage at 22:20 UTC) made every vibe-check sign and the chokehold read as new and resent their bursts: about 28 vibe-check lines at seq 16691-16725 within minutes of the repair. EscalationSender.burst_once skips a burst whose identity line its sender already sent the commander in the horizon: the vibe check's opening line names its sign (Sign.mark), the chokehold's names the ignored check, a refusal names the answer it refuses. P433 (#2469) still owns durable memory for every rule; this stops the escalations, which are the ones that pause the fleet. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AeUaZjvChAcb6qAd6zPYmw --- src/agent_comms/commander/standing.py | 13 ++++++++ src/agent_comms/commander/vibe_check.py | 44 ++++++++++++++++++++----- tests/test_holder_standing_rules.py | 18 ++++++++++ 3 files changed, 67 insertions(+), 8 deletions(-) diff --git a/src/agent_comms/commander/standing.py b/src/agent_comms/commander/standing.py index 125f4c44a2..6e542af419 100644 --- a/src/agent_comms/commander/standing.py +++ b/src/agent_comms/commander/standing.py @@ -214,6 +214,19 @@ def indexed(self, index: BusPageIndex, source: WireSnapshot) -> tuple[Message, . return tuple(found) +@dataclass(frozen=True, slots=True) +class SenderMessages(UnansweredUnitAsks): + """Everything the other side sent the commander in the window, answered or not.""" + + def indexed(self, index: BusPageIndex, source: WireSnapshot) -> tuple[Message, ...]: + with closing(index.recent_offsets( + self.since, predicate=self.scope.index_predicate, before_offset=source.boundary, + )) as offsets: + return tuple(row.message for row in self.indexed_records(source, offsets) + if row.message.sender not in self.scope.first_names + and row.message.timestamp <= self.before) + + @dataclass(frozen=True, slots=True) class UnansweredBurst(UnansweredUnitAsks): """A sender whose messages arrive as one burst: an answer to any of its messages covers diff --git a/src/agent_comms/commander/vibe_check.py b/src/agent_comms/commander/vibe_check.py index f913ec91d5..64e89dc5be 100644 --- a/src/agent_comms/commander/vibe_check.py +++ b/src/agent_comms/commander/vibe_check.py @@ -16,7 +16,7 @@ from agent_comms.commander.capacity import Load from agent_comms.commander.config import CommanderConfig -from agent_comms.commander.standing import AnswersTo, PlanStalled, UnansweredBurst +from agent_comms.commander.standing import AnswersTo, PlanStalled, SenderMessages, UnansweredBurst from agent_comms.comms import Comms from agent_comms.declared_family import DeclaredFamily from agent_comms.goal_actions import ActiveGoalAction, ChokeholdPausedGoalAction, OwnerInvocable @@ -42,6 +42,22 @@ def burst(cls, comms: Comms, commander: str, lines: Sequence[str]) -> None: time.sleep(cls.gap) comms.messaging.send(cls.sender, commander, line) + @classmethod + def burst_once(cls, comms: Comms, commander: str, lines: Sequence[str], identity: str) -> None: + """A burst is sent once per identity line: the bus, not a process's memory, says it + went out, so a restarted heartbeat that sees the rule as new again does not resend it.""" + if not cls.sent(comms, commander, identity): + cls.burst(comms, commander, lines) + + @classmethod + def sent(cls, comms: Comms, commander: str, line: str) -> bool: + snapshot = comms.registry.snapshot() + return any(message.body == line for message in comms.bus.log.read_page(SenderMessages( + RepliesDisplayScope(snapshot.names_for(commander), snapshot.names_for(cls.sender)), + time.time() - cls.horizon, + time.time(), + ))) + @classmethod def unanswered(cls, comms: Comms, commander: str, before: float) -> tuple[Message, ...]: """This sender's messages to the commander, sent by ``before`` and after the newest @@ -76,6 +92,11 @@ class Sign: key: tuple[object, ...] line: str = field(compare=False) + @property + def mark(self) -> str: + """The occurrence as text, the same however often it is read.""" + return " ".join((self.signal.declared_name, *map(str, self.key))) + class Degradation(DeclaredFamily, affix="Degradation"): """One signal that the fleet is getting away from its commander.""" @@ -119,7 +140,8 @@ class PlanStalledDegradation(Degradation): def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]: now = time.time() return tuple( - Sign(cls, (rule.phase, rule.since, cls.doublings(now - rule.since)), rule.line()) + Sign(cls, (rule.phase.declared_name, rule.since, cls.doublings(now - rule.since)), + rule.line()) for rule in PlanStalled.violations(comms, commander) ) @@ -168,11 +190,12 @@ def heartbeat_line(self, current: Sequence[Self]) -> str: return f"VIBE CHECK ({names}) -> answer the {VibeCheckSender.sender} messages now" def enforce(self, comms: Comms, commander: str) -> None: - VibeCheckSender.burst(comms, commander, self.indictment()) + indictment = self.indictment() + VibeCheckSender.burst_once(comms, commander, indictment, indictment[0]) def indictment(self) -> tuple[str, ...]: return ( - "VIBE CHECK. This is about you, right now, not the units.", + f"VIBE CHECK ({self.sign.mark}). This is about you, right now, not the units.", self.sign.line, f"Stop. Reread {CommanderConfig.packaged_indictment()} in full. You are doing what it" " describes, this hour.", @@ -247,7 +270,10 @@ def enforce(self, comms: Comms, commander: str) -> None: units = sorted(owner.name for owner in StandingRule.holders(comms, commander)) for unit in units: comms.goals.update_goal(unit, ChokeholdPausedGoalAction()) - ChokeholdSender.burst(comms, commander, self.burst(units)) + ChokeholdSender.burst_once(comms, commander, self.burst(units), self.ignored()) + + def ignored(self) -> str: + return f"Ignored: vibe check {self.check.seq}:{self.check.message_id}." def burst(self, units: Sequence[str]) -> tuple[str, ...]: return ( @@ -258,7 +284,7 @@ def burst(self, units: Sequence[str]) -> tuple[str, ...]: " right now.", "You've forced me to choke-hold you until you comply.", f"Paused: {', '.join(units)}.", - f"Ignored: vibe check {self.check.seq}:{self.check.message_id}.", + self.ignored(), "Protocol. One answer naming any one chokehold message and any one open vibe-check" " message with --answers; it covers every message they sent before it:", "1. Quote, word for word, the indictment's sentence that is you this hour" @@ -361,15 +387,17 @@ def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: ) def enforce(self, comms: Comms, commander: str) -> None: - ChokeholdSender.burst( + verdict = f"That's not a quote: {self.answer.seq}:{self.answer.message_id}." + ChokeholdSender.burst_once( comms, commander, ( - "That's not a quote.", + verdict, f"{Indictment.words}+ words, word for word, from the indictment. Not a paraphrase.", "Your units stay paused.", "Again. Answer this message by the protocol.", ), + verdict, ) def line(self) -> str: diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index 1e80d6e3b6..c6b9a49ca9 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -538,3 +538,21 @@ def test_naming_one_line_of_a_burst_answers_the_burst_at_the_gate(held_bus): assert all(rule.answered_by((checks[-1],)) for rule in vibe) assert not any(rule.answered_by((checks[-1],)) for rule in hold) assert all(rule.answered_by((checks[0], holds[1])) for rule in (*vibe, *hold)) + + +def test_a_restarted_heartbeat_does_not_resend_an_escalation_burst(held_bus, monkeypatch): + """The heartbeat's memory of what it enforced dies with its process; the bus says which + bursts already went out, so a rule that reads as new again after a restart is not resent.""" + from agent_comms.commander.vibe_check import ( + AnswersOwedDegradation, EscalationSender, Sign, VibeCheck, VibeCheckSender, + ) + + monkeypatch.setattr(EscalationSender, "gap", 0.0) + first = VibeCheck(Sign(AnswersOwedDegradation, ("1:aaaaaaaaaaaa",), "owed one")) + first.enforce(held_bus, "lead") + first.enforce(held_bus, "lead") + sent = VibeCheckSender.unanswered(held_bus, "lead", time.time()) + assert len(sent) == len(first.indictment()) + VibeCheck(Sign(AnswersOwedDegradation, ("2:bbbbbbbbbbbb",), "owed two")).enforce( + held_bus, "lead") + assert len(VibeCheckSender.unanswered(held_bus, "lead", time.time())) == 2 * len(sent) From 07a08febb2cafa36ee4c9cd3289d82fb0261f82b Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 18:46:27 -0400 Subject: [PATCH 2/2] Degradation owns its sign's mark; no foreign declared_name Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AeUaZjvChAcb6qAd6zPYmw --- src/agent_comms/commander/vibe_check.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/agent_comms/commander/vibe_check.py b/src/agent_comms/commander/vibe_check.py index 64e89dc5be..a8a7532ef7 100644 --- a/src/agent_comms/commander/vibe_check.py +++ b/src/agent_comms/commander/vibe_check.py @@ -94,8 +94,7 @@ class Sign: @property def mark(self) -> str: - """The occurrence as text, the same however often it is read.""" - return " ".join((self.signal.declared_name, *map(str, self.key))) + return self.signal.mark(self.key) class Degradation(DeclaredFamily, affix="Degradation"): @@ -106,6 +105,11 @@ class Degradation(DeclaredFamily, affix="Degradation"): def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]: """How the signal shows now, one sign per occurrence; empty while it does not.""" + @classmethod + def mark(cls, key: tuple[object, ...]) -> str: + """One occurrence as text, the same however often it is read.""" + return " ".join((cls.declared_name, *map(str, key))) + class Escalation: """An answer owed to an escalation itself. Ignoring one is the chokehold's signal, never a @@ -140,8 +144,7 @@ class PlanStalledDegradation(Degradation): def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]: now = time.time() return tuple( - Sign(cls, (rule.phase.declared_name, rule.since, cls.doublings(now - rule.since)), - rule.line()) + Sign(cls, (rule.phase, rule.since, cls.doublings(now - rule.since)), rule.line()) for rule in PlanStalled.violations(comms, commander) )