#!/usr/bin/env python3 """Launch one fresh, bounded campaign quantum with explicit model and paired Git worktrees.""" from __future__ import annotations import argparse import json import os from pathlib import Path import secrets import subprocess import sys from campaign import (Campaign, ControlError, MODELS, ROLE_STATUS, atomic_json, digest, file_hash, now, read_json) COMPACTION_MODEL = "openai/gpt-5.5" def command(args, cwd=None, env=None): try: result = subprocess.run(args, cwd=cwd, env=env, text=True, capture_output=True, check=False) except OSError as exc: raise ControlError(f"cannot execute {args[0]}: {exc}") from exc if result.returncode: raise ControlError(f"command failed ({result.returncode}): {args!r}: {result.stderr.strip()}") return result.stdout.rstrip("\n") def git(path, *args): return command(["git", "-C", str(path), *args]) def absolute_directory(value): p = Path(value) if not p.is_absolute() or not p.is_dir() or any(char in str(p) for char in '\n\r"\\'): raise ControlError(f"explicit absolute directory required: {value}") return p.resolve() def check_worktrees(campaign, contract, engine, re_tree): trees = {"engine": absolute_directory(engine), "re": absolute_directory(re_tree)} if trees["engine"] == trees["re"] or trees["engine"].is_relative_to(trees["re"]) or trees["re"].is_relative_to(trees["engine"]): raise ControlError("paired worktrees must be distinct, non-nested directories") common_dirs = [] for kind, tree in trees.items(): source = contract["baseline"][kind] canonical = absolute_directory(source["path"]) if tree == canonical: raise ControlError(f"{kind}: worker must use a distinct linked worktree, not canonical source") if Path(git(tree, "rev-parse", "--show-toplevel")).resolve() != tree: raise ControlError(f"{kind}: supplied path is not worktree root") common = Path(git(canonical, "rev-parse", "--path-format=absolute", "--git-common-dir")).resolve() if Path(git(tree, "rev-parse", "--path-format=absolute", "--git-common-dir")).resolve() != common: raise ControlError(f"{kind}: worktree is not paired with canonical source repository") entries = git(canonical, "-c", "core.quotePath=false", "worktree", "list", "--porcelain").splitlines() if f"worktree {tree}" not in entries: raise ControlError(f"{kind}: not a registered Git worktree") if git(tree, "rev-parse", "HEAD") != source["commit"]: raise ControlError(f"{kind}: worktree HEAD does not match pinned baseline") if git(canonical, "cat-file", "-t", source["commit"]) != "commit": raise ControlError(f"{kind}: baseline is not a commit") common_dirs.append(common) if common_dirs[0] == common_dirs[1]: raise ControlError("engine and RE must be different source repositories") if Path(contract["baseline"]["re"]["path"]).resolve() != campaign.root: raise ControlError("RE baseline/state root mismatch") return trees def source_identity(trees): result = {} for kind, tree in trees.items(): files = {} for name in sorted(set(git(tree, "ls-files", "-z", "--cached", "--others", "--exclude-standard").split("\0")) - {""}): p = tree / name if p.is_symlink(): files[name] = {"symlink": os.readlink(p)} elif p.is_file(): files[name] = {"sha256": file_hash(p), "mode": p.stat().st_mode & 0o777} elif p.is_dir(): raise ControlError(f"submodule/directory requires explicit manifest support: {p}") else: files[name] = None result[kind] = {"path": str(tree), "commit": git(tree, "rev-parse", "HEAD"), "manifest_sha256": digest(files), "files": files} return result def check_launch(campaign, contract, role, actor): if role not in MODELS or contract["status"] not in ROLE_STATUS[role]: raise ControlError("role cannot launch in this contract status") recovery = role in {"lead", "resolver"} and (contract["status"] == "blocked" or bool(campaign.open_surprises(contract["id"]))) if campaign.open_surprises(contract["id"]) and not recovery: raise ControlError("unresolved surprise: resolve using campaign CLI before launching another quantum") if role not in {"lead", "resolver", "verifier", "lab"} and contract["owner"] != {"name": actor, "role": role}: raise ControlError("launch role/actor does not match contract owner") if role == "verifier" and actor == contract["owner"]["name"]: raise ControlError("verifier must be independent") if not recovery and (role not in {"lead", "resolver", "architecture-review"} or contract["checkpoint"]): campaign.check_checkpoint(contract, require_artifacts=False, fresh=False) if contract.get("evidence"): campaign.check_evidence(contract) if contract["status"] == "implementing" and sum(c["status"] == "implementing" for c in campaign.contracts()) > campaign.models["implementation_wip"]: raise ControlError("implementation WIP overflow") return recovery def configuration(campaign, role, trees=None, recovery=False): path = campaign.root / "opencode.json" config = read_json(path) name = campaign.models["agents"][role] agent = config.get("agent", {}).get(name, {}) if agent.get("model") != MODELS[role] or agent.get("steps") != 40: raise ControlError("repo-local agent must explicitly match requested model and 40-step bound") if config.get("small_model") != COMPACTION_MODEL or config.get("agent", {}).get("compaction", {}).get("model") != COMPACTION_MODEL: raise ControlError("compaction routing must explicitly use openai/gpt-5.5") prompt = agent.get("prompt", "") if prompt.startswith("{file:") and prompt.endswith("}"): prompt_path = (path.parent / prompt[6:-1]).resolve() if not prompt_path.is_relative_to(campaign.root) or not prompt_path.is_file(): raise ControlError("missing role prompt file") prompt = prompt_path.read_text() # Environment content is the last configuration layer; freeze the expanded role prompt. permission = {"read": "allow", "glob": "allow", "grep": "allow", "list": "allow", "bash": "allow", "external_directory": {"*": "deny", **{pattern: "allow" for p in [campaign.root, *(trees or {}).values()] for pattern in (str(p), str(p) + "/*")}}, "task": "deny", "question": "deny"} # Preserve role-specific denies while making bounded noninteractive access explicit. permission.update(agent.get("permission", {})) permission["task"] = "deny" permission["question"] = "deny" if recovery: permission["edit"] = {"*": "deny", str(campaign.root / "campaign/runtime") + "/*": "allow", str(campaign.root / "campaign/contracts") + "/*": "allow"} permission["bash"] = {"*": "deny", **{f"git -C * {action} *": "allow" for action in ("status", "diff", "log", "rev-parse", "show")}, **{f"python3 {campaign.root}/tools/campaign.py --state-root {campaign.root} {action} *": "allow" for action in ("checkpoint", "resolve", "surprise", "status", "validate")}} overlay = {"model": MODELS[role], "agent": {name: {"model": MODELS[role], "steps": 40, "prompt": prompt, "permission": permission}}} env = dict(os.environ) for key in list(env): if key.startswith("OPENCODE_"): env.pop(key) env["OPENCODE_CONFIG"] = str(path) env["OPENCODE_CONFIG_CONTENT"] = json.dumps(overlay) files = {str(p.relative_to(campaign.root)): file_hash(p) for p in sorted( {path, campaign.base / "models.json", *(campaign.base / "agents").glob("**/*"), *(campaign.root / ".opencode").glob("**/*")}) if p.is_file()} return name, env, {"path": str(path), "sha256": file_hash(path), "overlay": overlay, "expanded_prompt_sha256": digest(prompt), "canonical_files": files} def effective_configuration(executable, cwd, env, agent, model): effective = json.loads(command([executable, "debug", "config"], cwd=cwd, env=env)) actual = effective.get("agent", {}).get(agent, {}) requested = json.loads(env["OPENCODE_CONFIG_CONTENT"])["agent"][agent] if any(actual.get(k) != requested[k] for k in ("model", "steps", "prompt")) or effective.get("model") != model: raise ControlError("effective agent model/steps/expanded prompt mismatch") actual_permissions = actual.get("permission", {}) for key, value in requested["permission"].items(): present = actual_permissions.get(key) if isinstance(value, dict): if not isinstance(present, dict) or any(present.get(k) != v for k, v in value.items()): raise ControlError("effective bounded permissions mismatch") elif present != value: raise ControlError("effective bounded permissions mismatch") # Store hashes, not effective provider credentials. Full digest detects inherited config drift. return {"sha256": digest(effective), "agent_sha256": digest(actual)} def run(args): campaign = Campaign(args.state_root) contract = campaign.load(args.contract) record_contract = contract campaign.identity(args.actor, args.role, MODELS[args.role]) recovery = check_launch(campaign, contract, args.role, args.actor) trees = check_worktrees(campaign, contract, args.engine_worktree, args.re_worktree) agent, env, config = configuration(campaign, args.role, trees, recovery) available = command([args.opencode, "models"], cwd=campaign.root, env=env).splitlines() model = campaign.models["roles"][args.role] if model not in {line.strip() for line in available}: raise ControlError(f"requested model unavailable in opencode models: {model}; no fallback") run_id = "run-" + secrets.token_hex(12) authority = ("RESOLUTION-ONLY: read evidence, record decisions/checkpoints; no implementation or architecture source edits. " if recovery else "Sol loop authority: perform approved architecture/planning work in owned scope. " if args.role in {"lead", "architecture-review"} else "Workers cannot change architecture. ") prompt = ( f"Start a FRESH bounded campaign quantum (40 steps). No delegation.\n" f"Role={args.role}; requested model={model}; actor={args.actor}; checkpoint session={run_id}.\n" f"Canonical state root: {campaign.root}\n" f"Contract: {campaign.contract_path(args.contract)}\n" f"Latest checkpoint: {contract['checkpoint']}\n" f"Engine worktree: {trees['engine']}\nRE worktree: {trees['re']}\n" "Read canonical campaign/README.md, contract, latest checkpoint and open surprises before work. " "Verify source/worktree/resource identities. Only edit owned scope. Missing state means blocked. " "Do not commit, stage, push or mutate lab resources without explicit authority and a lease. " "Checkpoint every 20 tool calls or 15 minutes, before experiments, compaction and stopping; " "persist under canonical campaign/ using tools/campaign.py with the session above, exact model, " "actor and role. Include observations versus decisions, source identities, artifact paths, tests, " "blockers and ONE exact next action. Fresh checkpoint at quantum end is mandatory. " "Record a surprise and pause affected work if assumptions are falsified. " + authority + "Do not rely on chat resume or automatic compaction." ) cmd = [args.opencode, "run", "--format", "json", "--model", model, "--agent", agent, prompt] record = {"schema": "sots-run/1", "id": run_id, "contract": args.contract, "role": args.role, "actor": args.actor, "requested_model": model, "checkpoint_session": run_id, "cwd": str(trees[args.cwd]), "worktrees": {k: str(v) for k, v in trees.items()}, "config": config, "command": cmd, "max_steps": 40, "status": "dry-run", "resolution_only": recovery} config["effective"] = effective_configuration(args.opencode, trees[args.cwd], env, agent, model) if args.dry_run: return record relative = f"campaign/runtime/runs/{run_id}.json" active_path = campaign.path(f"campaign/runtime/runs/active-{args.contract}.json") with campaign.lock(): contract = campaign.load(args.contract) if contract != record_contract: raise ControlError("contract/checkpoint changed before reservation") if check_launch(campaign, contract, args.role, args.actor) != recovery: raise ControlError("launch scope changed before reservation") trees = check_worktrees(campaign, contract, args.engine_worktree, args.re_worktree) current_config = configuration(Campaign(args.state_root), args.role, trees, recovery)[2] if current_config != {k: v for k, v in config.items() if k != "effective"}: raise ControlError("canonical config changed before reservation") if active_path.exists() and read_json(active_path).get("status") == "running": raise ControlError("contract already has a running quantum; inspect and explicitly close interrupted run") for path in campaign.path("campaign/runtime/runs").glob("run-*.json"): other = read_json(path) if other.get("status") == "running" and set(other.get("worktrees", {}).values()) & {str(v) for v in trees.values()}: raise ControlError("worktree already reserved by running quantum") record.update(status="running", started_at=now(), source_before=source_identity(trees)) atomic_json(campaign.path(relative), record) atomic_json(active_path, {"contract": args.contract, "run": run_id, "status": "running"}) log_relative = f"campaign/runtime/runs/{run_id}.jsonl" stderr_relative = f"campaign/runtime/runs/{run_id}.stderr.log" record.update(log=log_relative, stderr=stderr_relative) observed_sessions = set() observed_models = set() events = 0 malformed_events = 0 error_events = 0 completed_sessions = set() failure = None try: with campaign.path(log_relative).open("w") as log, campaign.path(stderr_relative).open("w") as errors: process = subprocess.Popen(cmd, cwd=trees[args.cwd], env=env, stdout=subprocess.PIPE, stderr=errors, text=True) try: for line in process.stdout: log.write(line) log.flush() try: event = json.loads(line) except ValueError: malformed_events += 1 continue events += 1 if isinstance(event, dict): session = event.get("sessionID") if isinstance(session, str): if session.strip(): observed_sessions.add(session) if event.get("type") == "error": error_events += 1 if event.get("type") == "step_finish" and event.get("part", {}).get("reason") == "stop" and isinstance(session, str) and session.strip(): completed_sessions.add(session) # Keep raw events as authority; collect model IDs only when explicitly emitted. for part in (event, event.get("part", {}), event.get("info", {})): if isinstance(part, dict) and part.get("providerID") and part.get("modelID"): observed_models.add(f"{part['providerID']}/{part['modelID']}") record["returncode"] = process.wait() finally: process.stdout.close() if process.poll() is None: process.terminate() try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill() process.wait() if record["returncode"]: raise ControlError(f"opencode exited {record['returncode']}") if not events or malformed_events: raise ControlError("missing or malformed runner JSON events") if error_events or not completed_sessions or len(observed_sessions) != 1 or completed_sessions != observed_sessions: raise ControlError("runner error or missing successful completed step/session") if observed_models and observed_models != {model}: raise ControlError("observed runner model differs from requested model") with campaign.lock(): current_config = configuration(Campaign(args.state_root), args.role, trees, recovery)[2] if current_config != {k: v for k, v in config.items() if k != "effective"} or effective_configuration(args.opencode, trees[args.cwd], env, agent, model) != config["effective"]: raise ControlError("canonical/effective model, prompt or config changed during run") final = campaign.load(args.contract) cp = campaign.check_checkpoint(final, since=record["started_at"], session=run_id, model=model, require_artifacts=False) if cp["actor"] != args.actor or cp["role"] != args.role: raise ControlError("end checkpoint actor/role mismatch") record["checkpoint"] = final["checkpoint"] record["source_after"] = source_identity(trees) if recovery and record["source_after"] != record["source_before"]: raise ControlError("resolution-only run changed worktree source") record["status"] = "complete" except (ControlError, OSError, ValueError, TypeError, AttributeError, KeyboardInterrupt) as exc: failure = str(exc) or "interrupted" record.update(status="incomplete", error=failure) finally: record.update(ended_at=now(), event_count=events, malformed_events=malformed_events, error_events=error_events, completed_sessions=sorted(completed_sessions), observed_model_status="emitted" if observed_models else "unavailable", observed_sessions=sorted(observed_sessions), observed_models=sorted(observed_models)) with campaign.lock(): atomic_json(campaign.path(relative), record) atomic_json(active_path, {"contract": args.contract, "run": run_id, "status": record["status"]}) if failure: raise ControlError(f"{failure}; incomplete run persisted at {relative}") return record def parser(): p = argparse.ArgumentParser(description=__doc__) p.add_argument("--state-root", required=True) p.add_argument("--role", choices=MODELS, required=True) p.add_argument("--actor", required=True) p.add_argument("--contract", required=True) p.add_argument("--engine-worktree", required=True) p.add_argument("--re-worktree", required=True) p.add_argument("--cwd", choices=["engine", "re"], required=True) p.add_argument("--opencode", default="opencode") p.add_argument("--dry-run", action="store_true") return p def main(argv=None): try: print(json.dumps(run(parser().parse_args(argv)), indent=2, sort_keys=True)) return 0 except (ControlError, OSError, KeyError, TypeError, ValueError) as exc: print(f"run_agent: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main())