fix(governor): disabled means disabled, and timed commands stop accruing heat

enabled gated only Governor.check(). The heat model, the heartbeat
piggyback and the list_devices footer all ran unconditionally, so a
governor that was switched off could still report heat, still print
COOLDOWN, and still drive the phone's ACTIVE->COOLDOWN transition while
blocking nothing. tick() now returns early when disabled and clears
carried-over state; to_dict() reports enabled:false with zeroed values;
the list_devices footer is omitted.

Separately, current_intensity was only ever cleared by record_stop(),
whose three callers are an explicit stop, a phone emergency stop and the
dead man's switch. A pattern ending on its own duration told the server
nothing, so heat integrated at the last commanded intensity against idle
hardware forever. Commands now pass their duration through to
record_command() and the intensity expires when it lapses. duration:0
still means run-until-stop. For escalate the expiry is duration +
hold_seconds, and only when hold_seconds > 0, per the hold contract.

Adds tests/verify_governor.py: 22 offline checks over the disabled path,
timed-command expiry, and the enabled-path invariants. Existing
verify_server.py still passes 23/23.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Aletheia
2026-07-27 20:16:24 +02:00
parent b8a52e9658
commit 63d7176ee6
4 changed files with 290 additions and 13 deletions

170
tests/verify_governor.py Normal file
View File

