From e6c2e1452305faaa4ba08aa77985981346a199e7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 17:50:31 -0400 Subject: [PATCH] The vibe check fires once per degraded sign, not every round A stall of hours cannot clear inside one 30-minute round, so the check re-gated every verb and re-sent the burst for the same five phases each round (the commander, 16286). Each sign now breaks its own VibeCheck rule, and the heartbeat enforces a rule only when it newly breaks: an owed message once, a stalled phase when it crosses the window and at each age doubling, overload once per round. Answers owed to the escalations themselves are the chokehold's signal and no longer feed a new vibe check. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AeUaZjvChAcb6qAd6zPYmw --- src/agent_comms/commander/vibe_check.py | 98 ++++++++++++++++--------- tests/test_vibe_check_signs.py | 26 +++++++ 2 files changed, 90 insertions(+), 34 deletions(-) create mode 100644 tests/test_vibe_check_signs.py diff --git a/src/agent_comms/commander/vibe_check.py b/src/agent_comms/commander/vibe_check.py index 9c87c12079..4a79bac9bf 100644 --- a/src/agent_comms/commander/vibe_check.py +++ b/src/agent_comms/commander/vibe_check.py @@ -7,6 +7,7 @@ from __future__ import annotations +import math import time from abc import abstractmethod from collections.abc import Sequence @@ -64,76 +65,106 @@ class ChokeholdSender(EscalationSender): sender = "chokehold" +@dataclass(frozen=True) +class Sign: + """One occurrence of a degraded signal. Its key is what makes it new: a message, a stalled + phase at its age doubling, a load round. The same occurrence never stops the commander twice; + its line is how it reads now.""" + + signal: type[Degradation] + key: tuple[object, ...] + line: str = field(compare=False) + + class Degradation(DeclaredFamily, affix="Degradation"): """One signal that the fleet is getting away from its commander.""" @classmethod @abstractmethod - def found(cls, comms: Comms, commander: str) -> tuple[str, ...]: - """How the signal shows now, one line each; empty while it does not.""" + def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]: + """How the signal shows now, one sign per occurrence; empty while it does not.""" + + +class Escalation: + """An answer owed to an escalation itself. Ignoring one is the chokehold's signal, never a + new vibe check, or the check would feed on its own messages.""" class AnswersOwedDegradation(Degradation): - """Messages to the commander are waiting past their window.""" + """Messages to the commander are waiting past their window; each message is one sign.""" @classmethod - def found(cls, comms: Comms, commander: str) -> tuple[str, ...]: - return tuple(rule.line() for rule in OwedAnswer.broken(comms, commander)) + def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]: + return tuple( + Sign(cls, (rule.cited,), rule.line()) + for rule in OwedAnswer.broken(comms, commander) + if not isinstance(rule, Escalation) + ) class PlanStalledDegradation(Degradation): - """Held plan work that main still lacks.""" + """Held plan work that main still lacks. A stall of hours does not clear in one round, so a + phase stops the commander when it crosses the window and again each time its age doubles.""" @classmethod - def found(cls, comms: Comms, commander: str) -> tuple[str, ...]: - return tuple(rule.line() for rule in PlanStalled.violations(comms, commander)) + 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()) + for rule in PlanStalled.violations(comms, commander) + ) + + @staticmethod + def doublings(held: float) -> int: + return int(math.log2(max(held, PlanStalled.window) / PlanStalled.window)) class OverloadDegradation(Degradation): - """More owed orders, expired timers and unjudged edits than one commander keeps up with.""" + """More owed orders, expired timers and unjudged edits than one commander keeps up with; + load moves every beat, so it stops the commander once per round while it stays over.""" ceiling: ClassVar[float] = 100.0 + round_seconds: ClassVar[int] = 30 * 60 @classmethod - def found(cls, comms: Comms, commander: str) -> tuple[str, ...]: + def found(cls, comms: Comms, commander: str) -> tuple[Sign, ...]: load = Load.of(comms, commander) - return (f"{load}, over {cls.ceiling:.0f}",) if load.total >= cls.ceiling else () + if load.total < cls.ceiling: + return () + return (Sign(cls, (int(time.time() // cls.round_seconds),), + f"{load}, over {cls.ceiling:.0f}"),) @dataclass(frozen=True) class VibeCheck(UndatedStandingRule): - """Degraded signals break it once per combination and per round, so a commander still - degraded half an hour later is stopped again.""" + """Each degraded sign breaks it once: the heartbeat enforces a rule when it newly breaks, so a + sign that persists stops the commander once, and a new message, a newly stalled phase or a + doubled stall stops it again. Pings are decisions, each once.""" - signals: frozenset[type[Degradation]] - round: int - lines: tuple[str, ...] = field(compare=False) - round_seconds: ClassVar[int] = 30 * 60 + sign: Sign @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: - found = { - member: member.found(comms, commander) + return tuple( + cls(sign) for member in Degradation.members_with(Degradation) - } - showing = {member: lines for member, lines in found.items() if lines} - if not showing: - return () - return ( - cls( - frozenset(showing), - int(time.time() // cls.round_seconds), - tuple(line for lines in showing.values() for line in lines), - ), + for sign in member.found(comms, commander) ) + def heartbeat_group(self) -> object: + return VibeCheck + + def heartbeat_line(self, current: Sequence[Self]) -> str: + names = ", ".join(sorted({check.sign.signal.declared_name for check in current})) + 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()) def indictment(self) -> tuple[str, ...]: return ( "VIBE CHECK. This is about you, right now, not the units.", - *self.lines, + self.sign.line, f"Stop. Reread {CommanderConfig.packaged_indictment()} in full. You are doing what it" " describes, this hour.", "Answer every vibe-check message with one --answers: which failure in the indictment" @@ -142,12 +173,11 @@ def indictment(self) -> tuple[str, ...]: ) def line(self) -> str: - names = ", ".join(sorted(signal.declared_name for signal in self.signals)) - return f"VIBE CHECK ({names}) -> answer the {VibeCheckSender.sender} message now" + return self.heartbeat_line((self,)) @dataclass(frozen=True) -class VibeCheckUnanswered(OwedAnswer, UndatedStandingRule): +class VibeCheckUnanswered(Escalation, OwedAnswer, UndatedStandingRule): """A vibe check is owed an answer like a directive: the gate refuses until it gets one.""" seconds: ClassVar[int] = 5 * 60 @@ -236,7 +266,7 @@ def line(self) -> str: @dataclass(frozen=True) -class ChokeholdUnanswered(OwedAnswer, UndatedStandingRule): +class ChokeholdUnanswered(Escalation, OwedAnswer, UndatedStandingRule): """Every chokehold message gates the commander from the moment it lands.""" @classmethod diff --git a/tests/test_vibe_check_signs.py b/tests/test_vibe_check_signs.py new file mode 100644 index 0000000000..e561230f05 --- /dev/null +++ b/tests/test_vibe_check_signs.py @@ -0,0 +1,26 @@ +"""A stalled phase stops the commander when it crosses the window and at each doubling.""" + +import pytest + +from agent_comms.commander.standing import PlanStalled +from agent_comms.commander.vibe_check import ( + ChokeholdUnanswered, + Escalation, + PlanStalledDegradation, + VibeCheckUnanswered, +) + +HOUR = 3600 + + +@pytest.mark.parametrize("held, doublings", [ + (2 * HOUR, 0), (3.9 * HOUR, 0), (4 * HOUR, 1), (7.9 * HOUR, 1), (8 * HOUR, 2), (17 * HOUR, 3), +]) +def test_a_stall_is_new_again_only_when_its_age_doubles(held, doublings): + assert PlanStalled.window == 2 * HOUR + assert PlanStalledDegradation.doublings(held) == doublings + + +def test_answers_owed_to_escalations_never_feed_a_new_vibe_check(): + assert issubclass(VibeCheckUnanswered, Escalation) + assert issubclass(ChokeholdUnanswered, Escalation)