mirror of
https://github.com/AletheiaVox/signal_bridge_remote.git
synced 2026-10-07 03:18:17 +08:00
63d7176ee6
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>
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""
|
||
Signal Bridge Remote — Session Intensity Governor
|
||
|
||
Tracks cumulative session intensity ("heat") per user and enforces
|
||
cooldown periods when the threshold is reached.
|
||
|
||
Heat model:
|
||
- Accumulates at: current_intensity × HEAT_RATE per second
|
||
- Dissipates at: COOL_RATE per second when intensity = 0
|
||
- Partial dissipation when running at low intensity:
|
||
net_rate = (intensity × HEAT_RATE) - COOL_RATE
|
||
- Cooldown triggers at COOLDOWN_THRESHOLD (default 90%)
|
||
- Cooldown exits when heat falls to COOLDOWN_EXIT (default 30%)
|
||
AND at least COOLDOWN_DURATION seconds have passed
|
||
|
||
The governor state is piggybacked on heartbeat pings so the phone
|
||
can display heat level and cooldown countdown in real time.
|
||
|
||
This is a soft safety layer — the hard safety is the dead man's switch.
|
||
The governor exists to pace sessions, not to prevent hardware damage.
|
||
"""
|
||
from __future__ import annotations
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
|
||
from . import config
|
||
|
||
log = logging.getLogger("signal_bridge.governor")
|
||
|
||
|
||
@dataclass
|
||
class GovernorConfig:
|
||
"""Per-user governor config (overrides server defaults)."""
|
||
heat_rate: float = config.GOVERNOR_HEAT_RATE
|
||
cool_rate: float = config.GOVERNOR_COOL_RATE
|
||
cooldown_threshold: float = config.GOVERNOR_COOLDOWN_THRESHOLD
|
||
cooldown_exit: float = config.GOVERNOR_COOLDOWN_EXIT
|
||
cooldown_duration: float = config.GOVERNOR_COOLDOWN_DURATION
|
||
enabled: bool = config.GOVERNOR_ENABLED
|
||
|
||
|
||
@dataclass
|
||
class GovernorState:
|
||
"""Per-user heat tracking state."""
|
||
heat: float = 0.0 # 0..100
|
||
current_intensity: float = 0.0 # last known intensity (0..1)
|
||
intensity_expires_at: float = 0.0 # 0 = runs until an explicit stop
|
||
in_cooldown: bool = False
|
||
cooldown_entered_at: float = 0.0
|
||
cooldown_count: int = 0 # total cooldowns this session
|
||
last_tick: float = field(default_factory=time.time)
|
||
cfg: GovernorConfig = field(default_factory=GovernorConfig)
|
||
|
||
def tick(self) -> None:
|
||
"""Update heat based on elapsed time since last tick."""
|
||
now = time.time()
|
||
dt = now - self.last_tick
|
||
self.last_tick = now
|
||
|
||
if not self.cfg.enabled:
|
||
# Disabled means the whole subsystem is off, not just enforcement.
|
||
# Clear anything left over from before the toggle so no stale heat
|
||
# or cooldown is reported to the AI or acted on by the phone.
|
||
if self.heat or self.in_cooldown or self.current_intensity:
|
||
self.heat = 0.0
|
||
self.in_cooldown = False
|
||
self.current_intensity = 0.0
|
||
self.intensity_expires_at = 0.0
|
||
return
|
||
|
||
if dt <= 0 or dt > 10:
|
||
# Sanity: skip huge jumps (e.g., system clock change)
|
||
return
|
||
|
||
# A command with a declared duration stops on its own; the phone never
|
||
# reports that, so expire it here. Without this, current_intensity is
|
||
# sticky forever and heat integrates against hardware sitting idle.
|
||
if self.intensity_expires_at and now >= self.intensity_expires_at:
|
||
self.current_intensity = 0.0
|
||
self.intensity_expires_at = 0.0
|
||
|
||
if self.in_cooldown:
|
||
# During cooldown: always dissipate, intensity is forced to 0
|
||
self.heat -= self.cfg.cool_rate * dt
|
||
self.heat = max(0.0, self.heat)
|
||
|
||
# Check cooldown exit conditions
|
||
elapsed = now - self.cooldown_entered_at
|
||
if (self.heat <= self.cfg.cooldown_exit
|
||
and elapsed >= self.cfg.cooldown_duration):
|
||
self.in_cooldown = False
|
||
log.info(
|
||
f"Cooldown ended: heat={self.heat:.1f}% "
|
||
f"after {elapsed:.0f}s"
|
||
)
|
||
else:
|
||
# Normal operation: accumulate or dissipate
|
||
net = (self.current_intensity * self.cfg.heat_rate
|
||
- self.cfg.cool_rate)
|
||
# Only dissipate if there's actual heat to lose
|
||
if net < 0 and self.heat <= 0:
|
||
return
|
||
self.heat += net * dt
|
||
self.heat = max(0.0, min(100.0, self.heat))
|
||
|
||
# Check cooldown trigger
|
||
if self.heat >= self.cfg.cooldown_threshold:
|
||
self.in_cooldown = True
|
||
self.cooldown_entered_at = now
|
||
self.cooldown_count += 1
|
||
self.current_intensity = 0.0
|
||
self.intensity_expires_at = 0.0
|
||
log.warning(
|
||
f"Cooldown triggered: heat={self.heat:.1f}% "
|
||
f"(cooldown #{self.cooldown_count})"
|
||
)
|
||
|
||
def record_command(self, intensity: float, duration: float = 0.0) -> None:
|
||
"""
|
||
Record that a command was sent at a given intensity.
|
||
|
||
`duration` is how long the command runs before the phone stops it on
|
||
its own. 0 means it runs until an explicit stop, which is the only
|
||
case where the intensity should stay set indefinitely.
|
||
"""
|
||
self.current_intensity = max(0.0, min(1.0, intensity))
|
||
self.intensity_expires_at = (
|
||
time.time() + duration if duration > 0 else 0.0
|
||
)
|
||
|
||
def record_stop(self) -> None:
|
||
"""Record that devices were stopped."""
|
||
self.current_intensity = 0.0
|
||
self.intensity_expires_at = 0.0
|
||
|
||
@property
|
||
def cooldown_remaining(self) -> int:
|
||
"""Seconds remaining in cooldown (0 if not in cooldown)."""
|
||
if not self.in_cooldown:
|
||
return 0
|
||
elapsed = time.time() - self.cooldown_entered_at
|
||
# Time-based minimum
|
||
time_remaining = max(0, self.cfg.cooldown_duration - elapsed)
|
||
# Heat-based: estimate time to reach exit threshold
|
||
if self.cfg.cool_rate > 0:
|
||
heat_remaining = max(
|
||
0,
|
||
(self.heat - self.cfg.cooldown_exit)
|
||
/ self.cfg.cool_rate,
|
||
)
|
||
else:
|
||
heat_remaining = 0
|
||
return int(max(time_remaining, heat_remaining))
|
||
|
||
@property
|
||
def predicted_seconds(self) -> int | None:
|
||
"""
|
||
At current intensity, how many seconds until cooldown triggers?
|
||
Returns None if intensity is 0 or heat is dissipating.
|
||
"""
|
||
if self.in_cooldown or self.current_intensity <= 0:
|
||
return None
|
||
net = (self.current_intensity * self.cfg.heat_rate
|
||
- self.cfg.cool_rate)
|
||
if net <= 0:
|
||
return None # heat is stable or dissipating
|
||
remaining_heat = self.cfg.cooldown_threshold - self.heat
|
||
if remaining_heat <= 0:
|
||
return 0
|
||
return int(remaining_heat / net)
|
||
|
||
def to_dict(self) -> dict:
|
||
"""Serialize for piggybacking on heartbeat pings."""
|
||
if not self.cfg.enabled:
|
||
# Report nothing rather than stale numbers. A disabled governor
|
||
# must never put a cooldown on the wire — the phone drives its own
|
||
# ACTIVE→COOLDOWN state machine off in_cooldown, and the AI reads
|
||
# the list_devices footer as if it meant something.
|
||
return {
|
||
"enabled": False,
|
||
"heat_pct": 0.0,
|
||
"in_cooldown": False,
|
||
"cooldown_remaining": 0,
|
||
"cooldown_count": self.cooldown_count,
|
||
"predicted_seconds": None,
|
||
}
|
||
return {
|
||
"enabled": True,
|
||
"heat_pct": round(self.heat, 1),
|
||
"in_cooldown": self.in_cooldown,
|
||
"cooldown_remaining": self.cooldown_remaining,
|
||
"cooldown_count": self.cooldown_count,
|
||
"predicted_seconds": self.predicted_seconds,
|
||
}
|
||
|
||
|
||
class Governor:
|
||
"""
|
||
Central governor managing per-user heat state.
|
||
|
||
Usage:
|
||
- Call tick(user_id) on every heartbeat to update heat
|
||
- Call check(user_id) before sending commands — returns (allowed, reason)
|
||
- Call record_command(user_id, intensity) after sending a command
|
||
- Call record_stop(user_id) when devices are stopped
|
||
- Call get_state(user_id) to get current state for heartbeat piggyback
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._states: dict[str, GovernorState] = {}
|
||
|
||
def _get(self, user_id: str) -> GovernorState:
|
||
if user_id not in self._states:
|
||
self._states[user_id] = GovernorState()
|
||
return self._states[user_id]
|
||
|
||
def tick(self, user_id: str) -> None:
|
||
"""Advance the heat model. Call on every heartbeat."""
|
||
self._get(user_id).tick()
|
||
|
||
def check(self, user_id: str) -> tuple[bool, str]:
|
||
"""
|
||
Check if a command is allowed.
|
||
Returns (allowed: bool, reason: str).
|
||
"""
|
||
state = self._get(user_id)
|
||
if not state.cfg.enabled:
|
||
return True, ""
|
||
|
||
if state.in_cooldown:
|
||
remaining = state.cooldown_remaining
|
||
return False, (
|
||
f"Cooldown active ({remaining}s remaining). "
|
||
f"Session heat reached {state.cfg.cooldown_threshold:.0f}%. "
|
||
f"Wait for cooldown to complete before sending more commands."
|
||
)
|
||
return True, ""
|
||
|
||
def apply_user_config(self, user_id: str, effective: dict) -> None:
|
||
"""Apply per-user config overrides from the database."""
|
||
state = self._get(user_id)
|
||
state.cfg = GovernorConfig(
|
||
enabled=bool(effective.get("governor_enabled", True)),
|
||
heat_rate=float(effective.get("heat_rate", config.GOVERNOR_HEAT_RATE)),
|
||
cool_rate=float(effective.get("cool_rate", config.GOVERNOR_COOL_RATE)),
|
||
cooldown_threshold=float(effective.get("cooldown_threshold", config.GOVERNOR_COOLDOWN_THRESHOLD)),
|
||
cooldown_exit=float(effective.get("cooldown_exit", config.GOVERNOR_COOLDOWN_EXIT)),
|
||
cooldown_duration=float(effective.get("cooldown_duration", config.GOVERNOR_COOLDOWN_DURATION)),
|
||
)
|
||
log.info(f"Applied user config for {user_id}: heat_rate={state.cfg.heat_rate}, "
|
||
f"cool_rate={state.cfg.cool_rate}, threshold={state.cfg.cooldown_threshold}")
|
||
|
||
def record_command(
|
||
self, user_id: str, intensity: float, duration: float = 0.0
|
||
) -> None:
|
||
"""Record that a command was dispatched."""
|
||
self._get(user_id).record_command(intensity, duration)
|
||
|
||
def record_stop(self, user_id: str) -> None:
|
||
"""Record that devices were stopped."""
|
||
self._get(user_id).record_stop()
|
||
|
||
def get_state(self, user_id: str) -> dict:
|
||
"""Get governor state dict for heartbeat piggyback."""
|
||
return self._get(user_id).to_dict()
|
||
|
||
def remove_user(self, user_id: str) -> None:
|
||
"""Clean up when a user disconnects."""
|
||
self._states.pop(user_id, None)
|
||
|
||
|
||
# Singleton
|
||
governor = Governor()
|