| """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}") |
|
|
