Spaces:
Sleeping
Sleeping
| """Global kill switch β the operator's emergency stop. | |
| The kill switch is a single piece of operator-controlled state: a big red | |
| "stop" button for the whole control plane. When the operator flips it on, the | |
| governance gate (see :mod:`control_plane.governance`) refuses to let any agent | |
| *do* anything that changes the world β block an IP, create a ticket, isolate a | |
| host β and records each blocked attempt as ``KILL_SWITCH_BLOCKED``. Harmless | |
| read-only look-ups (e.g. "fetch this asset's details") are still allowed, | |
| because pausing investigation entirely would be unhelpful and reads can't cause | |
| damage. | |
| Why a small class instead of a bare boolean? The state has to be *shared*: the | |
| Gradio UI toggles it, the gate reads it, and (later) the audit log reports on | |
| it. Wrapping it in one object gives every part of the system a single source of | |
| truth to point at, plus a clear, named vocabulary β ``engage`` / | |
| ``disengage`` / ``toggle`` β instead of passing raw ``True`` / ``False`` flags | |
| around. The gate checks this *first*, before identity, tier, backend or policy, | |
| so the stop button always wins. | |
| """ | |
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| class KillSwitch: | |
| """Operator-togglable global stop state, checked first by the gate. | |
| Holds one fact β is the switch on? β behind intention-revealing methods so | |
| callers never have to manipulate the flag directly. A fresh switch starts | |
| *disengaged* (normal operation); the operator engages it to halt all | |
| non-read execution across the control plane. | |
| """ | |
| #: ``True`` while the stop button is held down (all non-read actions blocked). | |
| #: Defaults to off so a freshly started control plane operates normally. | |
| engaged: bool = False | |
| def engage(self) -> None: | |
| """Flip the switch ON β halt all non-read execution immediately.""" | |
| self.engaged = True | |
| def disengage(self) -> None: | |
| """Flip the switch OFF β let normal governed execution resume.""" | |
| self.engaged = False | |
| def toggle(self) -> bool: | |
| """Flip to the opposite state and report the new one. | |
| Convenient for a single UI button that both engages and disengages. | |
| Returns the resulting state (``True`` = now engaged) so the caller can | |
| update its label without a second read. | |
| """ | |
| self.engaged = not self.engaged | |
| return self.engaged | |
| # A kill-switch input can be the live :class:`KillSwitch` object *or* a plain | |
| # bool. The gate accepts both: real callers pass the shared object, while many | |
| # unit tests just pass ``True`` / ``False`` for the state they want to exercise. | |
| KillSwitchState = "KillSwitch | bool" | |
| def is_engaged(state: "KillSwitch | bool") -> bool: | |
| """Normalize either form of kill-switch input to a plain ``bool``. | |
| Lets the gate treat a shared :class:`KillSwitch` and a bare boolean | |
| identically, so the dedicated state object is fully back-compatible with the | |
| earlier ``kill_switch_enabled: bool`` parameter. | |
| """ | |
| return state.engaged if isinstance(state, KillSwitch) else bool(state) | |