From 723ec25d0c636daee85e86f9997ad448f4ef42a0 Mon Sep 17 00:00:00 2001 From: Tuomo Vendelin Date: Sat, 10 Oct 2026 22:34:08 +0300 Subject: [PATCH] feat: add a read-only supervisor that observes and decides the next action Phase P3 of the autonomous Reconcile-loop plan. scripts/supervise.py observes the ledger, Git, PR/check state and session processes, and a pure decide() picks exactly one next action. It performs no effects: `next` prints the decision, `status` explains it, and `run --dry-run` logs decisions under a single-instance lock. - config: committed policy is read from the remote default branch; an ignored local file may override runner/budget keys only; autonomy needs committed grants. - policy: path globs, required checks on an exact SHA, and the L2 merge gate. - gh: read-only wrapper where any failure is Unknown, never success. - Table-driven decide tests, plus CLI tests against real Git and a fake gh. Co-Authored-By: Claude Opus 5.5 --- .agent/supervisor.example.toml | 85 +++++++ SUPERVISION.md | 72 +++++- scripts/supervise.py | 163 +++++++++++++ scripts/supervisor/__init__.py | 0 scripts/supervisor/config.py | 203 ++++++++++++++++ scripts/supervisor/decide.py | 283 ++++++++++++++++++++++ scripts/supervisor/gh.py | 63 +++++ scripts/supervisor/observe.py | 208 ++++++++++++++++ scripts/supervisor/policy.py | 124 ++++++++++ tests/fakes/gh | 35 +++ tests/test_supervisor_cli.py | 255 +++++++++++++++++++ tests/test_supervisor_config.py | 155 ++++++++++++ tests/test_supervisor_decide.py | 417 ++++++++++++++++++++++++++++++++ tests/test_supervisor_policy.py | 179 ++++++++++++++ 14 files changed, 2239 insertions(+), 3 deletions(-) create mode 100644 .agent/supervisor.example.toml create mode 100644 scripts/supervise.py create mode 100644 scripts/supervisor/__init__.py create mode 100644 scripts/supervisor/config.py create mode 100644 scripts/supervisor/decide.py create mode 100644 scripts/supervisor/gh.py create mode 100644 scripts/supervisor/observe.py create mode 100644 scripts/supervisor/policy.py create mode 100755 tests/fakes/gh create mode 100644 tests/test_supervisor_cli.py create mode 100644 tests/test_supervisor_config.py create mode 100644 tests/test_supervisor_decide.py create mode 100644 tests/test_supervisor_policy.py diff --git a/.agent/supervisor.example.toml b/.agent/supervisor.example.toml new file mode 100644 index 0000000..ce2f9cc --- /dev/null +++ b/.agent/supervisor.example.toml @@ -0,0 +1,85 @@ +# Copy to .agent/supervisor.toml and commit it. The supervisor reads this file from the remote +# default branch, never from the working tree; see SUPERVISION.md. An ignored +# .agent/supervisor.local.toml may override [runner] and [budgets] keys only. + +[supervisor] +# 1: plan, execute and review automatically; a person merges and reconciles. +# 2: also merge when [policy.merge] grants it. 3: also reconcile and start the next round. +autonomy = 1 +remote = "origin" +# Agent worktrees live outside the repository. Empty means ../-worktrees. +worktree_root = "" +max_parallel = 1 +max_assignments = 1 +poll_seconds = 60 +launch_grace_seconds = 120 + +[budgets] +round_usd = 10.0 +sessions_per_round = 8 +max_rounds_per_run = 1 +ci_timeout_minutes = 60 + +[stop] +# Park after this many consecutive closed rounds with no behavior change. +stall_rounds = 2 +max_repairs = 2 + +[notify] +# argv run when the supervisor's outcome changes; empty disables notification. +command = [] + +[policy.merge] +auto_merge = false +method = "squash" +# Check names that must succeed on the exact reviewed head before an automatic merge. +required_checks = [] +protected_paths = [".github/**", ".agent/**", "AGENTS.md", "SPEC.md", "SUPERVISION.md"] +max_changed_lines = 400 +publish_reviews = true + +[policy.reconcile] +auto_reconcile = false +allow_reconcile_merge = false +# Tracked project records a reconcile assignment may change. +record_paths = [] + +[runner] +default = "claude" +# Credentials removed from every agent environment; the supervisor publishes, agents do not. +env_drop = ["GH_TOKEN", "GITHUB_TOKEN", "GH_ENTERPRISE_TOKEN", "SSH_AUTH_SOCK"] + +[runner.commands.claude] +# Placeholders are substituted per argv element; no shell is involved. +argv = [ + "claude", "-p", "--output-format", "json", "--no-session-persistence", + "--session-id", "{session_uuid}", "--model", "{model}", + "--permission-mode", "{permission_mode}", "{prompt}", +] +# Envelope field names are confirmed against a recorded run before the runner is enabled. +output = { result = "result", is_error = "is_error", cost_usd = "total_cost_usd", session_id = "session_id" } + +[runner.roles.planner] +model = "opus" +budget_usd = 3.0 +timeout_minutes = 30 +permission_mode = "default" + +[runner.roles.executor] +model = "opus" +budget_usd = 5.0 +timeout_minutes = 60 +permission_mode = "acceptEdits" + +[runner.roles.reviewer] +# Prefer a different model from the executor to reduce correlated review. +model = "sonnet" +budget_usd = 2.0 +timeout_minutes = 30 +permission_mode = "default" + +[runner.roles.reconciler] +model = "opus" +budget_usd = 3.0 +timeout_minutes = 30 +permission_mode = "acceptEdits" diff --git a/SUPERVISION.md b/SUPERVISION.md index 45e3e08..5efe9fc 100644 --- a/SUPERVISION.md +++ b/SUPERVISION.md @@ -5,7 +5,8 @@ supervisor can drive rounds unattended. Main remains the lean manual workflow; t superset of it and receives main by merge. Everything here is additive: a round created without `--supervised` is a version 1 round and behaves exactly as on main. -> Status: draft. The ledger records below exist; the supervisor that uses them is not built yet. +> Status: draft. The ledger records below exist. The supervisor observes and decides but performs +> no effects yet: it cannot launch agents, push, publish or merge. ## Authority @@ -93,7 +94,72 @@ planner reporting no executable work or an owed decision. The inventory lists li worktrees and resources at that moment. While any halt is uncleared, no session may start and the round cannot close. Only an operator clears a halt, with evidence of what was inspected or resolved. +## Supervisor (read-only preview) + +`scripts/supervise.py` is a stateless loop manager. Each tick it observes the ledger (through +`round.py status`), Git, PR and check state (through `gh`; any failure is Unknown) and session +processes, then a pure `decide()` picks exactly one next action. In this version it only reports: + +```sh +python3 scripts/supervise.py next # one decision as JSON; writes nothing +python3 scripts/supervise.py status # round, assignments, sessions, halts, decision +python3 scripts/supervise.py run --dry-run # log decisions; poll while waiting +``` + +Add `--new-round` to let a level 1 or 2 supervisor propose one new round. `--autonomy N` can only +lower the level. Exit codes: 0 action, wait or routine park; 10 needs attention (halt, hold, stop, +stall or budget); 11 transient or unknown observation; 2 configuration; 3 another instance holds +`.agent-runs/supervisor/lock`. `run` appends each decision to `.agent-runs/supervisor/log.jsonl` +and keeps the latest in `last.json`; it stops at the first decision that needs an effect. + +### Configuration + +Copy [the example](.agent/supervisor.example.toml) to `.agent/supervisor.toml` and commit it. The +supervisor reads it from the remote default branch (`git show REMOTE/BRANCH:...`), never from the +working tree, so an agent cannot raise its own authority by editing it. An ignored +`.agent/supervisor.local.toml` may override `[runner]` and `[budgets]` keys only. Unknown keys fail. +Level 2 needs `policy.merge.auto_merge` and non-empty `required_checks`; level 3 also needs +`policy.reconcile.auto_reconcile` and `allow_reconcile_merge`. A missing grant exits 2 naming it. + +### Decision rules + +First match wins: more than one open round stops; a version 1 round is parked as manual; live +sessions are contained even while halted (kill past deadline, mark lost, acknowledge); an uncleared +halt parks; a dirty coordinator checkout or one with commits not on the remote default stops; an +unfetched remote head asks for a fetch; exited sessions are imported and in-flight merges verified. +Without an open round the supervisor parks on stall (`stop.stall_rounds` closed rounds with delta +`none`), run limit or autonomy below 3 unless `--new-round`, else proposes the next round at the +remote default head. A round without assignments launches one planner. Each idle assignment then +yields at most one candidate, chosen as Merge > LaunchReviewer > LaunchExecutor > Cleanup within +`max_parallel` and the round budget; a hold on one assignment never blocks another: + +| Assignment | Decision | +| --- | --- | +| `assigned` | launch executor | +| `blocked` result | hold | +| result, PR missing, closed, merged externally or head moved | hold | +| result, no review of its head | launch reviewer | +| `needs-fix` | launch repair executor, or hold after `stop.max_repairs` | +| review `blocked` | hold | +| `ready`, level 1 | park awaiting a person's merge | +| `ready`, level 2+ | merge gate: merge, wait for checks, or hold | +| `merged`/`cancelled` with a worktree | clean up, or hold if the worktree is dirty | + +When every assignment is terminal, levels 1 and 2 park awaiting reconciliation; level 3 creates a +reconcile assignment from the remote head, runs the final gate after it merges and closes the round. + +### Merge gate + +All must hold: the latest review is `ready` for the latest result head; the PR is open, not a +draft, at that head, from the assigned branch into the default branch, `MERGEABLE` and `CLEAN` or +`HAS_HOOKS`; every `required_checks` name succeeded on that exact SHA (newest run wins; missing, +skipped, neutral and cancelled fail; pending past `ci_timeout_minutes` after the review holds); every +changed path is inside the owned globs and outside `protected_paths`; the change is within +`max_changed_lines` and has no binary files; a reconcile assignment also needs +`allow_reconcile_merge`. A branch behind or conflicting with the default branch holds; the supervisor +never updates branches or bypasses protection. + ## Planned -The supervisor (`scripts/supervise.py`), role contracts, merge policy and reconciliation automation -follow in later changes on this branch. +Launching agents, importing their output, publishing reviews, merging and reconciliation follow in +later changes on this branch. diff --git a/scripts/supervise.py b/scripts/supervise.py new file mode 100644 index 0000000..ce5df09 --- /dev/null +++ b/scripts/supervise.py @@ -0,0 +1,163 @@ +"""Deterministic Reconcile-loop supervisor (read-only preview). See SUPERVISION.md. + +Each tick observes the ledger, Git, PRs/checks and session processes, then decides one next action. +This version performs no effects: `next` prints the decision, `status` explains it and +`run --dry-run` logs decisions while waiting. Exit codes: 0 action/wait/routine park, +10 needs attention, 11 transient or unknown observation, 2 configuration, 3 another instance runs. +""" + +import argparse +import fcntl +import json +import subprocess +import sys +import time +from datetime import datetime, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from supervisor import config, observe # noqa: E402 +from supervisor.decide import decide # noqa: E402 + +STATE = ".agent-runs/supervisor" +PASSIVE = {"Wait"} + + +def utc_now(): + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def default_branch(repo, remote): + done = subprocess.run( + ["git", "-C", str(repo), "symbolic-ref", "--short", f"refs/remotes/{remote}/HEAD"], + text=True, capture_output=True, check=False, + ) + prefix = f"{remote}/" + name = done.stdout.strip() + return name[len(prefix):] if done.returncode == 0 and name.startswith(prefix) else "main" + + +def tick(repo, args, settings, autonomy, rounds_this_run=0): + value = observe.snapshot( + repo, settings, autonomy, args.branch, + new_round=args.new_round, rounds_this_run=rounds_this_run, + ) + return value, decide(value) + + +def describe(value, chosen): + lines = [f"Autonomy: L{value['autonomy']} default: {value['default_branch']}" + f" @ {(value['remote']['default_sha'] or 'unknown')[:12]}"] + manifest = value["round"] + if manifest is None: + closed = ", ".join(f"{r['id']}({r['delta']})" for r in value["rounds"]["closed"]) or "none" + lines.append(f"Open round: none closed: {closed}") + else: + lines.append(f"Open round: {manifest['id']} v{manifest['version']} base {manifest['base_sha'][:12]}") + for aid, a in sorted(manifest["assignments"].items()): + reviews = a.get("reviews") or [] + verdict = f" review={reviews[-1]['verdict']}" if reviews else "" + lines.append(f" {aid} {a['state']} {a['branch']} results={len(a['results'])}{verdict}") + for name, s in sorted(manifest.get("sessions", {}).items()): + lines.append(f" {name} {s['role']} {s['state']} {s['assignment'] or '-'}") + for h in manifest.get("halts", []): + state = "cleared" if h["cleared_utc"] else "UNCLEARED" + lines.append(f" {h['id']} {state} {h['code']}: {h['detail']}") + details = {k: v for k, v in chosen.items() if k not in {"kind", "exit"}} + lines.append(f"Next: {chosen['kind']} (exit {chosen['exit']}) {json.dumps(details, sort_keys=True)}") + return "\n".join(lines) + + +def hold_lock(repo): + state = repo / STATE + state.mkdir(parents=True, exist_ok=True) + handle = (state / "lock").open("a") + try: + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + return None + return handle + + +def record(repo, value, chosen, dry_run): + entry = {"utc": utc_now(), "dry_run": dry_run, "round": (value["round"] or {}).get("id"), + "action": chosen} + state = repo / STATE + with (state / "log.jsonl").open("a") as log: + log.write(json.dumps(entry, sort_keys=True) + "\n") + temporary = state / "last.json.tmp" + temporary.write_text(json.dumps(entry, indent=2, sort_keys=True) + "\n") + temporary.replace(state / "last.json") + + +def run(repo, args, settings, autonomy): + if not args.dry_run: + print("Effects are not available in this version; use run --dry-run", file=sys.stderr) + return 2 + ignored = subprocess.run( + ["git", "-C", str(repo), "check-ignore", "-q", f"{STATE}/lock"], check=False + ) + if ignored.returncode: + print("Add .agent-runs/ to .gitignore before running the supervisor", file=sys.stderr) + return 2 + lock = hold_lock(repo) + if lock is None: + print("Another supervisor instance holds the lock", file=sys.stderr) + return 3 + while True: + value, chosen = tick(repo, args, settings, autonomy) + record(repo, value, chosen, dry_run=True) + print(json.dumps(chosen, sort_keys=True), flush=True) + if args.once or chosen["kind"] not in PASSIVE: + # A dry run cannot perform the chosen effect, so the state will not advance further. + return chosen["exit"] + time.sleep(settings["supervisor"]["poll_seconds"]) + + +def parser(): + command = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) + command.add_argument("--repo", type=Path, default=Path.cwd()) + command.add_argument("--remote", default="origin") + command.add_argument("--branch", help="Default branch (default: the remote's HEAD)") + command.add_argument("--autonomy", type=int, choices=[1, 2, 3], help="Lower the level") + actions = command.add_subparsers(dest="action", required=True) + for name in ("next", "status", "run"): + sub = actions.add_parser(name) + sub.add_argument("--new-round", action="store_true", help="Allow one new round below L3") + if name == "run": + sub.add_argument("--once", action="store_true") + sub.add_argument("--dry-run", action="store_true") + return command + + +def main(argv=None): + args = parser().parse_args(argv) + repo = args.repo.resolve() + args.branch = args.branch or default_branch(repo, args.remote) + try: + settings = config.load(repo, args.remote, args.branch) + if settings["supervisor"]["remote"] != args.remote: + raise config.ConfigError( + f"supervisor.remote is {settings['supervisor']['remote']!r}; pass --remote to match" + ) + autonomy = config.effective_autonomy(settings, args.autonomy) + except config.ConfigError as error: + print(f"Configuration: {error}", file=sys.stderr) + return 2 + try: + if args.action == "run": + return run(repo, args, settings, autonomy) + value, chosen = tick(repo, args, settings, autonomy) + except observe.ObservationError as error: + print(f"Cannot observe: {error}", file=sys.stderr) + return 10 + if args.action == "next": + print(json.dumps(chosen, sort_keys=True)) + else: + print(describe(value, chosen)) + return chosen["exit"] + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/supervisor/__init__.py b/scripts/supervisor/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scripts/supervisor/config.py b/scripts/supervisor/config.py new file mode 100644 index 0000000..38f30b6 --- /dev/null +++ b/scripts/supervisor/config.py @@ -0,0 +1,203 @@ +"""Load and validate supervisor settings; derive the autonomy that committed policy grants.""" + +import copy +import subprocess +import tomllib +from pathlib import Path + +COMMITTED = ".agent/supervisor.toml" +LOCAL = ".agent/supervisor.local.toml" +ROLES = ("planner", "executor", "reviewer", "reconciler") +LOCAL_SECTIONS = {"runner", "budgets"} + +# Every key and its default; the default's type is the required type. +DEFAULTS = { + "supervisor": { + "autonomy": 1, + "remote": "origin", + "worktree_root": "", + "max_parallel": 1, + "max_assignments": 3, + "poll_seconds": 60, + "launch_grace_seconds": 120, + }, + "budgets": { + "round_usd": 20.0, + "sessions_per_round": 12, + "max_rounds_per_run": 1, + "ci_timeout_minutes": 60, + }, + "stop": {"stall_rounds": 2, "max_repairs": 2}, + "notify": {"command": []}, + "policy": { + "merge": { + "auto_merge": False, + "method": "squash", + "required_checks": [], + "protected_paths": [], + "max_changed_lines": 400, + "publish_reviews": True, + }, + "reconcile": { + "auto_reconcile": False, + "allow_reconcile_merge": False, + "record_paths": [], + }, + }, + "runner": {"default": "claude", "env_drop": [], "commands": {}, "roles": {}}, +} +COMMAND = {"argv": [], "output": {}} +ROLE = { + "runner": "", + "model": "", + "budget_usd": 5.0, + "timeout_minutes": 30, + "permission_mode": "", + "allowed_tools": [], + "disallowed_tools": [], +} +POSITIVE = { + "supervisor.autonomy", + "supervisor.max_parallel", + "supervisor.max_assignments", + "supervisor.poll_seconds", + "budgets.sessions_per_round", + "budgets.max_rounds_per_run", + "budgets.ci_timeout_minutes", + "stop.stall_rounds", + "policy.merge.max_changed_lines", + "timeout_minutes", +} +CHOICES = {"supervisor.autonomy": {1, 2, 3}, "policy.merge.method": {"squash", "merge", "rebase"}} + + +class ConfigError(ValueError): + pass + + +def fail(message): + raise ConfigError(message) + + +def check_value(name, value, default): + leaf = name.rsplit(".", 1)[-1] + if isinstance(default, bool) or isinstance(default, str): + ok = type(value) is type(default) + elif isinstance(default, float): + ok = type(value) in (int, float) + elif isinstance(default, int): + ok = type(value) is int + else: + ok = isinstance(value, list) and all(isinstance(v, str) and v for v in value) + if not ok: + fail(f"{name} must be {type(default).__name__}" + (" of strings" if isinstance(default, list) else "")) + if isinstance(value, (int, float)) and not isinstance(value, bool): + if (name in POSITIVE or leaf in POSITIVE) and value < 1: + fail(f"{name} must be at least 1") + if value < 0: + fail(f"{name} must not be negative") + if name in CHOICES and value not in CHOICES[name]: + fail(f"{name} must be one of {sorted(CHOICES[name])}") + return float(value) if isinstance(default, float) else value + + +def merge(name, value, defaults): + if not isinstance(value, dict): + fail(f"{name} must be a table") + out = copy.deepcopy(defaults) + for key, item in value.items(): + path = f"{name}.{key}" if name else key + if key not in defaults: + fail(f"Unknown setting {path}") + if path == "runner.commands": + out[key] = {n: merge(f"{path}.{n}", v, COMMAND) for n, v in check_table(path, item)} + for n, command in out[key].items(): + if not command["argv"]: + fail(f"{path}.{n}.argv must not be empty") + for field, target in command["output"].items(): + if not isinstance(target, str) or not target: + fail(f"{path}.{n}.output.{field} must be a string") + elif path == "runner.roles": + for role, _ in check_table(path, item): + if role not in ROLES: + fail(f"Unknown role {path}.{role}") + out[key] = {role: merge(f"{path}.{role}", v, ROLE) for role, v in item.items()} + elif path.endswith(".output"): + out[key] = dict(check_table(path, item)) + elif isinstance(defaults[key], dict): + out[key] = merge(path, item, defaults[key]) + else: + out[key] = check_value(path, item, defaults[key]) + return out + + +def check_table(name, value): + if not isinstance(value, dict): + fail(f"{name} must be a table") + return value.items() + + +def deep_update(base, extra): + for key, value in extra.items(): + if isinstance(value, dict) and isinstance(base.get(key), dict): + deep_update(base[key], value) + else: + base[key] = value + + +def parse(committed, local=None): + """Validate committed TOML text, then apply a local override of runner/budget keys only.""" + try: + data = tomllib.loads(committed) + override = tomllib.loads(local) if local is not None else {} + except tomllib.TOMLDecodeError as error: + fail(f"Invalid TOML: {error}") + extra = set(override) - LOCAL_SECTIONS + if extra: + fail(f"The local override may set only {sorted(LOCAL_SECTIONS)}, not {sorted(extra)}") + deep_update(data, override) + return merge("", data, DEFAULTS) + + +def effective_autonomy(settings, ceiling=None): + """The configured level, refused unless committed policy grants it; a ceiling only lowers.""" + level = settings["supervisor"]["autonomy"] + merge_policy = settings["policy"]["merge"] + reconcile = settings["policy"]["reconcile"] + missing = [] + if level >= 2: + missing += [g for g in ("auto_merge",) if not merge_policy[g]] + missing += [] if merge_policy["required_checks"] else ["required_checks"] + if level >= 3: + missing += [g for g in ("auto_reconcile", "allow_reconcile_merge") if not reconcile[g]] + if missing: + fail(f"Autonomy {level} needs committed grants: {', '.join(missing)}") + return min(level, ceiling) if ceiling else level + + +def git(repo, *args): + return subprocess.run( + ["git", "-C", str(repo), *args], text=True, capture_output=True, check=False + ) + + +def load(repo, remote, branch): + """Read policy from the committed default branch, never from the working tree.""" + repo = Path(repo).resolve() + shown = git(repo, "show", f"refs/remotes/{remote}/{branch}:{COMMITTED}") + if shown.returncode: + fail(f"No committed {COMMITTED} on {remote}/{branch}: {shown.stderr.strip()}") + local = repo / LOCAL + local_text = None + if local.exists(): + if git(repo, "check-ignore", "-q", LOCAL).returncode: + fail(f"{LOCAL} must be ignored by Git") + local_text = local.read_text() + settings = parse(shown.stdout, local_text) + supervisor = settings["supervisor"] + root = Path(supervisor["worktree_root"] or f"../{repo.name}-worktrees") + root = (repo / root).resolve() if not root.is_absolute() else root.resolve() + if root == repo or root.is_relative_to(repo): + fail(f"supervisor.worktree_root must be outside the repository: {root}") + supervisor["worktree_root"] = str(root) + return settings diff --git a/scripts/supervisor/decide.py b/scripts/supervisor/decide.py new file mode 100644 index 0000000..5b71e55 --- /dev/null +++ b/scripts/supervisor/decide.py @@ -0,0 +1,283 @@ +"""Pure next-action decision for the supervisor. + +`decide(snapshot)` reads one observation snapshot (see observe.py) and returns exactly one action. +It performs no I/O. Rules, first match wins: + +1. Ledger shape: more than one open round stops; a version 1 open round is parked as manual. +2. Containment of live sessions, even while halted: kill past deadline, mark lost, acknowledge. +3. An uncleared halt parks the supervisor. A dirty or diverged coordinator checkout stops it. +4. Settle finished work: import an exited session, verify an in-flight merge. +5. No open round: stall, run limit and autonomy checks, else start the next round. +6. Planning: launch one planner while the round has no assignments. +7. Per assignment, aggregated as Merge > LaunchReviewer > LaunchExecutor > Cleanup within the + parallel and budget limits; a hold on one assignment never blocks another. +8. Round end: reconcile, run the final gate and close, as autonomy allows. + +Exit codes: 0 action, wait or routine park; 10 needs attention; 11 transient or unknown. +""" + +from datetime import datetime, timezone + +from .policy import merge_gate + +LIVE_SESSION = {"launching", "running"} +TERMINAL = {"merged", "cancelled"} +ATTENTION_PARKS = {"halted", "stalled", "budget"} + + +def action(kind, exit=0, **fields): + return {"kind": kind, "exit": exit, **fields} + + +def epoch(value): + return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc).timestamp() + + +def park(reason, **fields): + return action("Park", 10 if reason in ATTENTION_PARKS else 0, reason=reason, **fields) + + +def latest_review(assignment): + if not assignment["results"]: + return None + current = [r for r in assignment["reviews"] if r["result"] == assignment["results"][-1]] + return current[-1] if current else None + + +def next_round_id(ids): + numbers = [int(i[1:]) for i in ids if i[:1] == "R" and i[1:].isdigit()] + candidate = max(numbers, default=0) + 1 + while f"R{candidate}" in ids: + candidate += 1 + return f"R{candidate}" + + +def contain(snapshot, sessions): + """Return the most urgent containment action for live agent processes, if any.""" + now = snapshot["now"] + grace = snapshot["limits"]["launch_grace_seconds"] + found = {} + for name, record in sorted(sessions.items()): + if record["state"] not in LIVE_SESSION or record["role"] == "merger": + continue + process = snapshot["processes"].get(name) or {"pid": None, "alive": False, "exit": None} + if process["exit"] is not None: + continue + if process["alive"] and now > epoch(record["deadline_utc"]): + found.setdefault("KillSession", name) + elif not process["alive"] and ( + record["state"] == "running" or now - epoch(record["started_utc"]) > grace + ): + found.setdefault("MarkLost", name) + elif process["alive"] and record["state"] == "launching": + found.setdefault("AckSession", name) + for kind in ("KillSession", "MarkLost", "AckSession"): + if kind in found: + name = found[kind] + return action(kind, session=name, pid=(snapshot["processes"].get(name) or {}).get("pid")) + return None + + +def settle(snapshot, sessions): + for name, record in sorted(sessions.items()): + if record["state"] not in LIVE_SESSION or record["role"] == "merger": + continue + process = snapshot["processes"].get(name) or {} + if process.get("exit") is not None: + return action("ImportSession", session=name, role=record["role"]) + for name, record in sorted(sessions.items()): + if record["state"] in LIVE_SESSION and record["role"] == "merger": + return action("VerifyMerge", session=name, assignment=record["assignment"]) + return None + + +def start_round(snapshot): + limits = snapshot["limits"] + closed = snapshot["rounds"]["closed"] + recent = closed[-limits["stall_rounds"] :] + if len(recent) == limits["stall_rounds"] and all(r["delta"] == "none" for r in recent): + return park("stalled", rounds=[r["id"] for r in recent]) + if snapshot["rounds_this_run"] >= limits["max_rounds_per_run"]: + return park("max-rounds") + if snapshot["autonomy"] < 3 and not snapshot["new_round"]: + return park("no-round") + return action( + "InitRound", + round_id=next_round_id(snapshot["rounds"]["ids"]), + base=snapshot["remote"]["default_sha"], + ) + + +def evaluate(snapshot, aid, a, notes): + """Return a candidate action for one idle assignment, or record why there is none.""" + state = a["state"] + + def hold(reason): + notes["holds"].append({"assignment": aid, "reason": reason}) + if state in TERMINAL: + tree = snapshot["worktrees"].get(aid) + if tree and tree["exists"]: + if tree["dirty"]: + return hold("worktree-dirty") + return action("Cleanup", assignment=aid) + return None + if state == "assigned": + return action("LaunchExecutor", assignment=aid, repair=False) + if state == "blocked": + return hold("result-blocked") + head = snapshot["results"][aid]["head_sha"] + observed = snapshot["prs"].get(aid) or {"unknown": "not observed"} + if "unknown" in observed: + notes["unknown"].append({"assignment": aid, "reason": observed["unknown"]}) + return None + pr = observed["pr"] + if pr is None: + return hold("pr-missing") + if pr["state"] == "MERGED": + return hold("merged-externally") + if pr["state"] != "OPEN": + return hold("pr-closed") + if pr["headRefOid"] != head: + return hold("head-moved") + review = latest_review(a) + if review is None: + return action("LaunchReviewer", assignment=aid, head=head) + if review["verdict"] == "needs-fix": + if len(a["results"]) - 1 >= snapshot["limits"]["max_repairs"]: + return hold("repairs-exhausted") + return action("LaunchExecutor", assignment=aid, repair=True) + if review["verdict"] != "ready": + return hold("review-blocked") + if snapshot["autonomy"] < 2: + notes["parks"].append({"assignment": aid, "reason": "awaiting-merge"}) + return None + checks = snapshot["checks"].get(aid) or {"unknown": "not observed"} + if "unknown" in checks: + notes["unknown"].append({"assignment": aid, "reason": checks["unknown"]}) + return None + diff = snapshot["diffs"].get(aid) or {"files": None, "changed_lines": 0} + timeout = snapshot["limits"]["ci_timeout_minutes"] * 60 + facts = { + "kind": a["kind"], + "branch": a["branch"], + "default_branch": snapshot["default_branch"], + "result_head": head, + "review": review, + "pr": pr, + "checks": checks["state"], + "checks_timed_out": snapshot["now"] - epoch(review["utc"]) > timeout, + "files": diff["files"], + "changed_lines": diff["changed_lines"], + "binary": diff.get("binary", []), + "owned_paths": a["owned_paths"], + } + verdict, reasons = merge_gate(facts, snapshot["gate"]) + if verdict == "merge": + return action("Merge", assignment=aid, head=head) + for reason in reasons: + (notes["holds"] if verdict == "hold" else notes["waits"]).append( + {"assignment": aid, "reason": reason} + ) + return None + + +def finish_round(snapshot, assignments): + reconcile = [a for a in assignments.values() if a["kind"] == "reconcile"] + if any(a["state"] == "merged" for a in reconcile): + gate = snapshot["final_gate"] + if gate is None: + return action("RunFinalGate") + if gate == "passed": + return action("CloseRound") + return action("Hold", 10, holds=[{"assignment": None, "reason": "final-gate-failed"}]) + if snapshot["autonomy"] < 3: + return park("awaiting-reconcile") + if reconcile: + return action("Hold", 10, holds=[{"assignment": None, "reason": "reconcile-cancelled"}]) + return action("CreateReconcile", start=snapshot["remote"]["default_sha"]) + + +def decide(snapshot): + rounds = snapshot["rounds"] + if len(rounds["open"]) > 1: + return action("Stop", 10, reason="multiple-open-rounds", rounds=rounds["open"]) + manifest = snapshot["round"] + if manifest and manifest["version"] != 2: + return park("manual-round", round=manifest["id"]) + sessions = manifest["sessions"] if manifest else {} + found = contain(snapshot, sessions) + if found: + return found + if manifest: + halts = [h["id"] for h in manifest["halts"] if not h["cleared_utc"]] + if halts: + return park("halted", halts=halts) + if not snapshot["checkout"]["clean"]: + return action("Stop", 10, reason="dirty-checkout") + if snapshot["remote"]["default_sha"] is None: + return action("Unknown", 11, reason=snapshot["remote"]["unknown"]) + if snapshot["checkout"]["head_on_default"] is None: + return action("Fetch") + if not snapshot["checkout"]["head_on_default"]: + return action("Stop", 10, reason="diverged-checkout") + if not manifest: + return start_round(snapshot) + found = settle(snapshot, sessions) + if found: + return found + + limits = snapshot["limits"] + workers = [s for s in sessions.values() if s["role"] != "merger"] + live = [s for s in workers if s["state"] in LIVE_SESSION] + spent = sum(s.get("cost_usd") or 0 for s in workers) + budget_left = len(workers) < limits["sessions_per_round"] and spent < limits["round_usd"] + slots = limits["max_parallel"] - len(live) + assignments = manifest["assignments"] + + if not assignments: + if any(s["role"] == "planner" for s in live): + return action("Wait", reason="planner-running") + if any(s["role"] == "planner" for s in workers): + return action("Hold", 10, holds=[{"assignment": None, "reason": "planner-produced-nothing"}]) + if not budget_left: + return park("budget") + return action("LaunchPlanner") + + notes = {"holds": [], "waits": [], "unknown": [], "parks": []} + busy = {s["assignment"] for s in live} + candidates = [] + for aid in sorted(assignments): + if aid in busy: + notes["waits"].append({"assignment": aid, "reason": "session-running"}) + continue + found = evaluate(snapshot, aid, assignments[aid], notes) + if found: + candidates.append(found) + extras = {"holds": notes["holds"]} if notes["holds"] else {} + launch_blocked = False + for kind in ("Merge", "LaunchReviewer", "LaunchExecutor", "Cleanup"): + for found in candidates: + if found["kind"] != kind: + continue + if kind.startswith("Launch"): + if slots < 1: + notes["waits"].append({"assignment": found["assignment"], "reason": "parallel-limit"}) + continue + if not budget_left: + launch_blocked = True + continue + return {**found, **extras} + + if all(a["state"] in TERMINAL for a in assignments.values()) and not notes["holds"]: + return finish_round(snapshot, assignments) + if notes["holds"]: + return action("Hold", 10, holds=notes["holds"]) + if notes["unknown"]: + return action("Unknown", 11, reason="observation-unavailable", unknown=notes["unknown"]) + if notes["waits"]: + return action("Wait", waits=notes["waits"]) + if launch_blocked: + return park("budget") + if notes["parks"]: + return park("awaiting-merge", assignments=[p["assignment"] for p in notes["parks"]]) + return action("Hold", 10, holds=[{"assignment": None, "reason": "no-rule-matched"}]) diff --git a/scripts/supervisor/gh.py b/scripts/supervisor/gh.py new file mode 100644 index 0000000..9fe9060 --- /dev/null +++ b/scripts/supervisor/gh.py @@ -0,0 +1,63 @@ +"""Thin, read-only GitHub CLI wrapper. Any failure is Unknown, never an empty success.""" + +import json +import subprocess + +PR_FIELDS = "number,state,isDraft,headRefOid,headRefName,baseRefName,mergeable,mergeStateStatus,url" + + +class Unknown: + def __init__(self, reason): + self.reason = reason + + def __repr__(self): + return f"Unknown({self.reason!r})" + + +def call(repo, *args): + try: + done = subprocess.run( + ["gh", *args], cwd=repo, text=True, capture_output=True, timeout=60, check=False + ) + except (OSError, subprocess.TimeoutExpired) as error: + return Unknown(f"gh {args[0]}: {error}") + if done.returncode: + return Unknown(f"gh {args[0]} exited {done.returncode}: {done.stderr.strip()[:200]}") + return done.stdout + + +def json_lines(text): + try: + return [json.loads(line) for line in text.splitlines() if line.strip()] + except json.JSONDecodeError as error: + return Unknown(f"gh returned invalid JSON: {error}") + + +def pull_request(repo, branch): + """The open PR for a branch, else the newest one; None when the branch has no PR.""" + out = call(repo, "pr", "list", "--head", branch, "--state", "all", "--json", PR_FIELDS) + if isinstance(out, Unknown): + return out + try: + prs = json.loads(out) + except json.JSONDecodeError as error: + return Unknown(f"gh pr list returned invalid JSON: {error}") + if not prs: + return None + return sorted(prs, key=lambda p: (p["state"] == "OPEN", p["number"]))[-1] + + +def check_runs(repo, sha): + out = call( + repo, "api", "--paginate", f"repos/{{owner}}/{{repo}}/commits/{sha}/check-runs", + "--jq", ".check_runs[] | {name, status, conclusion, id}", + ) + return out if isinstance(out, Unknown) else json_lines(out) + + +def statuses(repo, sha): + out = call( + repo, "api", "--paginate", f"repos/{{owner}}/{{repo}}/commits/{sha}/statuses", + "--jq", ".[] | {context, state, id}", + ) + return out if isinstance(out, Unknown) else json_lines(out) diff --git a/scripts/supervisor/observe.py b/scripts/supervisor/observe.py new file mode 100644 index 0000000..11e5505 --- /dev/null +++ b/scripts/supervisor/observe.py @@ -0,0 +1,208 @@ +"""Collect one read-only observation snapshot for decide(): ledger, Git, PRs/checks and processes.""" + +import json +import subprocess +import sys +import time +from pathlib import Path + +from . import gh, policy +from .decide import LIVE_SESSION, latest_review + +ROUND_PY = Path(__file__).resolve().parents[1] / "round.py" +RUNS = ".agent-runs" +SESSIONS = f"{RUNS}/supervisor/sessions" + + +class ObservationError(ValueError): + """Local state that cannot be read safely; the supervisor stops instead of guessing.""" + + +def git(repo, *args): + return subprocess.run( + ["git", "-C", str(repo), *args], text=True, capture_output=True, check=False + ) + + +def remote_head(repo, remote, branch): + done = git(repo, "ls-remote", "--exit-code", remote, f"refs/heads/{branch}") + if done.returncode or not done.stdout.strip(): + return None, f"git ls-remote {remote} {branch} failed: {done.stderr.strip()[:200]}" + return done.stdout.split()[0], None + + +def checkout(repo, remote_sha): + clean = not git(repo, "status", "--porcelain").stdout.strip() + on_default = None + if remote_sha and not git(repo, "cat-file", "-e", f"{remote_sha}^{{commit}}").returncode: + on_default = not git(repo, "merge-base", "--is-ancestor", "HEAD", remote_sha).returncode + return {"clean": clean, "head_on_default": on_default} + + +def round_order(name): + return (0, int(name[1:]), name) if name[:1] == "R" and name[1:].isdigit() else (1, 0, name) + + +def rounds(repo): + open_ids, closed, ids = [], [], [] + for path in sorted((repo / RUNS).glob("*/round.json"), key=lambda p: round_order(p.parent.name)): + try: + data = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError) as error: + raise ObservationError(f"Unreadable round record {path}: {error}") from None + ids.append(path.parent.name) + if data.get("state") == "closed": + closed.append({"id": path.parent.name, "delta": data.get("behavior_delta")}) + else: + open_ids.append(path.parent.name) + return {"open": open_ids, "closed": closed, "ids": ids} + + +def ledger(repo, round_id): + """Read the open round through round.py so the read takes the ledger lock.""" + done = subprocess.run( + [sys.executable, str(ROUND_PY), "--repo", str(repo), "status", round_id], + text=True, capture_output=True, check=False, + ) + if done.returncode: + raise ObservationError(f"round.py status {round_id}: {done.stderr.strip()}") + return json.loads(done.stdout) + + +def latest_results(repo, manifest): + found = {} + for aid, a in manifest["assignments"].items(): + if a["results"]: + path = repo / RUNS / manifest["id"] / a["results"][-1] + try: + found[aid] = json.loads(path.read_text()) + except (OSError, json.JSONDecodeError) as error: + raise ObservationError(f"Unreadable result {path}: {error}") from None + return found + + +def process(repo, record): + """A session process is alive only if its pid runs a command naming the session UUID.""" + uuid = record["session_uuid"] + exit_file = repo / SESSIONS / f"{uuid}.exit.json" + pid_file = repo / SESSIONS / f"{uuid}.pid" + pid = record.get("pid") + if pid is None and pid_file.exists(): + try: + pid = int(pid_file.read_text().strip()) + except ValueError: + pid = None + exit_record = None + if exit_file.exists(): + try: + exit_record = json.loads(exit_file.read_text()) + except json.JSONDecodeError: + exit_record = {"code": None, "invalid": True} + alive = False + if pid: + shown = subprocess.run( + ["ps", "-o", "args=", "-p", str(pid)], text=True, capture_output=True, check=False + ) + alive = shown.returncode == 0 and uuid in shown.stdout + return {"pid": pid, "alive": alive, "exit": exit_record} + + +def diff(repo, start, head): + names = git(repo, "diff", "--name-only", "--no-renames", start, head) + counts = git(repo, "diff", "--numstat", "--no-renames", start, head) + if names.returncode or counts.returncode: + return {"files": None, "changed_lines": 0, "binary": []} + total, binary = 0, [] + for line in counts.stdout.splitlines(): + added, removed, path = line.split("\t", 2) + if added == "-": + binary.append(path) + else: + total += int(added) + int(removed) + return {"files": names.stdout.splitlines(), "changed_lines": total, "binary": binary} + + +def required_checks(repo, gate, head): + runs = gh.check_runs(repo, head) + states = gh.statuses(repo, head) + for value in (runs, states): + if isinstance(value, gh.Unknown): + return {"unknown": value.reason} + state, detail = policy.required_checks_state(gate["required_checks"], runs, states) + return {"state": state, "detail": detail} + + +def worktree(path): + if not path.exists(): + return {"exists": False, "dirty": False} + done = git(path, "status", "--porcelain") + return {"exists": True, "dirty": bool(done.returncode or done.stdout.strip())} + + +def snapshot(repo, settings, autonomy, branch, new_round=False, rounds_this_run=0, now=None): + repo = Path(repo).resolve() + supervisor = settings["supervisor"] + merge_policy = settings["policy"]["merge"] + default_sha, unknown = remote_head(repo, supervisor["remote"], branch) + value = { + "now": now if now is not None else time.time(), + "autonomy": autonomy, + "new_round": new_round, + "rounds_this_run": rounds_this_run, + "default_branch": branch, + "limits": { + "max_parallel": supervisor["max_parallel"], + "launch_grace_seconds": supervisor["launch_grace_seconds"], + "sessions_per_round": settings["budgets"]["sessions_per_round"], + "round_usd": settings["budgets"]["round_usd"], + "max_rounds_per_run": settings["budgets"]["max_rounds_per_run"], + "ci_timeout_minutes": settings["budgets"]["ci_timeout_minutes"], + "max_repairs": settings["stop"]["max_repairs"], + "stall_rounds": settings["stop"]["stall_rounds"], + }, + "gate": { + "required_checks": merge_policy["required_checks"], + "protected_paths": merge_policy["protected_paths"], + "max_changed_lines": merge_policy["max_changed_lines"], + "allow_reconcile_merge": settings["policy"]["reconcile"]["allow_reconcile_merge"], + }, + "remote": {"default_sha": default_sha, "unknown": unknown}, + "checkout": checkout(repo, default_sha), + "rounds": rounds(repo), + "round": None, + "results": {}, + "processes": {}, + "prs": {}, + "checks": {}, + "diffs": {}, + "worktrees": {}, + "final_gate": None, + } + if len(value["rounds"]["open"]) != 1: + return value + manifest = ledger(repo, value["rounds"]["open"][0]) + value["round"] = manifest + if manifest["version"] != 2: + return value + value["results"] = latest_results(repo, manifest) + for name, record in manifest["sessions"].items(): + if record["state"] in LIVE_SESSION and record["role"] != "merger": + value["processes"][name] = process(repo, record) + trees = Path(supervisor["worktree_root"]) / manifest["id"] + for aid, a in manifest["assignments"].items(): + if a["state"] in {"merged", "cancelled"}: + value["worktrees"][aid] = worktree(trees / aid) + if a["state"] != "result-ready": + continue + pr = gh.pull_request(repo, a["branch"]) + value["prs"][aid] = {"unknown": pr.reason} if isinstance(pr, gh.Unknown) else {"pr": pr} + review = latest_review(a) + head = value["results"][aid]["head_sha"] + if autonomy >= 2 and review and review["verdict"] == "ready" and review["head_sha"] == head: + value["checks"][aid] = required_checks(repo, value["gate"], head) + value["diffs"][aid] = diff(repo, a["start_sha"] or manifest["base_sha"], head) + gate_file = repo / RUNS / "supervisor" / manifest["id"] / "final-gate.json" + if gate_file.exists(): + value["final_gate"] = json.loads(gate_file.read_text()).get("state") + return value + diff --git a/scripts/supervisor/policy.py b/scripts/supervisor/policy.py new file mode 100644 index 0000000..63e2ff8 --- /dev/null +++ b/scripts/supervisor/policy.py @@ -0,0 +1,124 @@ +"""Pure merge-policy helpers. No I/O: callers pass observations in and act on the verdict.""" + +import re + +PASSING_RUN = {"success"} +PENDING_RUN = {"queued", "in_progress", "waiting", "pending", "requested"} +STATUS_OUTCOME = {"success": "success", "pending": "pending", "failure": "failed", "error": "failed"} +MERGE_STATES = {"CLEAN", "HAS_HOOKS"} + + +def glob_regex(pattern): + """Translate a path glob: `**` spans directories, `*` and `?` stay within one segment.""" + out, i = [], 0 + while i < len(pattern): + if pattern.startswith("**/", i): + out.append("(?:.*/)?") + i += 3 + elif pattern.startswith("**", i): + out.append(".*") + i += 2 + elif pattern[i] == "*": + out.append("[^/]*") + i += 1 + elif pattern[i] == "?": + out.append("[^/]") + i += 1 + elif pattern[i] == "[" and "]" in pattern[i + 2 :]: + end = pattern.index("]", i + 2) + body = pattern[i + 1 : end].replace("\\", "\\\\") + out.append(f"[{'^' if body.startswith('!') else ''}{body.lstrip('!')}]") + i = end + 1 + else: + out.append(re.escape(pattern[i])) + i += 1 + return re.compile("".join(out) + r"\Z") + + +def glob_match(path, pattern): + if pattern.endswith("/**") and path == pattern[:-3]: + return False + return bool(glob_regex(pattern).match(path)) + + +def required_checks_state(required, runs, statuses): + """Return (passed|pending|failed, {name: outcome}) for the required names at one SHA. + + Check runs take precedence over legacy commit statuses of the same name; within a source the + newest (highest id) wins. Missing, skipped, neutral and cancelled count as failed. + """ + detail = {} + for name in required: + mine = sorted((r for r in runs if r["name"] == name), key=lambda r: r["id"]) + if mine: + newest = mine[-1] + if newest["status"] != "completed": + detail[name] = "pending" if newest["status"] in PENDING_RUN else "failed" + else: + detail[name] = "success" if newest["conclusion"] in PASSING_RUN else "failed" + continue + legacy = sorted((s for s in statuses if s["context"] == name), key=lambda s: s["id"]) + detail[name] = STATUS_OUTCOME.get(legacy[-1]["state"], "failed") if legacy else "missing" + if not required or any(v in {"failed", "missing"} for v in detail.values()): + return "failed", detail + if any(v == "pending" for v in detail.values()): + return "pending", detail + return "passed", detail + + +def merge_gate(facts, gate): + """Return (merge|wait|hold, reasons). Any hold reason wins over waiting.""" + holds, waits = [], [] + head = facts["result_head"] + review = facts["review"] + if not (review and review["verdict"] == "ready" and review["head_sha"] == head): + holds.append("no-ready-review") + pr = facts["pr"] + if pr["state"] != "OPEN": + holds.append("pr-not-open") + if pr["isDraft"]: + holds.append("pr-draft") + if pr["headRefOid"] != head: + holds.append("head-moved") + if pr["headRefName"] != facts["branch"]: + holds.append("wrong-head-branch") + if pr["baseRefName"] != facts["default_branch"]: + holds.append("wrong-base") + checks = facts["checks"] + if checks == "failed": + holds.append("checks-failed") + elif checks == "pending" and facts["checks_timed_out"]: + holds.append("checks-timeout") + elif checks == "pending": + waits.append("checks-pending") + state = pr["mergeStateStatus"] + if pr["mergeable"] == "CONFLICTING" or state == "DIRTY": + holds.append("conflicting") + elif state == "BEHIND": + holds.append("behind") + elif pr["mergeable"] == "UNKNOWN" or state == "UNKNOWN": + waits.append("mergeability-unknown") + elif state not in MERGE_STATES and not (state == "BLOCKED" and checks == "pending"): + holds.append(f"merge-state-{state}") + files = facts["files"] + if files is None: + holds.append("diff-unavailable") + elif not files: + holds.append("empty-diff") + else: + for path in files: + if not any(glob_match(path, p) for p in facts["owned_paths"]): + holds.append(f"outside-owned:{path}") + if any(glob_match(path, p) for p in gate["protected_paths"]): + holds.append(f"protected:{path}") + if facts["changed_lines"] > gate["max_changed_lines"]: + holds.append("too-large") + # Binary changes have no line count, so they cannot be checked against the size limit. + holds += [f"binary:{path}" for path in facts.get("binary", [])] + if facts["kind"] == "reconcile" and not gate["allow_reconcile_merge"]: + holds.append("reconcile-merge-not-granted") + if holds: + return "hold", holds + if waits: + return "wait", waits + return "merge", [] diff --git a/tests/fakes/gh b/tests/fakes/gh new file mode 100755 index 0000000..f4ba17c --- /dev/null +++ b/tests/fakes/gh @@ -0,0 +1,35 @@ +#!/usr/bin/env python3 +"""Offline stand-in for the GitHub CLI used by supervisor tests. + +State comes from the JSON file named by FAKE_GH_STATE: + {"prs": {branch: [pr, ...]}, "check_runs": {sha: [...]}, "statuses": {sha: [...]}, + "fail": ["substring of the joined argv", ...]} +Every call is appended to FAKE_GH_LOG as one JSON argv line. Unknown calls fail loudly. +""" + +import json +import os +import sys + +argv = sys.argv[1:] +with open(os.environ["FAKE_GH_LOG"], "a") as log: + log.write(json.dumps(argv) + "\n") +with open(os.environ["FAKE_GH_STATE"]) as handle: + state = json.load(handle) +joined = " ".join(argv) +if any(fragment in joined for fragment in state.get("fail", [])): + print("fake gh: injected failure", file=sys.stderr) + sys.exit(1) + +if argv[:2] == ["pr", "list"]: + branch = argv[argv.index("--head") + 1] + print(json.dumps(state.get("prs", {}).get(branch, []))) +elif argv[:1] == ["api"]: + path = next(a for a in argv[1:] if a.startswith("repos/")) + sha = path.split("/commits/")[1].split("/")[0] + key = "check_runs" if path.endswith("/check-runs") else "statuses" + for item in state.get(key, {}).get(sha, []): + print(json.dumps(item)) +else: + print(f"fake gh: unsupported call {argv}", file=sys.stderr) + sys.exit(2) diff --git a/tests/test_supervisor_cli.py b/tests/test_supervisor_cli.py new file mode 100644 index 0000000..184d543 --- /dev/null +++ b/tests/test_supervisor_cli.py @@ -0,0 +1,255 @@ +"""End-to-end checks of the read-only supervisor CLI against real Git and a fake gh.""" + +import fcntl +import json +import os +import subprocess +import sys +import tempfile +import unittest +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SUPERVISE = ROOT / "scripts" / "supervise.py" +ROUND = ROOT / "scripts" / "round.py" +FAKES = ROOT / "tests" / "fakes" +POLICY = "[supervisor]\nautonomy = 1\n[policy.merge]\nrequired_checks = ['check']\n" + + +class SupervisorCliTests(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.tmp = Path(tmp.name) + self.origin = self.tmp / "origin.git" + self.repo = self.tmp / "project" + subprocess.run(["git", "init", "-q", "--bare", "-b", "main", self.origin], check=True) + subprocess.run(["git", "clone", "-q", self.origin, self.repo], check=True, + capture_output=True) + self.git("config", "user.name", "Test") + self.git("config", "user.email", "test@example.invalid") + (self.repo / ".gitignore").write_text(".agent-runs/\nHANDOFF.md\n") + (self.repo / ".agent").mkdir() + (self.repo / ".agent/supervisor.toml").write_text(POLICY) + self.git("add", ".") + self.git("commit", "-qm", "base") + self.git("push", "-q", "origin", "main") + self.base = self.git("rev-parse", "HEAD") + self.state = self.tmp / "gh-state.json" + self.log = self.tmp / "gh-log.jsonl" + self.gh_state({}) + + def git(self, *args, cwd=None): + return subprocess.run( + ["git", "-C", str(cwd or self.repo), *args], + text=True, capture_output=True, check=True, + ).stdout.strip() + + def gh_state(self, value): + self.state.write_text(json.dumps(value)) + + def gh_calls(self): + if not self.log.exists(): + return [] + return [json.loads(line) for line in self.log.read_text().splitlines()] + + def env(self): + return { + **os.environ, + "PATH": f"{FAKES}{os.pathsep}{os.environ['PATH']}", + "FAKE_GH_STATE": str(self.state), + "FAKE_GH_LOG": str(self.log), + } + + def supervise(self, *args, code=0): + done = subprocess.run( + [sys.executable, str(SUPERVISE), "--repo", str(self.repo), *args], + text=True, capture_output=True, env=self.env(), check=False, + ) + self.assertEqual(done.returncode, code, done.stdout + done.stderr) + return done + + def next_action(self, *args, code=0): + return json.loads(self.supervise("next", *args, code=code).stdout) + + def ledger(self, *args): + done = subprocess.run( + [sys.executable, str(ROUND), "--repo", str(self.repo), *args], + text=True, capture_output=True, check=False, + ) + self.assertEqual(done.returncode, 0, done.stderr) + return done.stdout + + def start_round(self): + spec = self.repo / ".agent-runs" / "spec.md" + spec.parent.mkdir(exist_ok=True) + spec.write_text("Document the behavior.\n") + self.ledger("init", "R1", "--base", "main", "--supervised") + self.ledger("assign", "R1", "A1", "--item", "TASK-1", "--branch", "task/a1", + "--spec", str(spec), "--owns", "docs/**") + + def submit_result(self): + self.git("switch", "-q", "-c", "task/a1") + (self.repo / "docs").mkdir() + (self.repo / "docs/a.md").write_text("Behavior\n") + self.git("add", "docs/a.md") + self.git("commit", "-qm", "docs") + head = self.git("rev-parse", "HEAD") + self.git("push", "-q", "origin", "task/a1") + self.git("switch", "-q", "main") + manifest = json.loads(self.ledger("status", "R1")) + result = self.repo / ".agent-runs" / "result.json" + result.write_text(json.dumps({ + "assignment": "A1", + "base_sha": self.base, + "packet_sha256": manifest["assignments"]["A1"]["packet_sha256"], + "head_sha": head, + "pr": "https://example.invalid/pull/1", + "checks": [{"command": ["true"], "status": "passed", "evidence": "ok"}], + })) + self.ledger("record", "R1", "--result", str(result)) + return head + + def pr(self, head, **changes): + return { + "number": 1, "state": "OPEN", "isDraft": False, "headRefOid": head, + "headRefName": "task/a1", "baseRefName": "main", "mergeable": "MERGEABLE", + "mergeStateStatus": "CLEAN", "url": "https://example.invalid/pull/1", **changes, + } + + def test_no_round_parks_at_l1_and_new_round_proposes_init(self): + self.assertEqual(self.next_action()["reason"], "no-round") + action = self.next_action("--new-round") + self.assertEqual(action, {"kind": "InitRound", "exit": 0, "round_id": "R1", "base": self.base}) + self.assertFalse((self.repo / ".agent-runs").exists()) + + def test_result_flow_reads_pr_state_and_never_writes(self): + self.start_round() + self.assertEqual(self.next_action()["kind"], "LaunchExecutor") + head = self.submit_result() + self.gh_state({"prs": {"task/a1": [self.pr(head)]}}) + before = self.ledger("status", "R1") + self.assertEqual(self.next_action(), { + "kind": "LaunchReviewer", "exit": 0, "assignment": "A1", "head": head + }) + self.ledger("review", "R1", "A1", "--head", head, "--verdict", "ready", + "--evidence", "manual", "--reviewer", "operator") + self.assertEqual(self.next_action()["reason"], "awaiting-merge") + self.gh_state({"prs": {"task/a1": [self.pr("f" * 40)]}}) + held = self.next_action(code=10) + self.assertEqual(held["holds"], [{"assignment": "A1", "reason": "head-moved"}]) + self.assertEqual(json.loads(self.ledger("status", "R1"))["assignments"]["A1"]["state"], + json.loads(before)["assignments"]["A1"]["state"]) + for call in self.gh_calls(): + self.assertIn(call[:2], (["pr", "list"], ["api", "--paginate"]), call) + + def test_l2_gate_reads_required_checks_on_the_exact_head(self): + (self.repo / ".agent/supervisor.toml").write_text( + POLICY.replace("autonomy = 1", "autonomy = 2") + "auto_merge = true\n" + ) + self.git("commit", "-qam", "grant L2") + self.git("push", "-q", "origin", "main") + self.base = self.git("rev-parse", "HEAD") + self.start_round() + head = self.submit_result() + self.ledger("review", "R1", "A1", "--head", head, "--verdict", "ready", + "--evidence", "manual", "--reviewer", "operator") + run = {"name": "check", "status": "completed", "conclusion": "success", "id": 5} + self.gh_state({"prs": {"task/a1": [self.pr(head)]}, "check_runs": {head: [run]}}) + self.assertEqual(self.next_action(), {"kind": "Merge", "exit": 0, "assignment": "A1", "head": head}) + self.assertIn(f"repos/{{owner}}/{{repo}}/commits/{head}/check-runs", sum(self.gh_calls(), [])) + self.gh_state({"prs": {"task/a1": [self.pr(head)]}}) + self.assertEqual(self.next_action(code=10)["holds"][0]["reason"], "checks-failed") + self.gh_state({"prs": {"task/a1": [self.pr(head)]}, "fail": ["check-runs"]}) + self.assertEqual(self.next_action(code=11)["kind"], "Unknown") + + def test_gh_failure_is_unknown_not_success(self): + self.start_round() + self.submit_result() + self.gh_state({"fail": ["pr list"]}) + action = self.next_action(code=11) + self.assertEqual(action["kind"], "Unknown") + + def test_dirty_or_diverged_checkout_stops(self): + (self.repo / "README.md").write_text("dirty\n") + self.assertEqual(self.next_action(code=10)["reason"], "dirty-checkout") + self.git("add", "README.md") + self.git("commit", "-qm", "local only") + self.assertEqual(self.next_action(code=10)["reason"], "diverged-checkout") + + def test_unfetched_remote_head_asks_for_fetch(self): + other = self.tmp / "other" + subprocess.run(["git", "clone", "-q", self.origin, other], check=True, capture_output=True) + (other / "x").write_text("x\n") + self.git("add", "x", cwd=other) + self.git("-c", "user.name=T", "-c", "user.email=t@example.invalid", "commit", "-qm", "x", cwd=other) + self.git("push", "-q", "origin", "main", cwd=other) + self.assertEqual(self.next_action()["kind"], "Fetch") + + def test_session_processes_are_observed_by_uuid(self): + self.start_round() + session_uuid = str(uuid.uuid4()) + deadline = (datetime.now(timezone.utc) + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ") + self.ledger("session", "start", "R1", "--role", "executor", "--assignment", "A1", + "--runner", "claude", "--session-uuid", session_uuid, "--cwd", str(self.tmp), + "--deadline", deadline) + child = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(60)", session_uuid] + ) + self.addCleanup(child.wait) + self.addCleanup(child.kill) + sessions = self.repo / ".agent-runs/supervisor/sessions" + sessions.mkdir(parents=True) + (sessions / f"{session_uuid}.pid").write_text(f"{child.pid}\n") + self.assertEqual(self.next_action(), {"kind": "AckSession", "exit": 0, "session": "S1", "pid": child.pid}) + self.ledger("session", "ack", "R1", "S1", "--pid", str(os.getpid())) + self.assertEqual(self.next_action()["kind"], "MarkLost") + (sessions / f"{session_uuid}.exit.json").write_text('{"code": 0}\n') + self.assertEqual(self.next_action()["kind"], "ImportSession") + + def test_halt_parks_with_attention(self): + self.start_round() + self.ledger("halt", "R1", "--code", "test", "--detail", "inspect", "--action", "look") + self.assertEqual(self.next_action(code=10), { + "kind": "Park", "exit": 10, "reason": "halted", "halts": ["H1"] + }) + + def test_status_explains_the_round(self): + self.start_round() + out = self.supervise("status").stdout + self.assertIn("Open round: R1 v2", out) + self.assertIn("A1 assigned task/a1", out) + self.assertIn("Next: LaunchExecutor", out) + + def test_configuration_errors_exit_2(self): + (self.repo / ".agent/supervisor.toml").write_text("[supervisor]\nautonomy = 2\n") + self.git("commit", "-qam", "unsafe") + self.git("push", "-q", "origin", "main") + done = self.supervise("next", code=2) + self.assertIn("auto_merge", done.stderr) + self.supervise("--autonomy", "1", "next", code=2) + + def test_run_is_dry_only_logged_and_single_instance(self): + self.supervise("run", "--once", code=2) + done = self.supervise("run", "--dry-run", "--once") + self.assertEqual(json.loads(done.stdout)["reason"], "no-round") + state = self.repo / ".agent-runs/supervisor" + entries = [json.loads(line) for line in (state / "log.jsonl").read_text().splitlines()] + self.assertEqual(entries[-1]["action"]["reason"], "no-round") + self.assertTrue(entries[-1]["dry_run"]) + self.assertEqual(json.loads((state / "last.json").read_text())["action"]["kind"], "Park") + with (state / "lock").open("a") as handle: + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + self.supervise("run", "--dry-run", "--once", code=3) + self.assertEqual(self.git("status", "--porcelain"), "") + + def test_effectful_decision_ends_a_dry_run(self): + self.start_round() + done = self.supervise("run", "--dry-run") + self.assertEqual(json.loads(done.stdout)["kind"], "LaunchExecutor") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_supervisor_config.py b/tests/test_supervisor_config.py new file mode 100644 index 0000000..28a1ca5 --- /dev/null +++ b/tests/test_supervisor_config.py @@ -0,0 +1,155 @@ +"""Supervisor configuration: validation, committed-policy source and autonomy grants.""" + +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + +from supervisor import config # noqa: E402 + +EXAMPLE = (ROOT / ".agent/supervisor.example.toml").read_text() + +L2 = """ +[supervisor] +autonomy = 2 +[policy.merge] +auto_merge = true +required_checks = ["check"] +""" + +L3 = L2.replace("autonomy = 2", "autonomy = 3") + """ +[policy.reconcile] +auto_reconcile = true +allow_reconcile_merge = true +""" + + +class ParseTests(unittest.TestCase): + def test_example_is_valid_and_conservative(self): + settings = config.parse(EXAMPLE) + self.assertEqual(settings["supervisor"]["autonomy"], 1) + self.assertFalse(settings["policy"]["merge"]["auto_merge"]) + self.assertEqual(config.effective_autonomy(settings), 1) + self.assertIn("planner", settings["runner"]["roles"]) + + def test_defaults_fill_an_empty_file(self): + settings = config.parse("") + self.assertEqual(settings["supervisor"]["max_parallel"], 1) + self.assertEqual(settings["stop"]["max_repairs"], 2) + + def test_rejections_name_the_problem(self): + cases = [ + ("[supervisor]\nautonmy = 1\n", "supervisor.autonmy"), + ("[surpervisor]\n", "surpervisor"), + ("[supervisor]\nautonomy = 4\n", "supervisor.autonomy"), + ("[supervisor]\nmax_parallel = 0\n", "supervisor.max_parallel"), + ("[supervisor]\nmax_parallel = true\n", "supervisor.max_parallel"), + ("[budgets]\nround_usd = -1\n", "budgets.round_usd"), + ("[policy.merge]\nauto_merge = 1\n", "policy.merge.auto_merge"), + ("[policy.merge]\nmethod = 'fast'\n", "policy.merge.method"), + ("[policy.merge]\nrequired_checks = 'check'\n", "policy.merge.required_checks"), + ("[notify]\ncommand = [1]\n", "notify.command"), + ("[runner.commands.x]\nargv = []\n", "runner.commands.x.argv"), + ("[runner.roles.boss]\nmodel = 'x'\n", "runner.roles.boss"), + ("[runner.roles.planner]\ntimeout_minutes = 0\n", "timeout_minutes"), + ("not toml [", "TOML"), + ] + for text, fragment in cases: + with self.subTest(fragment=fragment): + with self.assertRaises(config.ConfigError) as caught: + config.parse(text) + self.assertIn(fragment, str(caught.exception)) + + def test_local_override_is_limited_to_runner_and_budgets(self): + settings = config.parse(EXAMPLE, "[budgets]\nround_usd = 3.5\n") + self.assertEqual(settings["budgets"]["round_usd"], 3.5) + settings = config.parse(EXAMPLE, "[runner.roles.executor]\nmodel = 'local'\n") + self.assertEqual(settings["runner"]["roles"]["executor"]["model"], "local") + self.assertEqual(settings["runner"]["roles"]["planner"]["model"], "opus") + for local in ("[policy.merge]\nauto_merge = true\n", "[supervisor]\nautonomy = 3\n"): + with self.subTest(local=local): + with self.assertRaises(config.ConfigError) as caught: + config.parse(EXAMPLE, local) + self.assertIn("local", str(caught.exception)) + + +class AutonomyTests(unittest.TestCase): + def test_grants_are_required(self): + with self.assertRaises(config.ConfigError) as caught: + config.effective_autonomy(config.parse("[supervisor]\nautonomy = 2\n")) + self.assertIn("auto_merge", str(caught.exception)) + missing_checks = L2.replace('required_checks = ["check"]', "required_checks = []") + with self.assertRaises(config.ConfigError) as caught: + config.effective_autonomy(config.parse(missing_checks)) + self.assertIn("required_checks", str(caught.exception)) + with self.assertRaises(config.ConfigError) as caught: + config.effective_autonomy(config.parse(L2.replace("autonomy = 2", "autonomy = 3"))) + self.assertIn("auto_reconcile", str(caught.exception)) + self.assertEqual(config.effective_autonomy(config.parse(L2)), 2) + self.assertEqual(config.effective_autonomy(config.parse(L3)), 3) + + def test_ceiling_can_lower_but_never_raise(self): + self.assertEqual(config.effective_autonomy(config.parse(L3), ceiling=1), 1) + self.assertEqual(config.effective_autonomy(config.parse(EXAMPLE), ceiling=3), 1) + + +class LoadTests(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.origin = Path(tmp.name) / "origin.git" + self.repo = Path(tmp.name) / "project" + subprocess.run(["git", "init", "-q", "--bare", "-b", "main", self.origin], check=True) + subprocess.run(["git", "clone", "-q", self.origin, self.repo], check=True, + capture_output=True) + self.git("config", "user.name", "Test") + self.git("config", "user.email", "test@example.invalid") + (self.repo / ".gitignore").write_text(".agent-runs/\n.agent/supervisor.local.toml\n") + self.settings = self.repo / ".agent/supervisor.toml" + self.settings.parent.mkdir() + self.settings.write_text("[supervisor]\nautonomy = 1\n") + self.git("add", ".") + self.git("commit", "-qm", "policy") + self.git("push", "-q", "origin", "main") + + def git(self, *args): + subprocess.run(["git", "-C", self.repo, *args], check=True, capture_output=True) + + def test_policy_comes_from_the_committed_remote_branch(self): + self.settings.write_text(L3) + settings = config.load(self.repo, "origin", "main") + self.assertEqual(settings["supervisor"]["autonomy"], 1) + self.assertEqual( + Path(settings["supervisor"]["worktree_root"]), + self.repo.parent / "project-worktrees", + ) + + def test_missing_committed_policy_fails(self): + with self.assertRaises(config.ConfigError) as caught: + config.load(self.repo, "origin", "develop") + self.assertIn("origin/develop", str(caught.exception)) + + def test_local_override_must_be_ignored(self): + local = self.repo / ".agent/supervisor.local.toml" + local.write_text("[budgets]\nround_usd = 1.0\n") + self.assertEqual(config.load(self.repo, "origin", "main")["budgets"]["round_usd"], 1.0) + (self.repo / ".gitignore").write_text(".agent-runs/\n") + with self.assertRaises(config.ConfigError) as caught: + config.load(self.repo, "origin", "main") + self.assertIn("ignored", str(caught.exception)) + + def test_worktree_root_must_be_outside_the_repository(self): + self.settings.write_text("[supervisor]\nworktree_root = 'trees'\n") + self.git("commit", "-qam", "inside") + self.git("push", "-q", "origin", "main") + with self.assertRaises(config.ConfigError) as caught: + config.load(self.repo, "origin", "main") + self.assertIn("worktree_root", str(caught.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_supervisor_decide.py b/tests/test_supervisor_decide.py new file mode 100644 index 0000000..3f97860 --- /dev/null +++ b/tests/test_supervisor_decide.py @@ -0,0 +1,417 @@ +"""Table-driven checks of the supervisor's pure decision function.""" + +import copy +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +from supervisor.decide import decide # noqa: E402 + +NOW = 1_800_000_000 # 2027-01-15T08:00:00Z +BASE = "b" * 40 +HEAD = "c" * 40 +HEAD2 = "d" * 40 +REMOTE = "e" * 40 + + +def utc(offset): + from datetime import datetime, timezone + + return datetime.fromtimestamp(NOW + offset, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def assignment(state="assigned", kind="work", results=0, reviews=(), branch=None, aid="A1"): + paths = [f"results/{aid}-{n}.json" for n in range(1, results + 1)] + return { + "item": f"TASK-{aid}", + "branch": branch or f"task/{aid.lower()}", + "packet": f"prompts/{aid}.md", + "packet_sha256": "f" * 64, + "state": state, + "results": paths, + "kind": kind, + "owned_paths": ["docs/**"], + "resources": [], + "start_sha": None, + "reviews": [ + {"result": paths[-1], "head_sha": head, "verdict": verdict, "utc": utc(-600)} + for verdict, head in reviews + ], + } + + +def session(role, state="running", assignment="A1", started=-60, deadline=3600, cost=None): + return { + "role": role, + "assignment": None if role == "planner" else assignment, + "session_uuid": f"uuid-{role}", + "cwd": "/tmp/x", + "head_sha": None, + "pid": 42 if state != "launching" else None, + "started_utc": utc(started), + "deadline_utc": utc(deadline), + "state": state, + "cost_usd": cost, + } + + +def open_pr(head=HEAD, **changes): + return { + "state": "OPEN", + "isDraft": False, + "headRefOid": head, + "headRefName": "task/a1", + "baseRefName": "main", + "mergeable": "MERGEABLE", + "mergeStateStatus": "CLEAN", + **changes, + } + + +def snapshot(assignments=None, sessions=None, halts=(), **changes): + manifest = None + if assignments is not None: + manifest = { + "version": 2, + "id": "R1", + "base_sha": BASE, + "state": "open", + "assignments": assignments, + "sessions": sessions or {}, + "halts": list(halts), + } + value = { + "now": NOW, + "autonomy": 1, + "new_round": False, + "rounds_this_run": 0, + "default_branch": "main", + "limits": { + "max_parallel": 1, + "sessions_per_round": 8, + "round_usd": 10.0, + "max_repairs": 2, + "stall_rounds": 2, + "max_rounds_per_run": 1, + "launch_grace_seconds": 120, + "ci_timeout_minutes": 60, + }, + "gate": { + "required_checks": ["check"], + "protected_paths": [".github/**"], + "max_changed_lines": 400, + "allow_reconcile_merge": False, + }, + "remote": {"default_sha": REMOTE, "unknown": None}, + "checkout": {"clean": True, "head_on_default": True}, + "rounds": {"open": ["R1"] if manifest else [], "closed": [], "ids": ["R1"] if manifest else []}, + "round": manifest, + "results": {}, + "processes": {}, + "prs": {}, + "checks": {}, + "diffs": {}, + "worktrees": {}, + "final_gate": None, + } + for key, item in changes.items(): + value[key] = item + if manifest: + for aid, a in assignments.items(): + if a["results"] and aid not in value["results"]: + value["results"][aid] = {"status": "result-ready", "head_sha": HEAD} + return value + + +def ready_for_merge(**changes): + value = snapshot( + {"A1": assignment("result-ready", results=1, reviews=[("ready", HEAD)])}, + autonomy=2, + prs={"A1": {"pr": open_pr()}}, + checks={"A1": {"state": "passed"}}, + diffs={"A1": {"files": ["docs/a.md"], "changed_lines": 3}}, + ) + value.update(changes) + return value + + +class DecideTests(unittest.TestCase): + def check(self, value, kind, exit=None, **fields): + action = decide(copy.deepcopy(value)) + self.assertEqual(action["kind"], kind, action) + if exit is not None: + self.assertEqual(action["exit"], exit, action) + for key, expected in fields.items(): + self.assertEqual(action.get(key), expected, action) + return action + + def test_guards(self): + manual = snapshot({}) + manual["round"]["version"] = 1 + two_open = snapshot({}, rounds={"open": ["R1", "R2"], "closed": [], "ids": ["R1", "R2"]}) + halted = snapshot( + {"A1": assignment()}, halts=[{"id": "H1", "cleared_utc": None, "code": "lost"}] + ) + cleared = snapshot( + {"A1": assignment()}, halts=[{"id": "H1", "cleared_utc": utc(-5), "code": "lost"}] + ) + cases = [ + ("manual v1 round", manual, "Park", 0, {"reason": "manual-round"}), + ("two open rounds", two_open, "Stop", 10, {"reason": "multiple-open-rounds"}), + ("uncleared halt", halted, "Park", 10, {"reason": "halted"}), + ("cleared halt", cleared, "LaunchExecutor", 0, {}), + ( + "dirty checkout", + snapshot({"A1": assignment()}, checkout={"clean": False, "head_on_default": True}), + "Stop", 10, {"reason": "dirty-checkout"}, + ), + ( + "diverged checkout", + snapshot(checkout={"clean": True, "head_on_default": False}), + "Stop", 10, {"reason": "diverged-checkout"}, + ), + ( + "remote object missing", + snapshot(checkout={"clean": True, "head_on_default": None}), + "Fetch", 0, {}, + ), + ( + "remote unknown", + snapshot(remote={"default_sha": None, "unknown": "ls-remote failed"}), + "Unknown", 11, {}, + ), + ] + for name, value, kind, exit, fields in cases: + with self.subTest(name): + self.check(value, kind, exit, **fields) + + def test_session_recovery(self): + def with_session(record, process): + return snapshot( + {"A1": assignment()}, + sessions={"S1": record}, + processes={"S1": process} if process is not None else {}, + ) + + alive = {"pid": 42, "alive": True, "exit": None} + dead = {"pid": 42, "alive": False, "exit": None} + exited = {"pid": 42, "alive": False, "exit": {"code": 0}} + cases = [ + ("launching with process", session("executor", "launching"), alive, "AckSession"), + ("launching within grace", session("executor", "launching"), {"pid": None, "alive": False, "exit": None}, "Wait"), + ("launching past grace", session("executor", "launching", started=-600), None, "MarkLost"), + ("running and alive", session("executor"), alive, "Wait"), + ("running but dead", session("executor"), dead, "MarkLost"), + ("past deadline", session("executor", deadline=-1), alive, "KillSession"), + ("exited not ended", session("executor"), exited, "ImportSession"), + ("exited while launching", session("executor", "launching"), exited, "ImportSession"), + ("merger in flight", session("merger"), None, "VerifyMerge"), + ] + for name, record, process, kind in cases: + with self.subTest(name): + self.check(with_session(record, process), kind, session="S1" if kind != "Wait" else None) + + def test_containment_still_runs_while_halted(self): + halts = [{"id": "H1", "cleared_utc": None, "code": "x"}] + value = snapshot( + {"A1": assignment()}, + sessions={"S1": session("executor", deadline=-1)}, + processes={"S1": {"pid": 42, "alive": True, "exit": None}}, + halts=halts, + ) + self.check(value, "KillSession", session="S1") + value["processes"]["S1"]["exit"] = {"code": 0} + value["round"]["sessions"]["S1"]["deadline_utc"] = utc(60) + self.check(value, "Park", 10, reason="halted") + + def test_kill_before_import_across_sessions(self): + value = snapshot( + {"A1": assignment(), "A2": assignment(aid="A2")}, + sessions={ + "S1": session("executor"), + "S2": session("executor", assignment="A2", deadline=-5), + }, + processes={ + "S1": {"pid": 1, "alive": False, "exit": {"code": 0}}, + "S2": {"pid": 2, "alive": True, "exit": None}, + }, + limits={**snapshot()["limits"], "max_parallel": 2}, + ) + self.check(value, "KillSession", session="S2") + + def test_no_open_round(self): + def closed(*deltas): + ids = [f"R{n}" for n in range(1, len(deltas) + 1)] + return { + "open": [], + "closed": [{"id": i, "delta": d} for i, d in zip(ids, deltas)], + "ids": ids, + } + + cases = [ + ("L1 parks", snapshot(), "Park", 0, {"reason": "no-round"}), + ("--new-round starts", snapshot(new_round=True), "InitRound", 0, {"round_id": "R1", "base": REMOTE}), + ("L3 starts", snapshot(autonomy=3, rounds=closed("changed")), "InitRound", 0, {"round_id": "R2"}), + ("stall", snapshot(autonomy=3, rounds=closed("changed", "none", "none")), "Park", 10, {"reason": "stalled"}), + ("one none is not a stall", snapshot(autonomy=3, rounds=closed("none")), "InitRound", 0, {}), + ("max rounds", snapshot(autonomy=3, rounds_this_run=1), "Park", 0, {"reason": "max-rounds"}), + ] + for name, value, kind, exit, fields in cases: + with self.subTest(name): + self.check(value, kind, exit, **fields) + + def test_planning(self): + planner_live = snapshot( + {}, sessions={"S1": session("planner")}, + processes={"S1": {"pid": 42, "alive": True, "exit": None}}, + ) + planner_done = snapshot({}, sessions={"S1": session("planner", "exited")}) + cases = [ + ("launch planner", snapshot({}), "LaunchPlanner", 0, {}), + ("planner running", planner_live, "Wait", 0, {}), + ("planner produced nothing", planner_done, "Hold", 10, {}), + ] + for name, value, kind, exit, fields in cases: + with self.subTest(name): + self.check(value, kind, exit, **fields) + + def test_assignment_table(self): + def one(a, **changes): + return snapshot({"A1": a}, **{"prs": {"A1": {"pr": open_pr()}}, **changes}) + + exhausted = assignment("result-ready", results=3, reviews=[("needs-fix", HEAD)]) + cases = [ + ("assigned", one(assignment()), "LaunchExecutor", {"repair": False}), + ("blocked result", one(assignment("blocked", results=1)), "Hold", {}), + ("needs review", one(assignment("result-ready", results=1)), "LaunchReviewer", {"head": HEAD}), + ( + "needs fix", + one(assignment("result-ready", results=1, reviews=[("needs-fix", HEAD)])), + "LaunchExecutor", {"repair": True}, + ), + ("repairs exhausted", one(exhausted), "Hold", {}), + ( + "review blocked", + one(assignment("result-ready", results=1, reviews=[("blocked", HEAD)])), + "Hold", {}, + ), + ( + "ready at L1", + one(assignment("result-ready", results=1, reviews=[("ready", HEAD)])), + "Park", {"reason": "awaiting-merge", "exit": 0}, + ), + ( + "PR missing", + one(assignment("result-ready", results=1), prs={"A1": {"pr": None}}), + "Hold", {}, + ), + ( + "PR closed", + one(assignment("result-ready", results=1), prs={"A1": {"pr": open_pr(state="CLOSED")}}), + "Hold", {}, + ), + ( + "merged externally", + one(assignment("result-ready", results=1), prs={"A1": {"pr": open_pr(state="MERGED")}}), + "Hold", {}, + ), + ( + "head moved", + one(assignment("result-ready", results=1), prs={"A1": {"pr": open_pr(HEAD2)}}), + "Hold", {}, + ), + ( + "PR unknown", + one(assignment("result-ready", results=1), prs={"A1": {"unknown": "gh failed"}}), + "Unknown", {"exit": 11}, + ), + ("terminal with worktree", one(assignment("merged", results=1), worktrees={"A1": {"exists": True, "dirty": False}}), "Cleanup", {}), + ("terminal dirty worktree", one(assignment("cancelled"), worktrees={"A1": {"exists": True, "dirty": True}}), "Hold", {}), + ] + for name, value, kind, fields in cases: + with self.subTest(name): + self.check(value, kind, **fields) + + def test_holds_name_the_assignment_and_reason(self): + value = snapshot({"A1": assignment("blocked", results=1)}) + action = self.check(value, "Hold", 10) + self.assertEqual(action["holds"], [{"assignment": "A1", "reason": "result-blocked"}]) + + def test_merge_gate_wiring(self): + self.check(ready_for_merge(), "Merge", 0, assignment="A1", head=HEAD) + self.check(ready_for_merge(checks={"A1": {"state": "pending"}}), "Wait", 0) + self.check(ready_for_merge(checks={"A1": {"state": "failed"}}), "Hold", 10) + self.check(ready_for_merge(checks={"A1": {"unknown": "api"}}), "Unknown", 11) + value = ready_for_merge(checks={"A1": {"state": "pending"}}) + value["round"]["assignments"]["A1"]["reviews"][0]["utc"] = utc(-3601 - 60 * 60) + self.check(value, "Hold", 10) + self.check(ready_for_merge(diffs={"A1": {"files": ["src/x.py"], "changed_lines": 1}}), "Hold") + + def test_old_head_review_is_ignored(self): + value = snapshot( + {"A1": assignment("result-ready", results=2)}, + prs={"A1": {"pr": open_pr()}}, + ) + value["round"]["assignments"]["A1"]["reviews"] = [ + {"result": "results/A1-1.json", "head_sha": HEAD2, "verdict": "ready", "utc": utc(-60)} + ] + self.check(value, "LaunchReviewer") + + def test_priority_and_parallelism(self): + assignments = { + "A1": assignment(), + "A2": assignment("result-ready", results=1, aid="A2"), + "A3": assignment("merged", results=1, aid="A3"), + } + prs = {"A2": {"pr": open_pr(headRefName="task/a2")}} + worktrees = {"A3": {"exists": True, "dirty": False}} + value = snapshot(assignments, prs=prs, worktrees=worktrees) + self.check(value, "LaunchReviewer", assignment="A2") + value["round"]["sessions"] = {"S1": session("reviewer", assignment="A2")} + value["processes"] = {"S1": {"pid": 1, "alive": True, "exit": None}} + self.check(value, "Cleanup", assignment="A3") + value["limits"]["max_parallel"] = 2 + self.check(value, "LaunchExecutor", assignment="A1") + + def test_hold_on_one_assignment_does_not_block_another(self): + value = snapshot( + {"A1": assignment("blocked", results=1), "A2": assignment(aid="A2")} + ) + action = self.check(value, "LaunchExecutor", assignment="A2") + self.assertEqual(action["holds"][0]["assignment"], "A1") + + def test_session_budget_parks(self): + sessions = {f"S{n}": session("executor", "exited", assignment="A9") for n in range(1, 9)} + self.check(snapshot({"A1": assignment()}, sessions=sessions), "Park", 10, reason="budget") + + def test_round_end(self): + merged = assignment("merged", results=1) + reconcile = assignment("merged", kind="reconcile", results=1, aid="A2") + cases = [ + ("L1 awaits reconcile", snapshot({"A1": merged}), "Park", {"reason": "awaiting-reconcile"}), + ("L3 creates reconcile", snapshot({"A1": merged}, autonomy=3), "CreateReconcile", {"start": REMOTE}), + ("all cancelled still reconciles", snapshot({"A1": assignment("cancelled")}, autonomy=3), "CreateReconcile", {}), + ("final gate", snapshot({"A1": merged, "A2": reconcile}, autonomy=3), "RunFinalGate", {}), + ("close", snapshot({"A1": merged, "A2": reconcile}, autonomy=3, final_gate="passed"), "CloseRound", {}), + ("red gate", snapshot({"A1": merged, "A2": reconcile}, autonomy=3, final_gate="failed"), "Hold", {}), + ( + "cancelled reconcile", + snapshot({"A1": merged, "A2": assignment("cancelled", kind="reconcile", aid="A2")}, autonomy=3), + "Hold", {}, + ), + ] + for name, value, kind, fields in cases: + with self.subTest(name): + self.check(value, kind, **fields) + + def test_decide_does_not_mutate_its_input(self): + value = ready_for_merge() + before = copy.deepcopy(value) + decide(value) + self.assertEqual(value, before) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_supervisor_policy.py b/tests/test_supervisor_policy.py new file mode 100644 index 0000000..b6486bc --- /dev/null +++ b/tests/test_supervisor_policy.py @@ -0,0 +1,179 @@ +"""Pure merge-policy checks for the supervisor: globs, required checks and the L2 gate.""" + +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +from supervisor import policy # noqa: E402 + +HEAD = "a" * 40 + + +def run(name, conclusion="success", status="completed", id=1): + return {"name": name, "status": status, "conclusion": conclusion, "id": id} + + +def facts(**changes): + value = { + "kind": "work", + "branch": "task/a1", + "default_branch": "main", + "result_head": HEAD, + "review": {"verdict": "ready", "head_sha": HEAD}, + "pr": { + "state": "OPEN", + "isDraft": False, + "headRefOid": HEAD, + "headRefName": "task/a1", + "baseRefName": "main", + "mergeable": "MERGEABLE", + "mergeStateStatus": "CLEAN", + }, + "checks": "passed", + "checks_timed_out": False, + "files": ["docs/guide.md"], + "changed_lines": 12, + "owned_paths": ["docs/**"], + } + value.update(changes) + return value + + +def pr(**changes): + return {**facts()["pr"], **changes} + + +GATE = { + "required_checks": ["check"], + "protected_paths": [".github/**", ".agent/supervisor.toml"], + "max_changed_lines": 400, + "allow_reconcile_merge": False, +} + + +class GlobTests(unittest.TestCase): + def test_glob_semantics(self): + cases = [ + ("docs/a.md", "docs/**", True), + ("docs/x/y/a.md", "docs/**", True), + ("docs", "docs/**", False), + ("docsx/a.md", "docs/**", False), + ("src/a.py", "src/*.py", True), + ("src/x/a.py", "src/*.py", False), + ("src/x/a.py", "src/**/*.py", True), + ("src/a.py", "src/**/*.py", True), + ("README.md", "README.md", True), + ("README.mdx", "README.md", False), + ("a1.txt", "a?.txt", True), + ("ab.txt", "a[0-9].txt", False), + ("a.b", "a.b", True), + ("axb", "a.b", False), + ] + for path, pattern, expected in cases: + with self.subTest(path=path, pattern=pattern): + self.assertEqual(policy.glob_match(path, pattern), expected) + + +class RequiredCheckTests(unittest.TestCase): + def test_all_required_successes_pass(self): + state, detail = policy.required_checks_state( + ["check", "lint"], [run("check"), run("lint"), run("optional", "failure")], [] + ) + self.assertEqual(state, "passed") + self.assertEqual(detail, {"check": "success", "lint": "success"}) + + def test_newest_run_wins(self): + runs = [run("check", "failure", id=1), run("check", "success", id=2)] + self.assertEqual(policy.required_checks_state(["check"], runs, [])[0], "passed") + runs = [run("check", "success", id=3), run("check", "failure", id=4)] + self.assertEqual(policy.required_checks_state(["check"], runs, [])[0], "failed") + + def test_non_success_conclusions_fail(self): + for conclusion in ("failure", "skipped", "neutral", "cancelled", "timed_out", None): + with self.subTest(conclusion=conclusion): + state, _ = policy.required_checks_state(["check"], [run("check", conclusion)], []) + self.assertEqual(state, "failed") + + def test_missing_required_check_fails(self): + state, detail = policy.required_checks_state(["check"], [run("other")], []) + self.assertEqual(state, "failed") + self.assertEqual(detail, {"check": "missing"}) + + def test_incomplete_run_is_pending(self): + state, _ = policy.required_checks_state( + ["check"], [run("check", None, status="in_progress")], [] + ) + self.assertEqual(state, "pending") + + def test_commit_statuses_count(self): + statuses = [{"context": "ci/legacy", "state": "pending", "id": 1}] + self.assertEqual(policy.required_checks_state(["ci/legacy"], [], statuses)[0], "pending") + statuses.append({"context": "ci/legacy", "state": "success", "id": 2}) + self.assertEqual(policy.required_checks_state(["ci/legacy"], [], statuses)[0], "passed") + statuses.append({"context": "ci/legacy", "state": "error", "id": 3}) + self.assertEqual(policy.required_checks_state(["ci/legacy"], [], statuses)[0], "failed") + + def test_no_required_checks_never_passes(self): + self.assertEqual(policy.required_checks_state([], [run("check")], [])[0], "failed") + + +class MergeGateTests(unittest.TestCase): + def assertGate(self, value, verdict, reason=None): + result, reasons = policy.merge_gate(value, GATE) + self.assertEqual(result, verdict, reasons) + if reason: + self.assertIn(reason, reasons) + + def test_clean_reviewed_pr_merges(self): + self.assertGate(facts(), "merge") + self.assertGate(facts(pr=pr(mergeStateStatus="HAS_HOOKS")), "merge") + + def test_holds(self): + cases = [ + (facts(review=None), "no-ready-review"), + (facts(review={"verdict": "needs-fix", "head_sha": HEAD}), "no-ready-review"), + (facts(review={"verdict": "ready", "head_sha": "b" * 40}), "no-ready-review"), + (facts(pr=pr(state="CLOSED")), "pr-not-open"), + (facts(pr=pr(state="MERGED")), "pr-not-open"), + (facts(pr=pr(isDraft=True)), "pr-draft"), + (facts(pr=pr(headRefOid="b" * 40)), "head-moved"), + (facts(pr=pr(headRefName="other")), "wrong-head-branch"), + (facts(pr=pr(baseRefName="develop")), "wrong-base"), + (facts(pr=pr(mergeable="CONFLICTING", mergeStateStatus="DIRTY")), "conflicting"), + (facts(pr=pr(mergeStateStatus="BEHIND")), "behind"), + (facts(pr=pr(mergeStateStatus="UNSTABLE")), "merge-state-UNSTABLE"), + (facts(pr=pr(mergeStateStatus="BLOCKED")), "merge-state-BLOCKED"), + (facts(checks="failed"), "checks-failed"), + (facts(checks="pending", checks_timed_out=True), "checks-timeout"), + (facts(files=["docs/a.md", "src/x.py"]), "outside-owned:src/x.py"), + ( + facts(files=[".github/workflows/x.yml"], owned_paths=["**"]), + "protected:.github/workflows/x.yml", + ), + (facts(changed_lines=401), "too-large"), + (facts(binary=["docs/logo.png"]), "binary:docs/logo.png"), + (facts(files=None), "diff-unavailable"), + (facts(files=[]), "empty-diff"), + (facts(kind="reconcile"), "reconcile-merge-not-granted"), + ] + for value, reason in cases: + with self.subTest(reason=reason): + self.assertGate(value, "hold", reason) + + def test_waits(self): + self.assertGate(facts(checks="pending"), "wait", "checks-pending") + self.assertGate(facts(checks="pending", pr=pr(mergeStateStatus="BLOCKED")), "wait") + self.assertGate(facts(pr=pr(mergeable="UNKNOWN", mergeStateStatus="UNKNOWN")), "wait") + + def test_hold_wins_over_wait(self): + self.assertGate(facts(checks="pending", changed_lines=999), "hold", "too-large") + + def test_reconcile_merge_with_grant(self): + gate = {**GATE, "allow_reconcile_merge": True} + self.assertEqual(policy.merge_gate(facts(kind="reconcile"), gate)[0], "merge") + + +if __name__ == "__main__": + unittest.main()