342 lines
20 KiB
Python
342 lines
20 KiB
Python
#!/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())
|