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)