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..a8a7532ef7 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,10 @@ class Sign: key: tuple[object, ...] line: str = field(compare=False) + @property + def mark(self) -> str: + return self.signal.mark(self.key) + class Degradation(DeclaredFamily, affix="Degradation"): """One signal that the fleet is getting away from its commander.""" @@ -85,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 @@ -168,11 +193,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 +273,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 +287,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 +390,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)