@@ -0,0 +1,170 @@
"""Verification of the session intensity governor's heat model.
Covers the three behaviours fixed in "disabled means disabled":
1. `enabled=False` stops the whole model, not just enforcement.
2. A disabled governor reports nothing - no heat, no cooldown, on the
heartbeat wire or in the list_devices footer.
3. A command with a declared duration expires instead of accumulating
heat forever against hardware that already stopped.
Plus the enabled-path invariants those fixes must not have broken.
Run from anywhere: python tests/verify_governor.py
Pure logic - no server, no database, no network, no hardware.
"""
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT))
from server.governor import GovernorConfig, GovernorState # noqa: E402
PASS = []
FAIL = []
def check(name, cond, detail=""):
(PASS if cond else FAIL).append(name)
print((" ok " if cond else " FAIL")
+ f" {name}" + (f" - {detail}" if detail and not cond else ""))
def advance(state: GovernorState, seconds: float, step: float = 1.0):
"""Run the model forward without sleeping.
The governor reads the wall clock, so simulate elapsed time by rewinding
every stored timestamp rather than by waiting. All three must move
together or the model sees an inconsistent clock: last_tick drives heat
integration, cooldown_entered_at drives the cooldown exit, and
intensity_expires_at drives the auto-stop.
Steps stay under the 10s sanity cap in tick().
"""
remaining = seconds
while remaining > 0:
dt = min(step, remaining)
state.last_tick -= dt
if state.cooldown_entered_at:
state.cooldown_entered_at -= dt
if state.intensity_expires_at:
state.intensity_expires_at -= dt
state.tick()
remaining -= dt
def new_state(enabled=True, **cfg):
return GovernorState(cfg=GovernorConfig(enabled=enabled, **cfg))
print("-- disabled means disabled " + "-" * 36)
# Heat must not accumulate at all while the governor is off.
s = new_state(enabled=False)
s.record_command(0.7)
advance(s, 60)
check("disabled: no heat accumulates", s.heat == 0.0, f"heat={s.heat}")
check("disabled: never enters cooldown", s.in_cooldown is False)
# State left over from before the toggle is cleared, not frozen in place.
s = new_state(enabled=True)
s.record_command(0.7)
advance(s, 30)
carried = s.heat
s.cfg = GovernorConfig(enabled=False)
advance(s, 1)
check("disabled: pre-existing heat is cleared",
carried > 0 and s.heat == 0.0, f"was {carried:.1f}, now {s.heat}")
# A disabled governor must put nothing on the wire. The phone drives its own
# ACTIVE->COOLDOWN transition off in_cooldown, and the AI reads the footer.
s = new_state(enabled=False)
s.heat = 95.0
s.in_cooldown = True
d = s.to_dict()
check("disabled: reports enabled=False", d["enabled"] is False)
check("disabled: reports zero heat", d["heat_pct"] == 0.0, str(d))
check("disabled: never reports cooldown", d["in_cooldown"] is False, str(d))
check("disabled: no predicted countdown", d["predicted_seconds"] is None)
# check() has always been correct; confirm it still is.
s = new_state(enabled=False)
s.in_cooldown = True
check("disabled: commands are never blocked", s.cfg.enabled is False)
print("\n-- timed commands expire " + "-" * 38)
# The bug: current_intensity was only ever cleared by an explicit stop, so a
# pattern that ended on its own duration kept integrating heat forever.
# At 0.7 the net rate is (0.7 * 3.0) - 2.0 = +0.1 heat/sec, so a 60s command
# builds ~6% heat and then must give it all back.
s = new_state()
s.record_command(0.7, duration=60)
advance(s, 59)
heat_while_running = s.heat
advance(s, 1)
check("timed command stops heating after its duration",
s.current_intensity == 0.0, f"intensity={s.current_intensity}")
check("heat was actually accumulating while it ran",
heat_while_running > 5.0, f"heat={heat_while_running:.1f}")
advance(s, 300)
check("heat dissipates to zero after expiry rather than climbing",
s.heat == 0.0, f"{heat_while_running:.1f} -> {s.heat:.1f}")
# Duration 0 means "runs until an explicit stop" - must stay sticky.
s = new_state()
s.record_command(0.7, duration=0)
advance(s, 60)
check("duration=0 keeps running (no false expiry)",
s.current_intensity == 0.7, f"intensity={s.current_intensity}")
check("duration=0 accumulates heat", s.heat > 0, f"heat={s.heat}")
# The regression this whole thing was found by: one finished escalate at 0.70
# reading as live heat ten minutes later.
s = new_state()
s.record_command(0.70, duration=15)
advance(s, 600)
check("the Friday specimen: no phantom heat 10 min after a 15s ramp",
s.heat == 0.0 and s.to_dict()["predicted_seconds"] is None,
f"heat={s.heat}, predicted={s.to_dict()['predicted_seconds']}")
# An explicit stop still clears everything, expiry included.
s = new_state()
s.record_command(0.8, duration=300)
s.record_stop()
check("explicit stop clears intensity", s.current_intensity == 0.0)
check("explicit stop clears the pending expiry", s.intensity_expires_at == 0.0)
print("\n-- enabled path still works " + "-" * 34)
# Sustained high intensity must still reach cooldown and refuse commands.
# At 1.0 the net rate is +1.0 heat/sec, so 90% arrives at ~90 seconds.
s = new_state()
s.record_command(1.0, duration=0)
advance(s, 92)
check("sustained intensity still triggers cooldown", s.in_cooldown is True,
f"heat={s.heat:.1f}")
check("cooldown reports a countdown", s.cooldown_remaining > 0,
f"remaining={s.cooldown_remaining}")
check("enabled: reports enabled=True", s.to_dict()["enabled"] is True)
# And cooldown must still end: heat back under the exit threshold (30%) AND
# the minimum duration (30s) elapsed. Heat dissipates at 2.0/sec, so heat is
# the binding constraint here, not the clock.
advance(s, 20)
check("cooldown holds while heat is still above the exit threshold",
s.in_cooldown is True, f"heat={s.heat:.1f}")
advance(s, 400)
check("cooldown ends once heat and time both clear",
s.in_cooldown is False, f"heat={s.heat:.1f}")
# Idle heat dissipates to zero and stops there.
s = new_state()
s.heat = 40.0
advance(s, 600)
check("idle heat dissipates to zero", s.heat == 0.0, f"heat={s.heat}")
print(f"\n{len(PASS)} passed, {len(FAIL)} failed")
sys.exit(1 if FAIL else 0)