Notebook: the operator layer, exercised in a cell

tiny_computer_agent_app.py

@@ -21,13 +21,18 @@ with app.setup:
     # The project's public API, imported once so no cell has to duplicate it.
     from tiny_computer_agent import (
         AgentLoop,
+        AuditLog,
         DryRunExecutor,
         Element,
+        FakeScreen,
         GrabScreen,
         MockBrain,
         PlannerBrain,
         Policy,
         PyAutoGuiExecutor,
+        Watchdog,
+        gate_from_spec,
+        verify_audit,
     )
     from tiny_computer_agent.actions import Action
     from tiny_computer_agent.brain import DecisionMenu
@@ -1585,6 +1590,79 @@ def demo():
     return
 
 
+@app.cell
+def operator_layer():
+    # The 2.0 operator layer, exercised rather than described: a gate that consents per action,
+    # a watchdog that refuses to wait, and an audit log that can be checked afterwards.
+    import json as _json
+    import tempfile as _tempfile
+
+    _goal = "click 'Save' to save the document"
+
+
+    def _room():
+        """A two-button screen, so the demo is about consent and evidence, not about OCR."""
+        return FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),
+                            Element("Cancel", x=440, y=120, width=80, height=24))])
+
+
+    class _Hung(DryRunExecutor):
+        """An executor that never returns - the failure a step budget cannot catch."""
+
+        def execute(self, action, observation):
+            import time
+
+            time.sleep(3)
+            return super().execute(action, observation)
+
+
+    _audit_path = Path(_tempfile.mkdtemp()) / "run.jsonl"
+    _audit = AuditLog(_audit_path)
+    _watchdog = Watchdog(step_timeout=30, wall_clock=120)
+
+    # 1. an ordinary episode, with a gate, a deadline and a record around it
+    _episode = AgentLoop(
+        screen=_room(), brain=PlannerBrain(), executor=DryRunExecutor(),
+        policy=Policy(step_budget=3), gate=gate_from_spec("labels:Save"),
+        audit=_audit, watchdog=_watchdog,
+    ).run(_goal)
+
+    # 2. the same goal behind a read-only gate: the operator's no is a step, not a crash
+    _refused = AgentLoop(screen=_room(), brain=PlannerBrain(), executor=DryRunExecutor(),
+                         policy=Policy(step_budget=3),
+                         gate=gate_from_spec("read-only")).run(_goal)
+
+    # 3. a call that hangs: without the watchdog this cell would never finish
+    _hung = AgentLoop(screen=_room(), brain=MockBrain([{"kind": "click", "label": "Save"}]),
+                      executor=_Hung(), watchdog=Watchdog(step_timeout=0.05)).run(_goal)
+
+    # 4. the record: check it, then tamper with a copy and let the verifier name the entry
+    _lines = _audit_path.read_text().splitlines()
+    _entry = _json.loads(_lines[0])
+    _entry["outcome"] = "nothing happened at all"           # a nicer story than the truth
+    _tampered = _audit_path.with_name("tampered.jsonl")
+    _tampered.write_text(_json.dumps(_entry, sort_keys=True, separators=(",", ":")) + "\n"
+                         + "\n".join(_lines[1:]) + "\n")
+
+    operator_rows = [
+        {"2.0 piece": "gate: labels:Save", "what happened": "let the whitelisted click through"},
+        {"2.0 piece": "gate: read-only", "what happened": f"refused - {_refused.steps[-1].error}"},
+        {"2.0 piece": "watchdog", "what happened": _hung.steps[-1].error},
+        {"2.0 piece": "audit: intact", "what happened": verify_audit(_audit_path)[1]},
+        {"2.0 piece": "audit: edited", "what happened": verify_audit(_tampered)[1]},
+    ]
+
+    operator_layer = mo.vstack([
+        mo.md(f"**The 2.0 operator layer.** One goal - `{_goal}` - three ways: consented, refused, "
+              f"and hung. {_watchdog.render()}"),
+        mo.ui.table(operator_rows, selection=None),
+        mo.md("```\n" + _episode.render() + "\n```"),
+    ])
+    operator_layer
+
+    return
+
+
 @app.cell
 def scoreboard():
     # The brains are separable, so they can be scored against the same tasks with no