File size: 7,437 Bytes
e2f048d 42ee5cf | 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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """The operator's yes, as a first-class part of the loop.
A :class:`~tiny_computer_agent.safety.Policy` answers "may the harness do this at all?".
A gate answers the other question: "does the person watching want it done *now*?" Both end
in the same place - a refusal that is recorded as a step and stops the episode, never a
traceback and never a silent click.
loop = AgentLoop(screen=..., brain=..., gate=InteractiveGate(), policy=Policy(dry_run=False))
loop = AgentLoop(..., gate=AllowOnlyLabels(["Save", "Cancel"])) # click nothing else
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol, Sequence
from .actions import Action
from .observe import Observation, normalise_label
READ_ONLY_KINDS = ("wait", "done")
"""Verbs that cannot change the machine, and so never need an operator's consent."""
def mutating(action: Action) -> bool:
"""True when this action would change something on the machine."""
return action.kind not in READ_ONLY_KINDS
@dataclass(frozen=True)
class Approval:
"""A gate's answer, with the reason it will be recorded under."""
approved: bool
reason: str = ""
def line(self) -> str:
return f"{'approved' if self.approved else 'refused'}: {self.reason}"
class Gate(Protocol):
"""Ask before acting. Return an :class:`Approval`; never raise to say no."""
def approve(self, action: Action, observation: Observation | None = None) -> Approval:
...
@dataclass
class AllowAll:
"""No gate: everything policy already allowed is approved. The default when none is given."""
reason: str = "no gate configured"
def approve(self, action: Action, observation: Observation | None = None) -> Approval:
return Approval(True, self.reason)
@dataclass
class DenyAll:
"""Refuse every action that would change anything - a one-line kill switch for a live loop."""
reason: str = "the operator said no"
def approve(self, action: Action, observation: Observation | None = None) -> Approval:
if not mutating(action):
return Approval(True, "read-only")
return Approval(False, self.reason)
@dataclass
class AllowOnlyLabels:
"""A whitelist of labels: the loop may click these and nothing else.
This is the gate for "watch it work, but it may only touch Save and Cancel" - the
coordinates still come from the index, so a renamed button is refused rather than
clicked at.
"""
labels: Sequence[str] = ()
reason: str = "not in the operator's whitelist"
def approve(self, action: Action, observation: Observation | None = None) -> Approval:
if not mutating(action):
return Approval(True, "read-only")
if action.label and self.matches(action.label):
return Approval(True, f"{action.label!r} is whitelisted")
return Approval(False, self.reason)
def matches(self, label: str) -> bool:
target = normalise_label(label)
for allowed in self.labels:
wanted = normalise_label(allowed)
if target and (target == wanted or wanted in target or target in wanted):
return True
return False
@dataclass
class InteractiveGate:
"""Ask a person, once per action, and remember what they said.
``always`` and ``never`` persist for the rest of the run, so a supervised session does not
turn into a keyboard exercise. ``reader``/``writer`` are injectable, so the gate is testable
and a console can route the question through its own widgets.
"""
prompt: str = "run this action?"
reader: Callable[[str], str] = input
writer: Callable[[str], None] = print
_always: bool = field(default=False, init=False, repr=False)
_never: bool = field(default=False, init=False, repr=False)
asked: int = field(default=0, init=False)
def approve(self, action: Action, observation: Observation | None = None) -> Approval:
if not mutating(action):
return Approval(True, "read-only")
if self._never:
return Approval(False, "the operator said never")
if self._always:
return Approval(True, "the operator said always")
self.asked += 1
question = f"{self.prompt} {action.describe()} [y]es / [n]o / [a]lways / [v]never: "
try:
answer = (self.reader(question) or "").strip().lower()
except (EOFError, KeyboardInterrupt):
return Approval(False, "the operator was not there to answer")
if answer.startswith("a"):
self._always = True
return Approval(True, "the operator said always")
if answer.startswith("v"):
self._never = True
return Approval(False, "the operator said never")
if answer.startswith("y"):
return Approval(True, "the operator said yes")
return Approval(False, "the operator said no")
@dataclass
class AllOf:
"""Every gate must approve - the strictest reading of a set of rules."""
gates: Sequence[Gate] = ()
def approve(self, action: Action, observation: Observation | None = None) -> Approval:
for gate in self.gates:
approval = gate.approve(action, observation)
if not approval.approved:
return approval
return Approval(True, "every gate approved")
@dataclass
class AnyOf:
"""Any gate may approve - the loosest, for "my colleague or my rulebook"."""
gates: Sequence[Gate] = ()
reason: str = "no gate approved"
def approve(self, action: Action, observation: Observation | None = None) -> Approval:
for gate in self.gates:
approval = gate.approve(action, observation)
if approval.approved:
return approval
return Approval(False, self.reason)
GATE_SPECS = ("allow-all", "read-only", "interactive", "labels:")
"""The spellings :func:`gate_from_spec` accepts, in the order its error lists them."""
def gate_from_spec(spec: str) -> Gate:
"""Build a gate from a one-line spelling, the one the command line takes.
A gate is the operator's yes, and an operator should not have to write Python to give
one::
allow-all everything policy already allowed
read-only nothing that changes the machine, whatever policy says
interactive ask me, once per action, and remember "always"/"never"
labels:Save,Cancel only these labels, and nothing else
An unknown spec raises ``ValueError`` naming the accepted spellings, because a typo in
a safety switch must never quietly mean "allow everything".
"""
wanted = (spec or "").strip().lower()
if wanted in ("allow-all", "allow", "none"):
return AllowAll()
if wanted in ("read-only", "readonly"):
return DenyAll(reason="read-only gate: nothing may change the machine")
if wanted in ("interactive", "ask"):
return InteractiveGate()
if wanted.startswith("labels:"):
labels = [part.strip() for part in spec.split(":", 1)[1].split(",") if part.strip()]
if not labels:
raise ValueError("labels: needs at least one label, e.g. labels:Save,Cancel")
return AllowOnlyLabels(tuple(labels))
accepted = " | ".join(GATE_SPECS)
raise ValueError(f"unknown gate {spec!r}: expected one of {accepted}")
|