An escalation burst is sent once per identity line, read from the bus

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

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:

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)