Spaces:
Sleeping
Sleeping
File size: 3,112 Bytes
fbe9dad ebd50f7 fbe9dad ebd50f7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """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
@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)
|