From 2fc961b1410efb3f5f5460701a0fb9284d1f7e7b Mon Sep 17 00:00:00 2001 From: Tuomo Vendelin Date: Sat, 10 Oct 2026 22:12:17 +0300 Subject: [PATCH] feat: record agent sessions and operator halts in supervised rounds `session start|ack|end` records each launched agent before launch, with role rules that refuse double dispatch and out-of-order work; `review` accepts a finished reviewer session. `halt` and `clear-halt` record conditions that need an operator; uncleared halts block new sessions and closure. Live sessions block cancellation and closure. Co-Authored-By: Claude Opus 5.5 --- SUPERVISION.md | 42 +++++++- scripts/round.py | 228 ++++++++++++++++++++++++++++++++++++++++- tests/test_round_v2.py | 151 ++++++++++++++++++++++++++- 3 files changed, 414 insertions(+), 7 deletions(-) diff --git a/SUPERVISION.md b/SUPERVISION.md index fb369d6..45e3e08 100644 --- a/SUPERVISION.md +++ b/SUPERVISION.md @@ -56,8 +56,44 @@ Ledger writers take an exclusive lock on `.agent-runs/ROUND/round.lock` for the command, so an operator's command and the supervisor cannot lose each other's updates. This is a single-host lock, not distributed coordination. +### Sessions + +Every launched agent process is recorded before launch (write-ahead), so a restarted supervisor can +find it again instead of dispatching twice: + +```sh +python3 scripts/round.py session start ROUND --role planner|executor|reviewer|merger \ + [--assignment A1] [--head FULL_HEAD] --runner NAME --session-uuid UUID --cwd PATH \ + --deadline YYYY-MM-DDTHH:MM:SSZ [--procedure NAME] # prints {"session": "S1"} +python3 scripts/round.py session ack ROUND S1 --pid PID +python3 scripts/round.py session end ROUND S1 --state exited|timed-out|killed|lost \ + [--exit-code N] [--cost-usd X] [--imported results/A1-1.json] +``` + +A session is `launching`, then `running`, then ended. At most one live session exists per assignment +and one live planner per round. Role rules: an executor needs an `assigned` assignment or a +`needs-fix` review of the latest result (a repair); a reviewer needs the latest result's head and no +review of it yet; a merger needs a `ready` review of that head. `review --session S` accepts only a +finished reviewer session for the same assignment and head. Cancellation refuses while the +assignment has a live session; a merged resolution tolerates only its live merger session. Closure +refuses any live session. A merger session has no process: it is the write-ahead record of a merge +attempt, which the supervisor verifies by observation before resolving. + +### Halts + +```sh +python3 scripts/round.py halt ROUND --code CODE --detail TEXT --action TEXT \ + [--assignment A1] [--session S1] [--inventory inventory.json] +python3 scripts/round.py clear-halt ROUND H1 --evidence TEXT +``` + +A halt records a condition that cannot be recomputed from the ledger and observations: a lost or +timed-out session, invalid agent output, a failed push, publish or merge, a red final gate, or a +planner reporting no executable work or an owed decision. The inventory lists live sessions, +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. + ## Planned -Sessions and halts (`session start|ack|end`, `halt`, `clear-halt`), the supervisor -(`scripts/supervise.py`), role contracts, merge policy and reconciliation automation follow in later -changes on this branch. +The supervisor (`scripts/supervise.py`), role contracts, merge policy and reconciliation automation +follow in later changes on this branch. diff --git a/scripts/round.py b/scripts/round.py index 6b77881..dede01c 100644 --- a/scripts/round.py +++ b/scripts/round.py @@ -189,6 +189,9 @@ def resolve(repo, root, manifest, args): assignment["state"] not in {"merged", "cancelled"}, "Assignment is terminal" ) require(text(args.evidence), "Resolution needs review or cancellation evidence") + if supervised(manifest): + merging = {"merger"} if args.outcome == "merged" else set() + require_no_live_sessions(manifest, args.assignment, allow=merging) if args.outcome == "merged": bound_packet(root, assignment) require(assignment["state"] == "result-ready", "No candidate result to resolve") @@ -327,6 +330,12 @@ def main(): close(repo, root, manifest, args) elif args.action == "review": review(root, manifest, args) + elif args.action == "session": + session(root, manifest, args) + elif args.action == "halt": + halt(manifest, args) + elif args.action == "clear-halt": + clear_halt(manifest, args) save(root / "round.json", manifest) print(f"{args.round}: {args.action} recorded") return 0 @@ -450,7 +459,18 @@ def review(root, manifest, args): result = read(inside(root, latest)) require(args.head == result["head_sha"], "Review does not match the latest result head") require(text(args.evidence), "Review needs an evidence reference") - require(text(args.reviewer), "Review needs a reviewer") + if args.session: + reviewer = manifest["sessions"].get(args.session) + require( + reviewer + and reviewer["role"] == "reviewer" + and reviewer["assignment"] == args.assignment + and reviewer["head_sha"] == args.head + and reviewer["state"] == "exited", + "Review session must be a finished reviewer of this head", + ) + else: + require(text(args.reviewer), "Review needs a reviewer") assignment["reviews"].append( { "result": latest, @@ -458,7 +478,7 @@ def review(root, manifest, args): "verdict": args.verdict, "evidence": args.evidence, "reviewer": args.reviewer, - "session": None, + "session": args.session, "utc": utc_now(), } ) @@ -469,9 +489,208 @@ def supervised_close(manifest, args): require(args.delta is None, "--delta needs a supervised round") return require(args.delta is not None, "Supervised closure needs --delta changed|none") + require_no_live_sessions(manifest) + require_no_halts(manifest) manifest["behavior_delta"] = args.delta +LIVE_SESSION = {"launching", "running"} +UTC = re.compile(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ\Z") + + +def require_supervised(manifest): + require(supervised(manifest), "Sessions and halts need a supervised round") + + +def require_no_halts(manifest): + require( + all(h["cleared_utc"] for h in manifest["halts"]), "Round has an uncleared halt" + ) + + +def require_no_live_sessions(manifest, assignment=None, allow=frozenset()): + for name, record in manifest["sessions"].items(): + require( + record["state"] not in LIVE_SESSION + or record["role"] in allow + or (assignment is not None and record["assignment"] != assignment), + f"Session {name} is still live", + ) + + +def utc_value(value): + require(text(value) and UTC.match(value), "Expected YYYY-MM-DDTHH:MM:SSZ") + datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ") + return value + + +def session(root, manifest, args): + require_supervised(manifest) + sessions = manifest["sessions"] + if args.step == "start": + start_session(root, manifest, args) + return + record = sessions[args.session] + if args.step == "ack": + require(record["state"] == "launching", "Only a launching session can be acknowledged") + require(args.pid > 0, "Process id must be positive") + record.update(pid=args.pid, state="running") + return + require(record["state"] in LIVE_SESSION, "Session has already ended") + require(args.cost_usd is None or args.cost_usd >= 0, "Cost must not be negative") + if args.imported: + require(inside(root, args.imported).is_file(), "Imported record does not exist") + record.update( + state=args.state, + exit_code=args.exit_code, + cost_usd=args.cost_usd, + imported=args.imported, + ended_utc=utc_now(), + ) + + +def start_session(root, manifest, args): + require_no_halts(manifest) + sessions = manifest["sessions"] + for field in ("runner", "session_uuid", "cwd"): + require(text(getattr(args, field)), f"Session needs --{field.replace('_', '-')}") + require( + all(s["session_uuid"] != args.session_uuid for s in sessions.values()), + "Session UUID already recorded", + ) + utc_value(args.deadline) + if args.role == "planner": + require(args.assignment is None, "Planner sessions take no assignment") + require( + not any( + s["role"] == "planner" and s["state"] in LIVE_SESSION + for s in sessions.values() + ), + "A planner session is already live", + ) + else: + assignment = manifest["assignments"][args.assignment] + require( + not any( + s["assignment"] == args.assignment and s["state"] in LIVE_SESSION + for s in sessions.values() + ), + "Assignment already has a live session", + ) + current = latest_review(assignment) if assignment["results"] else None + if args.role == "executor": + require( + assignment["state"] == "assigned" + or ( + assignment["state"] == "result-ready" + and current + and current["verdict"] == "needs-fix" + ), + "Executor needs an assigned or needs-fix assignment", + ) + elif args.role == "reviewer": + require(assignment["state"] == "result-ready", "No candidate result to review") + result = read(inside(root, assignment["results"][-1])) + require(args.head == result["head_sha"], "Reviewer head is not the latest result") + require(current is None, "Latest result already has a review") + else: + require(assignment["state"] == "result-ready", "No candidate result to merge") + require_ready_review(assignment, args.head) + name = f"S{len(sessions) + 1}" + sessions[name] = { + "role": args.role, + "assignment": args.assignment, + "procedure": args.procedure, + "runner": args.runner, + "session_uuid": args.session_uuid, + "cwd": args.cwd, + "head_sha": args.head, + "pid": None, + "started_utc": utc_now(), + "deadline_utc": args.deadline, + "state": "launching", + "exit_code": None, + "cost_usd": None, + "imported": None, + "ended_utc": None, + } + print(json.dumps({"session": name})) + + +def halt(manifest, args): + require_supervised(manifest) + for field in ("code", "detail", "action"): + require(text(getattr(args, field)), f"Halt needs --{field}") + if args.assignment is not None: + require(args.assignment in manifest["assignments"], "Unknown assignment") + if args.session is not None: + require(args.session in manifest["sessions"], "Unknown session") + inventory = read(Path(args.inventory)) if args.inventory else None + manifest["halts"].append( + { + "id": f"H{len(manifest['halts']) + 1}", + "utc": utc_now(), + "code": args.code, + "assignment": args.assignment, + "session": args.session, + "detail": args.detail, + "action": args.action_needed, + "inventory": inventory, + "cleared_utc": None, + "cleared_evidence": None, + } + ) + + +def clear_halt(manifest, args): + require_supervised(manifest) + matches = [h for h in manifest["halts"] if h["id"] == args.halt] + require(matches, "Unknown halt") + require(matches[0]["cleared_utc"] is None, "Halt already cleared") + require(text(args.evidence), "Clearing a halt needs evidence") + matches[0].update(cleared_utc=utc_now(), cleared_evidence=args.evidence) + + +def add_session_arguments(actions): + sub = actions.add_parser("session") + steps = sub.add_subparsers(dest="step", required=True) + start = steps.add_parser("start") + start.add_argument("round") + start.add_argument( + "--role", choices=["planner", "executor", "reviewer", "merger"], required=True + ) + start.add_argument("--assignment") + start.add_argument("--head") + start.add_argument("--procedure") + for field in ("runner", "session-uuid", "cwd", "deadline"): + start.add_argument(f"--{field}", required=True) + ack = steps.add_parser("ack") + ack.add_argument("round") + ack.add_argument("session") + ack.add_argument("--pid", type=int, required=True) + end = steps.add_parser("end") + end.add_argument("round") + end.add_argument("session") + end.add_argument( + "--state", choices=["exited", "timed-out", "killed", "lost"], required=True + ) + end.add_argument("--exit-code", type=int) + end.add_argument("--cost-usd", type=float) + end.add_argument("--imported", help="Record path inside the round directory") + sub = actions.add_parser("halt") + sub.add_argument("round") + for field in ("code", "detail"): + sub.add_argument(f"--{field}", required=True) + sub.add_argument("--action", dest="action_needed", required=True) + sub.add_argument("--assignment") + sub.add_argument("--session") + sub.add_argument("--inventory", help="JSON file listing live sessions and resources") + sub = actions.add_parser("clear-halt") + sub.add_argument("round") + sub.add_argument("halt") + sub.add_argument("--evidence", required=True) + + def add_supervised_arguments(actions): actions.choices["init"].add_argument("--supervised", action="store_true") assign_parser = actions.choices["assign"] @@ -488,7 +707,10 @@ def add_supervised_arguments(actions): "--verdict", choices=["ready", "needs-fix", "blocked"], required=True ) sub.add_argument("--evidence", required=True) - sub.add_argument("--reviewer", required=True) + who = sub.add_mutually_exclusive_group(required=True) + who.add_argument("--reviewer") + who.add_argument("--session", help="Finished reviewer session") + add_session_arguments(actions) if __name__ == "__main__": diff --git a/tests/test_round_v2.py b/tests/test_round_v2.py index 0ef0a57..e356a8a 100644 --- a/tests/test_round_v2.py +++ b/tests/test_round_v2.py @@ -12,7 +12,9 @@ SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "round.py" -class SupervisedRoundTests(unittest.TestCase): +class Helpers(unittest.TestCase): + """Temporary repository and CLI helpers shared by supervised-round tests.""" + def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) @@ -142,6 +144,7 @@ def close(self, *extra, ok=True): ok=ok, ) +class SupervisedRoundTests(Helpers): def test_default_init_stays_version_one_and_refuses_supervised_options(self): self.cli("init", "R1", "--base", "main") self.assertEqual( @@ -272,5 +275,151 @@ def test_ledger_writes_wait_for_the_round_lock(self): self.assertEqual(self.manifest()["assignments"]["A1"]["state"], "cancelled") +class SessionAndHaltTests(Helpers): + """Sessions record launched agent processes; halts record conditions needing an operator.""" + + DEADLINE = "2099-01-01T00:00:00Z" + + def session(self, role, *extra, ok=True, assignment="A1", head=None): + args = ["session", "start", "R1", "--role", role, "--runner", "claude", + "--session-uuid", f"uuid-{len(self.manifest()['sessions']) + 1}", + "--cwd", "/tmp/worktree", "--deadline", self.DEADLINE] + if assignment: + args += ["--assignment", assignment] + if head: + args += ["--head", head] + result = self.cli(*args, *extra, ok=ok) + return json.loads(result.stdout.splitlines()[0])["session"] if ok else None + + def end(self, session, state="exited", *extra, ok=True): + return self.cli("session", "end", "R1", session, "--state", state, *extra, ok=ok) + + def test_session_lifecycle_is_recorded(self): + self.start() + session = self.session("executor", "--procedure", "execute-task") + self.assertEqual(session, "S1") + record = self.manifest()["sessions"]["S1"] + self.assertEqual(record["state"], "launching") + self.assertEqual(record["role"], "executor") + self.assertEqual(record["assignment"], "A1") + self.assertEqual(record["session_uuid"], "uuid-1") + self.assertEqual(record["deadline_utc"], self.DEADLINE) + self.cli("session", "ack", "R1", "S1", "--pid", "0", ok=False) + self.cli("session", "ack", "R1", "S1", "--pid", "4242") + self.cli("session", "ack", "R1", "S1", "--pid", "4243", ok=False) + self.assertEqual(self.manifest()["sessions"]["S1"]["state"], "running") + self.end("S1", "exited", "--imported", "results/missing.json", ok=False) + self.end("S1", "exited", "--exit-code", "0", "--cost-usd", "-1", ok=False) + self.end("S1", "exited", "--exit-code", "0", "--cost-usd", "1.25") + record = self.manifest()["sessions"]["S1"] + self.assertEqual((record["state"], record["exit_code"], record["cost_usd"]), + ("exited", 0, 1.25)) + self.assertIsNotNone(record["ended_utc"]) + self.end("S1", "lost", ok=False) + + def test_session_start_validates_identity(self): + self.start() + self.session("executor", "--deadline", "tomorrow", ok=False) + self.session("planner", ok=False) + self.session("planner", assignment=None) + self.session("planner", assignment=None, ok=False) + self.session("executor", assignment="A9", ok=False) + self.cli( + "session", "start", "R1", "--role", "executor", "--assignment", "A1", + "--runner", "claude", "--session-uuid", "uuid-1", "--cwd", "/tmp/x", + "--deadline", self.DEADLINE, ok=False, + ) + self.session("executor") + + def test_one_live_session_per_assignment_and_role_rules(self): + self.start() + self.session("reviewer", head=self.base, ok=False) + self.session("merger", head=self.base, ok=False) + first = self.session("executor") + self.session("executor", ok=False) + self.end(first, "lost") + second = self.session("executor") + self.end(second) + self.submit() + self.session("executor", ok=False) + self.session("reviewer", head="0" * 40, ok=False) + reviewer = self.session("reviewer", head=self.base) + self.end(reviewer) + self.cli("review", "R1", "A1", "--head", self.base, "--verdict", "needs-fix", + "--evidence", "x", "--session", reviewer) + self.session("reviewer", head=self.base, ok=False) + repair = self.session("executor") + self.end(repair) + self.review("ready") + self.session("executor", ok=False) + self.session("merger", head="0" * 40, ok=False) + merger = self.session("merger", head=self.base) + self.merge() + self.end(merger) + self.assertEqual(self.manifest()["assignments"]["A1"]["state"], "merged") + + def test_session_review_needs_a_finished_matching_reviewer(self): + self.start() + executor = self.session("executor") + self.end(executor) + self.submit() + reviewer = self.session("reviewer", head=self.base) + args = ["review", "R1", "A1", "--head", self.base, "--verdict", "ready", + "--evidence", "x"] + self.cli(*args, "--session", reviewer, ok=False) + self.end(reviewer) + self.cli(*args, "--session", executor, ok=False) + self.cli(*args, "--session", reviewer, "--reviewer", "operator", ok=False) + self.cli(*args, "--session", reviewer) + self.assertEqual( + self.manifest()["assignments"]["A1"]["reviews"][-1]["session"], reviewer + ) + + def test_live_sessions_block_cancellation_and_closure(self): + self.start() + executor = self.session("executor") + self.cli("resolve", "R1", "A1", "--outcome", "cancelled", "--evidence", "x", ok=False) + self.end(executor, "killed") + self.cli("resolve", "R1", "A1", "--outcome", "cancelled", "--evidence", "Stopped") + planner = self.session("planner", assignment=None) + self.close("--delta", "none", ok=False) + self.end(planner) + self.close("--delta", "none") + + def test_halts_block_dispatch_and_closure_until_cleared(self): + self.start() + inventory = self.repo / ".agent-runs" / "inventory.json" + inventory.write_text(json.dumps({"live_sessions": [], "worktrees": ["/tmp/w"]})) + self.cli("halt", "R1", "--code", "session-lost", "--detail", "S1 vanished", + "--action", "Inspect /tmp/w", "--assignment", "A9", ok=False) + self.cli("halt", "R1", "--code", "session-lost", "--detail", "S1 vanished", + "--action", "Inspect /tmp/w", "--assignment", "A1", + "--inventory", str(inventory)) + halt = self.manifest()["halts"][0] + self.assertEqual((halt["id"], halt["code"], halt["assignment"]), + ("H1", "session-lost", "A1")) + self.assertEqual(halt["inventory"]["worktrees"], ["/tmp/w"]) + self.assertIsNone(halt["cleared_utc"]) + self.session("executor", ok=False) + self.cli("resolve", "R1", "A1", "--outcome", "cancelled", "--evidence", "Stopped") + self.close("--delta", "none", ok=False) + self.cli("clear-halt", "R1", "H2", "--evidence", "x", ok=False) + self.cli("clear-halt", "R1", "H1", "--evidence", "", ok=False) + self.cli("clear-halt", "R1", "H1", "--evidence", "Worktree inspected and removed") + self.cli("clear-halt", "R1", "H1", "--evidence", "again", ok=False) + self.assertIsNotNone(self.manifest()["halts"][0]["cleared_utc"]) + self.close("--delta", "none") + + def test_version_one_rounds_refuse_sessions_and_halts(self): + self.cli("init", "R1", "--base", "main") + self.assign("A1") + self.cli( + "session", "start", "R1", "--role", "executor", "--assignment", "A1", + "--runner", "claude", "--session-uuid", "u", "--cwd", "/tmp/x", + "--deadline", self.DEADLINE, ok=False, + ) + self.cli("halt", "R1", "--code", "x", "--detail", "d", "--action", "a", ok=False) + + if __name__ == "__main__": unittest.main()