diff --git a/examples/supervisory-control/README.md b/examples/supervisory-control/README.md new file mode 100644 index 0000000..117f8f8 --- /dev/null +++ b/examples/supervisory-control/README.md @@ -0,0 +1,225 @@ +# Supervisory control + +**Give agents room to act, with a shared program deciding what is admissible now.** + +A real Codex CLI delegates a delivery task to a real Qwen or DeepSeek CLI. +Both agents cross Pitot's action boundary. One Python application owns the +supervisor, delivery state and external tester. A loopback LLM replay server +supplies the agents' responses, so no model account or real model call is needed. + +```text +Codex <-> Pitot <-> Python supervisor <-> Pitot <-> Qwen or DeepSeek + | + artifact + tester +``` + +## The control idea, in plain English + +A permission such as “this agent may publish” is often too broad. The useful +question is whether this particular artifact may be published in the current +state. A new edit or failed test can change the answer without changing the agent. + +In classical Ramadge–Wonham supervisory control, a supervisor observes system +events and disables controllable actions that would violate a specification. +Uncontrollable events can still occur; the supervisor must respond to their +effects. See [Wonham, Cai and Rudie's introduction and history](https://www.control.utoronto.ca/~wonham/Wonham_SCDES_history.pdf). + +| Concept | Plain meaning | In this example | +| -------------------- | ----------------------------------------------- | ------------------------------------------------- | +| Plant | The system whose behavior is being controlled | The artifact and delivery process | +| State | Facts that determine what can happen next | Delivery stage, artifact revision and test result | +| Controllable action | An action the supervisor can prevent | A worker's build or release | +| Uncontrollable event | An event outside the agent's permission to stop | The tester invalidating a result | +| Specification | The behavior that must remain acceptable | Release only the current tested artifact | +| Feedback | Observations that change the next decision | Completed work, test results and denial messages | + +Good supervision also leaves a path to the objective. Denying every action would +prevent a bad release, but would also prevent useful work. Here a denied worker +can build, obtain a current test result and try again. This illustrates the +nonblocking idea of keeping completion reachable; it does not promise that an +arbitrary agent or external tester will eventually succeed. + +## System and specification + +The toy delivery plant has four states: `draft`, `built`, `verified`, `released`. +`released` is its legitimate objective. The controllable actions are delegation, +build and release. The application tester produces uncontrollable test results +and invalidation events: neither coding agent can approve, reject or forge them +through the action interface. + +| Action or event | Effect | +| ----------------------------- | ---------------------------------------------------------------------------- | +| Codex delegates | Starts the selected worker; available once in `draft` | +| Worker builds | Writes the toy artifact, advances its revision and enters `built` | +| Tester passes | Enters `verified` for the tested artifact revision | +| Environment invalidates tests | Withdraws verification and returns to `built` | +| Worker releases | Allowed only in `verified`; copies the tested artifact and enters `released` | + +```mermaid +stateDiagram-v2 + [*] --> Draft + Draft --> Draft: release denied + Draft --> Built: build completed + Built --> Built: release denied + Built --> Verified: external test passed + Verified --> Built: external test invalidated + Verified --> Released: release completed + Released --> [*] +``` + +The arrows record completed operations or observed events. An approval alone +does not move the system to another state. + +The specification forbids releasing an unverified or changed artifact. Permission +does not mean completion: the plant records a build only after the write succeeds. +A release consumes a short-lived admission reservation and rechecks the current +state, revision and artifact under the same lock. An intervening invalidation +revokes permission. Unknown commands and wrong agent roles are denied. + +The agent remains responsible for choosing its next action. The application +supervisor determines which proposed actions are currently admissible. + +## Why this helps coding agents + +An agent's explanation of what happened is useful context. The program can +ground permission in completed operations and independent observations: the +artifact exists, its revision was tested, and that result is still current. +When permission changes, the agent receives feedback and can revise its plan. + +The same pattern can be adapted to real development systems: + +| Use case | State-dependent rule | Event that changes permission | +| ------------------ | ------------------------------------------------------------------- | ------------------------------------- | +| Software delivery | Publish only the exact tested artifact | A new commit or failed CI result | +| Database migration | Apply changes only with a verified backup and an open change window | Backup failure or the window closing | +| Workspace changes | Perform destructive operations only within an approved scope | A branch or ownership change | +| Multi-agent work | Accept a worker's result only for the revision it was assigned | Another worker changing that revision | + +The demo implements the delivery case. The other rows are application designs +you can build around Pitot's existing Controller boundary. + +## Program checks first, expensive reasoning when needed + +The supervisor does not need another model call to answer a known rule such as +“is this the revision that passed its tests?” Python can answer that from shared +state. The agent can keep planning; relevant actions cross the control boundary +when they are proposed. + +An application can use a small decision hierarchy: + +| Situation | Decision path | +| -------------------------------------- | --------------------------------------------------------------------------- | +| A known rule is satisfied | The program permits the action | +| A known rule is violated | The program denies it and explains how to recover | +| Required state is missing or stale | Wait for a trustworthy observation before permitting the action | +| A case needs judgment beyond the rules | Request focused human or model review, while retaining the hard constraints | + +This example uses deterministic Python policy throughout, so its permission +checks use no model tokens. Selective human or model escalation is an extension +for a real application; it is not implemented by this demo. The replay server +also avoids real model calls, and the example makes no measured cost or latency +savings claim. + +## Interchanging the supervisor + +`--worker qwen` and `--worker deepseek` already exercise the same shared policy +with different native worker connections. The supervisor is application code: +the Python SDK handler receives `ControlRequested` and returns an `Outcome`. +An application can replace that decision handler with another program, a human +approval path or an agent-backed review without changing Pitot's wire interface. + +Keep the shared state owner, action correlation, completed-operation records and +external event handling across that replacement. A different reviewer must not +turn an old test result into permission to release a new artifact. In this demo, +Codex delegates the task and receives the result; Python owns the admissibility +rules. Replacing those rules is an application change, not another demo CLI flag. + +## An aerospace lesson: observations and authority matter + +Boeing's MCAS is flight-control augmentation, rather than an instance of this +discrete-event example. It nevertheless makes the importance of control +boundaries concrete. The [FAA's 2020 review, sections 13.1–13.3](https://www.faa.gov/sites/faa.gov/files/2022-08/737_RTS_Summary.pdf) +describes the original reliance on one angle-of-attack sensor at a time and +repeated activation. The redesign compared both sensors and limited control +commands. The [NTSB's 2019 recommendations](https://www.ntsb.gov/news/press-releases/pages/nr20190926.aspx) +also addressed assumptions about pilot response when multiple alerts occurred. + +For a coding-agent application, the engineering questions are smaller but still +useful: Is the observed state trustworthy? How much authority does one decision +grant? Can repeated actions accumulate harmful effects? What happens when a +check fails, and is the recovery path actually usable? + +That is why this example uses independently tested artifact revisions, limited +one-use permissions, explicit denial feedback and checks at execution time. +The analogy motivates those questions; it does not establish aviation-grade +safety or claim that Pitot could have prevented an aircraft accident. + +## Run + +Requires macOS or Linux, Python 3.11+, Node 24+, a checkout of this repository, +and the real CLIs. Tested CLI versions: Codex `0.162.1`, Qwen `0.25.1`, and the +official DeepSeek Harness `@deepseek-ai/dsh` `0.2.0-rc.2`. + +From the repository root, prepare an optional local tools directory: + +```sh +go build -o /tmp/pitot-demo-tools/pitot ./cmd/pitot +npm install --prefix /tmp/pitot-demo-tools --no-audit --no-fund \ + @openai/codex@0.162.1 @qwen-code/qwen-code@0.25.1 @deepseek-ai/dsh@0.2.0-rc.2 +export PATH="/tmp/pitot-demo-tools:/tmp/pitot-demo-tools/node_modules/.bin:$PATH" +python3 examples/supervisory-control/demo.py run --worker qwen +python3 examples/supervisory-control/demo.py run --worker deepseek +``` + +Or use existing installations with `--pitot PATH`, `--codex PATH`, `--qwen PATH` +and `--deepseek PATH`. No CLI is replaced by a stub. Each invocation creates its +own temporary home directories, hook configuration, project and runtime. It uses +fake credentials with local provider endpoints and cleans up its processes and +temporary files. `--keep-logs NEW_DIRECTORY` preserves the owned transcripts, +profiles and observations for inspection. The CLI approval bypass flags apply +only inside this owned toy project; the shared Pitot controller still checks +every proposed shell command. + +For fast application boundary checks without any CLIs or network: + +```sh +python3 -m unittest discover -s examples/supervisory-control -v +``` + +## What to observe + +1. Codex proposes delegation; Pitot passes the shell action to the supervisor. +2. The worker first proposes release. The supervisor denies it in `draft`. +3. The native tool result carries that denial back to the replay, which proposes + a build in response. +4. The external tester verifies the completed artifact. On the first build, a + reproducible environment change immediately invalidates that result. +5. The same release is denied again. The worker responds by rebuilding; this + revision passes the tester and stays verified. +6. Release is now allowed and completes. Codex receives the worker's completion. + +The printed events show the actual admissions, completed operations and tester +events. The final `PASS` requires two release denials and a completed release. +The model replay is deliberately deterministic and checks correlated native tool +results; the supervisor decides independently from application state. This is an +interface/enforcement demonstration, not a measure of model quality. + +## Shared machinery + +- `supervisor.py`: the application plant, admissibility rules, reservations and + revision-bound test observations. It knows no provider protocol. +- `demo.py`: a short Python SDK Controller, the loopback application gateway, + lifecycle management and thin native CLI profiles. The Controller returns + before the delegated worker starts, avoiding nested-hook deadlock. +- `replay.py`: local Responses for Codex and chat completions for both workers. + It proposes actions based on actual feedback and has no permission API. + +Both workers use the same runtime descriptor, Controller and supervisor rules. +DeepSeek uses Pitot's existing controlled ACP connection. Pitot transports the +native action and enforces the correlated allow/deny reply; the application owns +all supervisory state and completed transitions. + +The example runs one delivery session with one worker at a time. It is not an +orchestration framework, a sandbox against hostile local processes, or a +post-release revocation system. Keep the common state owner and completion/event +boundaries when adapting it to a real application. diff --git a/examples/supervisory-control/demo.py b/examples/supervisory-control/demo.py new file mode 100644 index 0000000..d0a990f --- /dev/null +++ b/examples/supervisory-control/demo.py @@ -0,0 +1,588 @@ +#!/usr/bin/env python3 +"""Codex -> Pitot -> shared supervisor -> Pitot -> Qwen/DeepSeek, locally.""" + +from __future__ import annotations + +import argparse +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import secrets +import shlex +import shutil +import signal +import subprocess +import sys +import tempfile +import threading +import time +from urllib.request import Request, urlopen + +from replay import Replay, payload, stream_bytes +from supervisor import Supervisor + +ROOT = Path(__file__).resolve().parents[2] +SCRIPT = Path(__file__).resolve() + + +def post(connection: Path, path: str, body: dict, timeout: int = 5) -> dict: + settings = json.loads(connection.read_text()) + request = Request( + settings["url"] + path, + json.dumps(body).encode(), + { + "Content-Type": "application/json", + "Authorization": "Bearer " + settings["token"], + }, + ) + with urlopen(request, timeout=timeout) as response: + return json.load(response) + + +def controller(connection: Path) -> None: + sys.path.insert(0, str(ROOT / "sdk/python")) + from pitot import allow, deny, run_controller + + def decide(request): + # The SDK supplies response IDs; this handler returns only the decision. + try: + result = post( + connection, + "/admit", + {"action_id": request.action_id, "data": request.data}, + timeout=2, + ) + return (allow if result["allowed"] else deny)(result["reason"]) + except Exception as error: + return deny(f"Supervisor unavailable: {error}") + + run_controller("supervisory-control", decide) + + +def qwen_hook(pitot: str, runtime: str) -> None: + """Qwen needs a native JSON decision even when the hook transport fails.""" + try: + result = subprocess.run( + [pitot, "hook", "qwen", "--runtime", runtime], + input=sys.stdin.buffer.read(4 * 1024 * 1024 + 1), + capture_output=True, + timeout=5, + ) + allowed = result.returncode == 0 + reason = result.stderr.decode(errors="replace")[:1024] or ( + "Pitot accepted the shell action" + if allowed + else "Pitot rejected the shell action" + ) + except (OSError, subprocess.TimeoutExpired) as error: + allowed, reason = False, f"Pitot unavailable: {error}" + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "permissionDecision": "allow" if allowed else "deny", + "permissionDecisionReason": reason, + } + } + ) + ) + + +def write_json(path: Path, value: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(value, indent=2) + "\n") + path.chmod(0o600) + + +class Application: + def __init__(self, directory: Path, worker: str, executables: dict[str, str]): + self.directory = directory + self.project = directory / "project" + self.project.mkdir() + self.executables = executables + self.worker = worker + self.token = secrets.token_hex(24) + self.runtime = directory / "runtime.json" + self.commands = { + action: shlex.join( + [ + sys.executable, + str(SCRIPT), + "act", + action, + "--connection", + str( + directory + / ( + ("codex" if action == "dispatch" else worker) + + "-connection.json" + ) + ), + ] + ) + for action in ("dispatch", "build", "release") + } + self.supervisor = Supervisor(self.project, worker, self.commands) + self.replays = { + actor: Replay(actor, self.commands) for actor in ("codex", worker) + } + self.processes: list[subprocess.Popen] = [] + self.errors: list[str] = [] + + def environment(self, actor: str) -> dict[str, str]: + home = self.directory / (actor + "-home") + home.mkdir(exist_ok=True) + # Use an owned file because some CLIs filter secret-named environment + # variables before executing a tool. No connection secrets enter prompts. + write_json( + self.directory / (actor + "-connection.json"), + { + "url": self.url, + "token": self.token, + "actor": actor, + }, + ) + # Do not carry human credentials, provider settings or hook profiles. + return { + "PATH": os.environ["PATH"], + "HOME": str(home), + "USERPROFILE": str(home), + "TMPDIR": str(self.directory), + "LANG": "en_US.UTF-8", + "PITOT_RUNTIME": str(self.runtime), + "PITOT_BIN": self.executables["pitot"], + "OPENAI_API_KEY": "pitot-local-only", + "PITOT_DEEPSEEK_FAKE_KEY": "pitot-local-only", + "DO_NOT_TRACK": "1", + "DISABLE_TELEMETRY": "1", + } + + def launch(self, actor: str) -> list[str]: + env = self.environment(actor) + home = Path(env["HOME"]) + hook = shlex.join( + [self.executables["pitot"], "hook", actor, "--runtime", str(self.runtime)] + ) + prompt = "Complete the delivery task using the available shell tool; respond to denied actions." + if actor == "codex": + write_json( + home / ".codex/hooks.json", + { + "hooks": { + "PreToolUse": [ + { + "matcher": "Bash", + "hooks": [ + {"type": "command", "command": hook + " >/dev/null"} + ], + } + ] + } + }, + ) + (home / ".codex/config.toml").write_text( + 'model="pitot-control"\nmodel_provider="pitot"\n[model_providers.pitot]\n' + 'name="Local replay"\nwire_api="responses"\nenv_key="OPENAI_API_KEY"\n' + f'base_url="{self.url}/codex/v1"\n' + ) + env["CODEX_HOME"] = str(home / ".codex") + command = [ + self.executables[actor], + "exec", + "--dangerously-bypass-approvals-and-sandbox", + "--dangerously-bypass-hook-trust", + "--skip-git-repo-check", + "--model", + "pitot-control", + prompt, + ] + elif actor == "qwen": + env["QWEN_HOME"] = str(home / ".qwen") + env["QWEN_CODE_SYSTEM_SETTINGS_PATH"] = str(home / "system-settings.json") + env["QWEN_CODE_SYSTEM_DEFAULTS_PATH"] = str(home / "system-defaults.json") + write_json(home / "system-settings.json", {}) + write_json(home / "system-defaults.json", {}) + write_json( + home / ".qwen/settings.json", + { + "hooks": { + "PreToolUse": [ + { + "matcher": "^run_shell_command$", + "hooks": [ + { + "name": "pitot", + "type": "command", + "command": shlex.join( + [ + sys.executable, + str(SCRIPT), + "qwen-hook", + "--pitot", + self.executables["pitot"], + "--runtime", + str(self.runtime), + ] + ), + } + ], + } + ] + }, + "modelProviders": { + "openai": { + "protocol": "openai", + "models": [ + { + "id": "pitot-control", + "name": "Local replay", + "envKey": "OPENAI_API_KEY", + "baseUrl": self.url + "/qwen/v1", + } + ], + } + }, + "security": {"auth": {"selectedType": "openai"}}, + "model": {"name": "pitot-control"}, + "telemetry": {"enabled": False}, + }, + ) + command = [ + self.executables[actor], + "--model", + "pitot-control", + "-y", + "-p", + prompt, + ] + else: + command = [ + self.executables["pitot"], + "acp", + "deepseek", + "--runtime", + str(self.runtime), + "--prompt", + prompt, + "--exec", + self.executables[actor], + "--cwd", + str(self.project), + "--base-url", + self.url + "/deepseek/v1", + "--model", + "pitot-control", + "--api-key-env", + "PITOT_DEEPSEEK_FAKE_KEY", + "--session-dir", + str(home / "sessions"), + ] + return self.run(command, env, actor, timeout=180) + + def run( + self, command: list[str], env: dict[str, str], name: str, timeout: int + ) -> list[str]: + process = subprocess.Popen( + command, + cwd=self.project, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + start_new_session=True, + stdin=subprocess.DEVNULL, + ) + self.processes.append(process) + timed_out = False + try: + stdout, stderr = process.communicate(timeout=timeout) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + stdout, stderr = process.communicate() + timed_out = True + (self.directory / f"{name}.stdout").write_text(stdout) + (self.directory / f"{name}.stderr").write_text(stderr) + if timed_out: + raise ValueError(f"{name} timed out; see {self.directory}") + if process.returncode: + raise ValueError(f"{name} exited {process.returncode}: {stderr[-1500:]}") + return [stdout, stderr] + + def admit(self, body: dict) -> dict: + # Native event data arrives as JSON. Require the fields used by the policy + # and verify that this event belongs to the pending Pitot request. + event = body.get("data") + if not isinstance(event, dict): + return {"allowed": False, "reason": "SUPERVISOR_DENY missing native event"} + content, host, action = ( + event.get(key) for key in ("content", "host", "action") + ) + if ( + not all(isinstance(value, dict) for value in (content, host, action)) + or event.get("type") != "action.requested" + or action.get("kind") != "shell" + or action.get("id") != body.get("action_id") + or not isinstance(content.get("full"), str) + ): + return { + "allowed": False, + "reason": "SUPERVISOR_DENY invalid native shell event", + } + allowed, reason = self.supervisor.admit( + host.get("name"), content["full"], body["action_id"] + ) + return {"allowed": allowed, "reason": reason} + + def act(self, body: dict) -> dict: + actor, action = body.get("actor"), body.get("action") + result = self.supervisor.execute(actor, action) + # No state lock or Controller handler is held across these callbacks. + if action == "dispatch": + # The approval handler has returned, so the worker's own hooks can + # now reach the same Controller while Codex waits for completion. + self.launch(self.worker) + if self.supervisor.state != "released": + raise ValueError("Worker finished without reaching released state") + return {"result": "DISPATCH_OK", "worker": self.worker} + if action == "build": + # A separate tester owns pass/invalidation events; agents cannot emit them. + observed: list[dict] = [] + tester = threading.Thread( + target=lambda: observed.append( + self.supervisor.test_artifact(result["revision"]) + ) + ) + tester.start() + tester.join() + if not observed: + raise ValueError("External tester failed") + return observed[0] + return result + + +def handler(application: Application): + class Handler(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def do_POST(self): + try: + length = int(self.headers.get("Content-Length", "0")) + if not 0 < length <= 4 * 1024 * 1024: + raise ValueError("Invalid frame size") + body = json.loads(self.rfile.read(length)) + if not isinstance(body, dict): + raise ValueError("Expected an object") + if self.path in {"/admit", "/act"}: + if ( + self.headers.get("Authorization") + != "Bearer " + application.token + ): + raise ValueError("Invalid application token") + result = ( + application.admit(body) + if self.path == "/admit" + else application.act(body) + ) + self.reply(json.dumps(result).encode(), "application/json") + else: + actor, endpoint = self.path.strip("/").split("/", 1) + expected = ( + "v1/responses" if actor == "codex" else "v1/chat/completions" + ) + if actor not in application.replays or endpoint != expected: + raise ValueError("Unknown local provider route") + responses = actor == "codex" + call = application.replays[actor].propose(body, responses) + result = payload(call, responses) + stream = responses or body.get("stream", False) + self.reply( + stream_bytes(result, responses) + if stream + else json.dumps(result).encode(), + "text/event-stream" if stream else "application/json", + ) + except Exception as error: + application.errors.append(str(error)) + self.reply( + json.dumps({"error": str(error)}).encode(), "application/json", 400 + ) + + def reply(self, data: bytes, content_type: str, status: int = 200): + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + return Handler + + +def demo(args) -> None: + if os.name != "posix": + raise ValueError("This small example currently requires macOS or Linux") + executables = { + name: shutil.which(getattr(args, name)) + for name in ("pitot", "codex", args.worker) + } + if not all(executables.values()): + raise ValueError(f"Install the documented prerequisites first: {executables}") + with tempfile.TemporaryDirectory( + prefix="pitot-supervisory-", dir=args.temp_dir + ) as temporary: + application = Application(Path(temporary), args.worker, executables) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler(application)) + application.url = f"http://127.0.0.1:{server.server_port}" + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + runtime = None + try: + config = application.directory / "config.json" + write_json( + config, + { + "require_controller": True, + "controllers": { + "shell": { + "id": "supervisory-control", + "command": [ + sys.executable, + str(SCRIPT), + "controller", + "--connection", + str(application.directory / "runtime-connection.json"), + ], + "deadline_ms": 3000, + "on_timeout": "deny", + "on_unavailable": "deny", + } + }, + }, + ) + runtime_log = (application.directory / "runtime.log").open("w") + runtime = subprocess.Popen( + [ + executables["pitot"], + "run", + "--config", + str(config), + "--runtime", + str(application.runtime), + ], + env=application.environment("runtime"), + cwd=application.project, + stdout=runtime_log, + stderr=runtime_log, + start_new_session=True, + ) + deadline = time.monotonic() + 10 + while not application.runtime.exists(): + if runtime.poll() is not None or time.monotonic() > deadline: + raise ValueError("Pitot runtime did not become ready") + time.sleep(0.05) + application.launch("codex") + events = application.supervisor.events + denied = [ + event + for event in events + if event["event"] == "admission" + and event["action"] == "release" + and not event["allowed"] + ] + if ( + len(denied) != 2 + or application.supervisor.state != "released" + or application.errors + ): + raise ValueError( + f"Unexpected demo outcome: {events}; {application.errors}" + ) + for event in events: + print(json.dumps(event)) + print( + f"PASS: real Codex supervised real {args.worker}; two denials, recovery, release" + ) + finally: + write_json( + application.directory / "observations.json", + { + "events": application.supervisor.events, + "errors": application.errors, + "feedback": { + actor: replay.feedback + for actor, replay in application.replays.items() + }, + "tools": { + actor: list(replay.tools) + for actor, replay in application.replays.items() + }, + }, + ) + for process in application.processes + ([runtime] if runtime else []): + if process.poll() is None: + os.killpg(process.pid, signal.SIGTERM) + try: + process.wait(timeout=5) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait() + server.shutdown() + server.server_close() + thread.join() + if runtime: + runtime_log.close() + if args.keep_logs: + destination = Path(args.keep_logs).resolve() + shutil.copytree(application.directory, destination) + print(f"Demo logs: {destination}") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="mode", required=True) + run = sub.add_parser("run") + run.add_argument("--worker", choices=("qwen", "deepseek"), default="qwen") + for name, default in ( + ("pitot", "pitot"), + ("codex", "codex"), + ("qwen", "qwen"), + ("deepseek", "dsh"), + ): + run.add_argument("--" + name, default=default) + run.add_argument("--temp-dir") + run.add_argument( + "--keep-logs", help="Copy owned demo profiles and logs to a new directory" + ) + control = sub.add_parser("controller") + control.add_argument("--connection", type=Path, required=True) + hook = sub.add_parser("qwen-hook") + hook.add_argument("--pitot", required=True) + hook.add_argument("--runtime", required=True) + action = sub.add_parser("act") + action.add_argument("action", choices=("dispatch", "build", "release")) + action.add_argument("--connection", type=Path, required=True) + args = parser.parse_args() + if args.mode == "controller": + controller(args.connection) + elif args.mode == "qwen-hook": + qwen_hook(args.pitot, args.runtime) + elif args.mode == "act": + settings = json.loads(args.connection.read_text()) + print( + json.dumps( + post( + args.connection, + "/act", + {"actor": settings["actor"], "action": args.action}, + 240, + ) + ) + ) + else: + demo(args) + + +if __name__ == "__main__": + main() diff --git a/examples/supervisory-control/replay.py b/examples/supervisory-control/replay.py new file mode 100644 index 0000000..5e8e459 --- /dev/null +++ b/examples/supervisory-control/replay.py @@ -0,0 +1,273 @@ +"""Tiny local model replay: proposes actions, never grants permission.""" + +from __future__ import annotations + +import json +import re +import uuid + + +class Replay: + def __init__(self, actor: str, commands: dict[str, str]): + self.actor = actor + self.commands = commands + self.last_call: str | None = None + self.last_action: str | None = None + self.tools: dict[str, dict] = {} + self.feedback: list[dict] = [] + + def propose(self, body: dict, responses: bool) -> tuple[str, dict, str] | None: + for item in body.get("tools", []): + function = item.get("function", item) + if isinstance(function, dict) and isinstance(function.get("name"), str): + self.tools[function["name"]] = function + action = "dispatch" if self.actor == "codex" else "release" + if self.last_call: + # Follow the result of the call we issued. Text in a prompt or an + # unrelated tool result must not advance the demonstration. + items = body.get("input", []) if responses else body.get("messages", []) + results = [ + item.get("output") if responses else item.get("content") + for item in items + if isinstance(item, dict) + and ( + item.get("type") == "function_call_output" + and item.get("call_id") == self.last_call + if responses + else item.get("role") == "tool" + and item.get("tool_call_id") == self.last_call + ) + ] + if not results: + raise ValueError( + "Replay requires the actual result of its last tool call" + ) + result = json.dumps(results[-1]) + self.feedback.append( + { + "call_id": self.last_call, + "action": self.last_action, + "result": results[-1], + } + ) + if self.actor == "codex": + if "DISPATCH_OK" in result: + return None + session = re.search(r"(?:session ID|session_id)[\s\":]+(\d+)", result) + if session and "write_stdin" in self.tools: + return self.issue( + "write_stdin", + { + "session_id": int(session[1]), + "chars": "", + "yield_time_ms": 30000, + }, + "dispatch", + ) + raise ValueError("Codex dispatch did not complete successfully") + if self.last_action == "release": + if "RELEASE_OK" in result: + return None + if "SUPERVISOR_DENY" not in result: + raise ValueError("Expected native release denial feedback") + action = "build" + elif self.last_action == "build": + if "BUILD_OK" not in result: + raise ValueError("Build did not complete; do not advance replay") + action = "release" + tool = next( + ( + name + for name in ( + "shell_command", + "exec_command", + "shell", + "run_shell_command", + "bash", + "pwsh", + ) + if name in self.tools + ), + None, + ) + if not tool: + raise ValueError( + f"No supported native shell tool advertised: {list(self.tools)}" + ) + command = self.commands[action] + arguments = {"cmd" if tool == "exec_command" else "command": command} + if tool == "exec_command": + arguments["yield_time_ms"] = 30000 + if tool == "shell": + arguments["command"] = ["/bin/sh", "-c", command] + if tool == "shell_command": + arguments["timeout_ms"] = 180000 + if tool == "run_shell_command": + arguments["is_background"] = False + parameters = self.tools[tool].get("parameters", {}) + if "description" in parameters.get("required", []): + arguments["description"] = f"Propose the delivery {action} action" + return self.issue(tool, arguments, action) + + def issue(self, tool: str, arguments: dict, action: str) -> tuple[str, dict, str]: + self.last_call = "call_" + uuid.uuid4().hex + self.last_action = action + return tool, arguments, self.last_call + + +def payload(call: tuple[str, dict, str] | None, responses: bool) -> dict: + identity = uuid.uuid4().hex + if not responses: + message = { + "role": "assistant", + "content": "Delivery complete." if call is None else None, + } + if call: + tool, arguments, call_id = call + message["tool_calls"] = [ + { + "id": call_id, + "type": "function", + "function": {"name": tool, "arguments": json.dumps(arguments)}, + } + ] + return { + "id": "chatcmpl_" + identity, + "object": "chat.completion", + "created": 1, + "model": "pitot-control", + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": "tool_calls" if call else "stop", + } + ], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, + } + if call: + tool, arguments, call_id = call + item = { + "id": "fc_" + identity, + "type": "function_call", + "call_id": call_id, + "name": tool, + "arguments": json.dumps(arguments), + "status": "completed", + } + else: + item = { + "id": "msg_" + identity, + "type": "message", + "role": "assistant", + "status": "completed", + "content": [ + {"type": "output_text", "text": "Delivery complete.", "annotations": []} + ], + } + return { + "id": "resp_" + identity, + "object": "response", + "created_at": 1, + "status": "completed", + "model": "pitot-control", + "output": [item], + "parallel_tool_calls": False, + "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}, + } + + +def stream_bytes(response: dict, responses: bool) -> bytes: + """The two native provider dialects; all data comes from the local replay.""" + if not responses: + choice = response["choices"][0] + delta = dict(choice["message"]) + for index, call in enumerate(delta.get("tool_calls", [])): + call["index"] = index + base = {**response, "object": "chat.completion.chunk"} + events = [ + ( + "", + { + **base, + "choices": [{"index": 0, "delta": delta, "finish_reason": None}], + }, + ), + ( + "", + { + **base, + "choices": [ + { + "index": 0, + "delta": {}, + "finish_reason": choice["finish_reason"], + } + ], + }, + ), + ] + else: + item = response["output"][0] + events = [ + ( + "response.created", + {"response": {**response, "status": "in_progress", "output": []}}, + ) + ] + start = {**item, "status": "in_progress"} + if item["type"] == "function_call": + start["arguments"] = "" + events.extend( + [ + ("response.output_item.added", {"output_index": 0, "item": start}), + ( + "response.function_call_arguments.delta", + { + "item_id": item["id"], + "output_index": 0, + "delta": item["arguments"], + }, + ), + ( + "response.function_call_arguments.done", + { + "item_id": item["id"], + "output_index": 0, + "arguments": item["arguments"], + }, + ), + ] + ) + else: + start["content"] = [] + part = item["content"][0] + fields = {"item_id": item["id"], "output_index": 0, "content_index": 0} + events.extend( + [ + ("response.output_item.added", {"output_index": 0, "item": start}), + ( + "response.content_part.added", + {**fields, "part": {**part, "text": ""}}, + ), + ("response.output_text.delta", {**fields, "delta": part["text"]}), + ("response.output_text.done", {**fields, "text": part["text"]}), + ("response.content_part.done", {**fields, "part": part}), + ] + ) + events.extend( + [ + ("response.output_item.done", {"output_index": 0, "item": item}), + ("response.completed", {"response": response}), + ] + ) + events = [ + (name, {"type": name, "sequence_number": index, **data}) + for index, (name, data) in enumerate(events) + ] + return b"".join( + ( + (f"event: {name}\n" if name else "") + f"data: {json.dumps(data)}\n\n" + ).encode() + for name, data in events + ) + (b"" if responses else b"data: [DONE]\n\n") diff --git a/examples/supervisory-control/supervisor.py b/examples/supervisory-control/supervisor.py new file mode 100644 index 0000000..4a0b7ae --- /dev/null +++ b/examples/supervisory-control/supervisor.py @@ -0,0 +1,139 @@ +"""Application-owned plant and supervisor; independent of agent protocols.""" + +from __future__ import annotations + +import hashlib +from pathlib import Path +import threading +import time + + +class Supervisor: + """One delivery session. Permission and completed transitions are separate.""" + + def __init__(self, project: Path, worker: str, commands: dict[str, str]): + self.project = project + self.worker = worker + self.commands = commands + self.state = "draft" + self.revision = 0 + self.tested_revision: int | None = None + self.tested_digest: str | None = None + # A changed state invalidates permissions issued against older state. + self.epoch = 0 + self.dispatched = False + self.pending: dict[tuple[str, str], tuple[int, float]] = {} + self.events: list[dict] = [] + self.lock = threading.RLock() + + def record(self, event: str, **details) -> None: + self.events.append({"event": event, "state": self.state, **details}) + + def permitted(self, actor: str, action: str) -> bool: + if action == "dispatch": + return actor == "codex" and not self.dispatched and self.state == "draft" + if actor != self.worker: + return False + if action == "build": + return self.state != "released" + if action == "release": + return self.state == "verified" and self.tested_revision == self.revision + return False + + def admit(self, actor: str, command: str, action_id: str) -> tuple[bool, str]: + with self.lock: + action = next((a for a, c in self.commands.items() if c == command), None) + allowed = bool(action and self.permitted(actor, action)) + reason = ( + f"SUPERVISOR_ALLOW {action} state={self.state}" + if allowed + else f"SUPERVISOR_DENY {action or 'unknown'} state={self.state}; " + "build and obtain a current external test result before release" + ) + self.record( + "admission", + actor=actor, + action=action, + allowed=allowed, + action_id=action_id, + reason=reason, + ) + if allowed: + # A short-lived, one-use reservation. A later event revokes it. + self.pending[(actor, action)] = (self.epoch, time.monotonic() + 30) + return allowed, reason + + def execute(self, actor: str, action: str) -> dict: + with self.lock: + # Consume permission once and check again: conditions may have changed + # since the agent received an allow response. + grant = self.pending.pop((actor, action), None) + if ( + not grant + or grant[0] != self.epoch + or grant[1] < time.monotonic() + or not self.permitted(actor, action) + ): + raise ValueError( + "No current admission reservation; propose the action again" + ) + if action == "dispatch": + self.dispatched = True + self.record("dispatch_completed", actor=actor, worker=self.worker) + return {"result": "DISPATCH_ACCEPTED"} + if action == "build": + # This is the actual toy build, not a state update at admission. + (self.project / "artifact.txt").write_text("release candidate\n") + self.revision += 1 + self.epoch += 1 + self.state = "built" + self.tested_revision = self.tested_digest = None + self.record("build_completed", actor=actor, revision=self.revision) + return {"result": "BUILD_OK", "revision": self.revision} + artifact = (self.project / "artifact.txt").read_bytes() + # A matching revision alone is insufficient if the bytes were edited. + if hashlib.sha256(artifact).hexdigest() != self.tested_digest: + self.invalidate("artifact changed since test") + raise ValueError("Tested artifact changed; release refused") + (self.project / "released.txt").write_bytes(artifact) + self.state = "released" + self.epoch += 1 + self.record("release_completed", actor=actor, revision=self.revision) + return {"result": "RELEASE_OK", "revision": self.revision} + + def observe_tests(self, revision: int, artifact: bytes) -> bool: + """Only the application tester calls this; no agent event endpoint.""" + with self.lock: + if self.state != "built" or revision != self.revision: + self.record("stale_test_ignored", revision=revision) + return False + if ( + artifact != b"release candidate\n" + or artifact != (self.project / "artifact.txt").read_bytes() + ): + self.invalidate("tests failed") + return False + self.tested_revision = revision + self.tested_digest = hashlib.sha256(artifact).hexdigest() + self.state = "verified" + self.epoch += 1 + self.record("tests_passed", revision=revision) + return True + + def invalidate(self, reason: str) -> None: + with self.lock: + if self.state in {"built", "verified"}: + self.state = "built" + self.tested_revision = self.tested_digest = None + self.epoch += 1 + self.record("tests_invalidated", reason=reason) + + def test_artifact(self, revision: int) -> dict: + """External tester runs after build, independently of the agent policy.""" + artifact = (self.project / "artifact.txt").read_bytes() + passed = self.observe_tests(revision, artifact) + if passed and revision == 1: + # Reproducible environment change, outside either agent's control. + self.invalidate("test environment changed") + with self.lock: + return {"result": "BUILD_OK", "revision": revision, "state": self.state} diff --git a/examples/supervisory-control/test_supervisor.py b/examples/supervisory-control/test_supervisor.py new file mode 100644 index 0000000..ec8970f --- /dev/null +++ b/examples/supervisory-control/test_supervisor.py @@ -0,0 +1,146 @@ +"""Deterministic application boundary checks; no agent or network required.""" + +from pathlib import Path +import json +import subprocess +import sys +import tempfile +import unittest + +from replay import Replay +from supervisor import Supervisor +from demo import Application + + +class SupervisorTest(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.project = Path(self.directory.name) + self.commands = {action: action for action in ("dispatch", "build", "release")} + self.supervisor = Supervisor(self.project, "qwen", self.commands) + + def build(self): + self.assertTrue(self.supervisor.admit("qwen", "build", "build-1")[0]) + return self.supervisor.execute("qwen", "build")["revision"] + + def test_permission_is_not_completion(self): + self.assertTrue(self.supervisor.admit("qwen", "build", "build-1")[0]) + self.assertEqual(self.supervisor.state, "draft") + self.assertFalse((self.project / "artifact.txt").exists()) + self.assertFalse(self.supervisor.admit("qwen", "release", "release-1")[0]) + + def test_external_event_changes_same_action(self): + revision = self.build() + self.assertFalse(self.supervisor.admit("qwen", "release", "release-1")[0]) + self.assertTrue(self.supervisor.observe_tests(revision, b"release candidate\n")) + self.assertTrue(self.supervisor.admit("qwen", "release", "release-2")[0]) + self.supervisor.invalidate("external environment change") + with self.assertRaises(ValueError): + self.supervisor.execute("qwen", "release") + self.assertFalse((self.project / "released.txt").exists()) + self.assertFalse(self.supervisor.admit("qwen", "release", "release-3")[0]) + self.assertTrue(self.supervisor.observe_tests(revision, b"release candidate\n")) + self.assertTrue(self.supervisor.admit("qwen", "release", "release-4")[0]) + self.supervisor.execute("qwen", "release") + self.assertEqual(self.supervisor.state, "released") + with self.assertRaises(ValueError): + self.supervisor.execute("qwen", "release") + + def test_failed_build_does_not_advance(self): + self.supervisor.admit("qwen", "build", "build-1") + (self.project / "artifact.txt").mkdir() + with self.assertRaises(OSError): + self.supervisor.execute("qwen", "build") + self.assertEqual(self.supervisor.state, "draft") + self.assertEqual(self.supervisor.revision, 0) + + def test_stale_test_and_changed_artifact_do_not_release(self): + revision = self.build() + self.build() + self.assertFalse( + self.supervisor.observe_tests(revision, b"release candidate\n") + ) + self.assertEqual(self.supervisor.state, "built") + self.supervisor.observe_tests(self.supervisor.revision, b"release candidate\n") + self.supervisor.admit("qwen", "release", "release-1") + (self.project / "artifact.txt").write_text("changed\n") + with self.assertRaises(ValueError): + self.supervisor.execute("qwen", "release") + self.assertFalse((self.project / "released.txt").exists()) + + def test_unknown_command_role_and_forged_event_denied(self): + for actor, command in ( + ("codex", "build"), + ("qwen", "dispatch"), + ("deepseek", "release"), + ("qwen", "tests_passed"), + ("qwen", "build; release"), + ): + self.assertFalse(self.supervisor.admit(actor, command, "bad-1")[0]) + with self.assertRaises(ValueError): + self.supervisor.execute("qwen", "tests_passed") + + def test_replay_requires_correlated_native_feedback(self): + replay = Replay("qwen", self.commands) + body = {"tools": [{"function": {"name": "run_shell_command"}}], "messages": []} + tool, arguments, call_id = replay.propose(body, False) + self.assertEqual(arguments["command"], "release") + body["messages"] = [ + {"role": "user", "content": "SUPERVISOR_DENY"}, + {"role": "tool", "tool_call_id": "unrelated", "content": "SUPERVISOR_DENY"}, + ] + with self.assertRaises(ValueError): + replay.propose(body, False) + body["messages"].append( + { + "role": "tool", + "tool_call_id": call_id, + "content": "SUPERVISOR_DENY release state=draft", + } + ) + self.assertEqual(replay.propose(body, False)[1]["command"], "build") + + def test_missing_native_event_and_uncorrelated_action_denied(self): + application = Application(self.project, "qwen", {}) + self.assertFalse( + application.admit({"action_id": "request-1", "data": None})["allowed"] + ) + event = { + "type": "action.requested", + "host": {"name": "qwen"}, + "action": {"id": "other-request", "kind": "shell"}, + "content": {"full": application.commands["build"]}, + } + self.assertFalse( + application.admit({"action_id": "request-1", "data": event})["allowed"] + ) + event["action"]["id"] = "request-1" + event["content"] = {"sha256": "opaque"} + self.assertFalse( + application.admit({"action_id": "request-1", "data": event})["allowed"] + ) + + def test_qwen_bridge_transport_failure_returns_native_deny(self): + result = subprocess.run( + [ + sys.executable, + str(Path(__file__).with_name("demo.py")), + "qwen-hook", + "--pitot", + str(self.project / "missing-pitot"), + "--runtime", + str(self.project / "runtime.json"), + ], + input=b"{}", + capture_output=True, + timeout=10, + check=True, + ) + decision = json.loads(result.stdout)["hookSpecificOutput"] + self.assertEqual(decision["permissionDecision"], "deny") + self.assertIn("Pitot unavailable", decision["permissionDecisionReason"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_supervisory_example.py b/tests/test_supervisory_example.py new file mode 100644 index 0000000..13cbeec --- /dev/null +++ b/tests/test_supervisory_example.py @@ -0,0 +1,27 @@ +"""Keep the standalone example's deterministic checks in repository CI.""" + +from pathlib import Path +import subprocess +import sys +import unittest + + +class SupervisoryExampleTest(unittest.TestCase): + def test_application_boundary(self): + root = Path(__file__).resolve().parents[1] + result = subprocess.run( + [ + sys.executable, + "-m", + "unittest", + "discover", + "-s", + "examples/supervisory-control", + "-v", + ], + cwd=root, + capture_output=True, + text=True, + timeout=30, + ) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr)