Something went wrong. Try again.
wip: experimental harness for learning about agent orchestration
Something went wrong. Try again.
agent-orchestrator agents.py
47 kB · 1067 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068#!/usr/bin/env python3"""agent-orchestrator — one driver, one config (agents.toml) for a fleet of agents.
A generic, reusable harness for running and supervising AI-agent sessions in tmux. Everyagent — a Builder/Adversary loop pair, a persistent supervisor, a one-shot task — is declaredin a single TOML config; the watchdog reads the SAME file, so there is no env-vs-file drift.Nothing about any particular project lives in this code: paths, the loop kickoff preamble, thehandoff conventions, and the on-complete hook are all supplied by the project's config.
Usage: agents.py up [name...] start enabled agents (or just the named ones); use-or-create agents.py down [name...] stop agents (or all) agents.py status one table: every agent — kind, backend, model, session, phase agents.py watchdog the supervisor loop (reads the config every tick) agents.py logs <name> tail an agent's session log agents.py phase [set N|next|show] inspect / move the loop phase agents.py tokens per-phase token + time report (needs [watchdog].log_tokens = true) agents.py selftest backend activity-detector regression checks (no config needed) agents.py init [dir] scaffold a starter agents.toml + prompts/ in a project dir
Options: --config PATH config file (default: ./agents.toml, else <script dir>/agents.toml)
Config is authoritative. A one-off override env AGENT_MODEL_<name> / AGENT_BACKEND_<name>affects a single invocation only; the persisted watchdog always re-reads the file."""
import hashlib, json, os, re, shlex, subprocess, sys, time, tomllibfrom datetime import datetime, timedeltafrom pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent
# ── config loading ──────────────────────────────────────────────────────────────
def _cfg_path(argv): if "--config" in argv: return Path(argv[argv.index("--config") + 1]) cwd_cfg = Path.cwd() / "agents.toml" return cwd_cfg if cwd_cfg.exists() else SCRIPT_DIR / "agents.toml"
def _resolve(base, p): """Resolve a possibly-relative config path against the project root.""" pp = Path(os.path.expanduser(str(p))) return pp if pp.is_absolute() else (Path(base) / pp)
def load_config(path): path = Path(path) with open(path, "rb") as f: raw = tomllib.load(f) defaults = raw.get("defaults", {}) # The project root: everything project-supplied (prompts, templates, relative paths) is # resolved against it. Defaults to the directory holding the config file; override with # defaults.project_dir (useful when the config lives in a sandbox but prompts live elsewhere). project_dir = _resolve(path.resolve().parent, defaults.get("project_dir", ".")).resolve()
session_prefix = defaults.get("session_prefix") if not session_prefix: die("config error: [defaults].session_prefix is required (e.g. session_prefix = \"myproj-\")") log_dir_raw = defaults.get("log_dir") if not log_dir_raw: die("config error: [defaults].log_dir is required")
cfg = { "watchdog": raw.get("watchdog", {}), "backends": raw.get("backend", {}), "defaults": defaults, "loop": raw.get("loop", {}), "project_dir": str(project_dir), "log_dir": str(_resolve(project_dir, log_dir_raw)), "session_prefix": session_prefix, } agents = {} for a in raw.get("agent", []): m = {**defaults, **a} m["session"] = a.get("session", session_prefix + a["name"]) m["kind"] = a.get("kind", "persistent") m["dir"] = str(_resolve(project_dir, a.get("dir", defaults.get("dir", ".")))) # one-off env override (single invocation; watchdog ignores via fresh load each tick) env_model = os.environ.get(f"AGENT_MODEL_{a['name']}") env_backend = os.environ.get(f"AGENT_BACKEND_{a['name']}") if env_model: m["model"] = env_model if env_backend: m["backend"] = env_backend agents[a["name"]] = m cfg["agents"] = agents cfg["services"] = {s["name"]: {**defaults, **s, "session": session_prefix + s["name"], "dir": str(_resolve(project_dir, s.get("dir", ".")))} for s in raw.get("service", [])} cfg["state_dir"] = os.path.join(cfg["log_dir"], "state") Path(cfg["state_dir"]).mkdir(parents=True, exist_ok=True) return cfg
def backend_of(cfg, agent): b = cfg["backends"].get(agent["backend"]) if not b: die(f"agent {agent['name']}: unknown backend {agent['backend']!r}") return b
# ── logging ───────────────────────────────────────────────────────────────────
def log(msg): print(f"[agents {datetime.now():%H:%M:%S}] {msg}", flush=True)
def die(msg): log(f"ERROR: {msg}") sys.exit(1)
# ── tmux helpers ────────────────────────────────────────────────────────────────# ALWAYS target sessions with an exact-match "=" prefix. tmux does prefix/fnmatch on bare# targets, so "-t myproj-assistant" would match "myproj-assistant3" — capturing or killing the# wrong session. "=name" forces an exact match.
def _run(cmd): """subprocess.run wrapper that never raises if the binary (e.g. tmux) is absent.""" try: return subprocess.run(cmd, capture_output=True, text=True) except FileNotFoundError: return subprocess.CompletedProcess(cmd, 127, "", "")
def TS(name): # exact target-SESSION (has-session, kill-session) return "=" + name
def TP(name): # exact target-PANE: "=session:" anchors the exact session, current window/pane return "=" + name + ":"
def session_alive(name): return _run(["tmux", "has-session", "-t", TS(name)]).returncode == 0
def session_command(name): r = _run(["tmux", "display-message", "-p", "-t", TP(name), "#{pane_current_command}"]) return r.stdout.strip() if r.returncode == 0 else ""
def kill_session(name): _run(["tmux", "kill-session", "-t", TS(name)])
def capture_pane(name, lines=40): r = _run(["tmux", "capture-pane", "-p", "-t", TP(name)]) return "\n".join(r.stdout.splitlines()[-lines:]) if r.returncode == 0 else ""
def pipe_to_log(session, log_path): _run(["tmux", "pipe-pane", "-o", "-t", TP(session), f"cat >> '{log_path}'"])
def new_session(session, cwd, cmd, log_path): Path(cwd).mkdir(parents=True, exist_ok=True) _run(["tmux", "new-session", "-d", "-s", session, "-c", cwd, cmd]) pipe_to_log(session, log_path)
def ping_session(session, msg, submit_key="Enter"): """Type a message into a session and submit it; retry submit until the prefix clears.""" if not session_alive(session): return prefix = msg[:28] _run(["tmux", "send-keys", "-t", TP(session), "-l", "--", msg]) time.sleep(0.5) for _ in range(10): _run(["tmux", "send-keys", "-t", TP(session), submit_key]) time.sleep(1) if prefix not in capture_pane(session, 20): return
# ── activity / limit / fatal detection (per-backend regexes from config) ─────────
def _re(backend, key): pat = backend.get(key) return re.compile(pat, re.I) if pat else None
def _session_log_path(cfg, session): return Path(cfg["log_dir"]) / f"{session}.log"
def _log_recently_touched(cfg, session, age_seconds): try: return (time.time() - _session_log_path(cfg, session).stat().st_mtime) <= age_seconds except FileNotFoundError: return False
def pane_active(cfg, agent, pane, *, use_log=True): """True when the pane shows the agent is working. A footer_ui backend (a TUI with a static footer that lingers after a turn) only counts the bottom rows as activity, and falls back to a recently-touched session log within a grace window.""" backend = backend_of(cfg, agent) active = _re(backend, "active_re") if backend.get("footer_ui"): bottom = "\n".join(pane.splitlines()[-10:]) hit = bool(active and active.search(bottom)) grace = int(backend.get("log_grace", 180)) return hit or (use_log and _log_recently_touched(cfg, agent["session"], grace)) return bool(active and active.search(pane))
# ── prompt assembly ───────────────────────────────────────────────────────────
DONE_PLACEHOLDER_RE = re.compile( r"^\s*(not yet|not done|not complete|incomplete|pending\b|tbd\b|n/?a\b|" r"written here only|only when|to be (written|filled)|when all|<.*>)", re.I)
def phases(cfg): return cfg["loop"].get("phases", [])def phase_idx_file(cfg): return os.path.join(cfg["state_dir"], cfg["loop"].get("state_file", "phase-idx"))
def cur_idx(cfg): try: v = Path(phase_idx_file(cfg)).read_text().strip() return int(v) if v.lstrip("-").isdigit() else 0 except FileNotFoundError: return 0
def cur_phase(cfg): ps = phases(cfg) return ps[cur_idx(cfg)] if ps else {}
def _state_subdir(cfg): return (cfg["loop"].get("handoff") or {}).get("state_subdir", "machine-docs")
def handoff_repo(cfg): h = cfg["loop"].get("handoff") or {} repo = h.get("repo") return str(_resolve(cfg["project_dir"], repo)) if repo else cfg["project_dir"]
def resolve_state_file(cfg, repo_dir, basename): sub = _state_subdir(cfg) p = Path(repo_dir) / sub / basename return p if p.exists() else Path(repo_dir) / basename
def phase_done(cfg, status_basename): repo = handoff_repo(cfg) try: lines = resolve_state_file(cfg, repo, status_basename).read_text().splitlines() except FileNotFoundError: return False marker = cfg["loop"].get("done_marker", "## DONE") for i, line in enumerate(lines): if not line.startswith(marker): continue body = next((nxt for nxt in lines[i+1:] if nxt.strip()), "") if DONE_PLACEHOLDER_RE.match(body): continue return True return False
def role_model(cfg, agent): """Per-phase override (phases[idx].models[role]) wins, else the agent's configured model.""" role = agent.get("role") if role: ov = (cur_phase(cfg).get("models") or {}).get(role) if ov: return ov return agent.get("model", "")
def _render_template(text, fields): for k, v in fields.items(): text = text.replace("{" + k + "}", str(v)) return text
def build_loop_kickoff(cfg, agent): """A loop agent's kickoff = the project's kickoff template (slots filled from the current phase) followed by the role prompt prompts/<role>.md. Both files are project-supplied; this code holds no project text.""" ph = cur_phase(cfg) fields = { "phase_id": ph.get("id", ""), "plan": ph.get("plan", ""), "status": ph.get("status", ""), "role": agent.get("role", ""), } pdir = Path(cfg["project_dir"]) preamble = "" tmpl = cfg["loop"].get("kickoff_template") if tmpl: preamble = _render_template(_resolve(pdir, tmpl).read_text(), fields) roles_dir = cfg["loop"].get("roles_dir", "prompts") role_prompt = (_resolve(pdir, roles_dir) / f"{agent['role']}.md").read_text() return preamble + role_prompt
def agent_prompt(cfg, agent): pdir = Path(cfg["project_dir"]) if agent["kind"] == "loop": return build_loop_kickoff(cfg, agent) if agent.get("prompt_file"): return _resolve(pdir, agent["prompt_file"]).read_text() return agent.get("prompt", "")
# ── resume id ───────────────────────────────────────────────────────────────────
def resume_id(cfg, agent): f = Path(cfg["state_dir"]) / f"{agent['name']}.id" if f.exists(): v = f.read_text().strip() if v: return v return None
# ── agent launch ────────────────────────────────────────────────────────────────
def _expected_proc(backend): """The process name a healthy session should be running; used for backend-mismatch healing. Only backends that declare process_name participate (so a generic exec backend, or the transient login shell during startup, is never mistaken for a mismatch).""" return backend.get("process_name")
def start_agent(cfg, agent, *, force=False): session = agent["session"] if session_alive(session): if not force: log(f"{session} already running — leaving it") return kill_session(session)
backend = backend_of(cfg, agent) model = role_model(cfg, agent) prompt = agent_prompt(cfg, agent) log_path = str(_session_log_path(cfg, session)) kf = Path(cfg["state_dir"]) / f"kickoff-{session}.txt" kf.write_text(prompt) cwd = agent.get("dir") or cfg["project_dir"] pid = cur_phase(cfg).get("id", "-") if agent["kind"] == "loop" else "-" delivery = backend.get("prompt_delivery", "arg")
if delivery == "ping": # TUI backend: launch, wait for it to connect, then type the prompt in. model_env = (f"OPENCODE_CONFIG_CONTENT={shlex.quote(json.dumps({'model': model}))} " if model and backend.get("model_env") else "") preamble = backend.get("preamble", "") sep = "; " if preamble else "" attach = _render_template(backend.get("attach", "{bin}"), {"bin": backend["bin"], "server": backend.get("server", ""), "dir": shlex.quote(cwd)}) cmd = f"{preamble}{sep}{model_env}{attach}" log(f"starting {session} ({agent['backend']}, kind={agent['kind']}, phase={pid}, " f"model={model or 'default'})") new_session(session, cwd, cmd, log_path) time.sleep(int(backend.get("connect_delay", 12))) boot = (f"Your full kickoff prompt is in {kf} — read it now with: " f"`cat '{kf}'` — then follow it exactly.") ping_session(session, boot, submit_key=backend.get("submit_key", "C-m"))
elif delivery == "exec": # Generic backend: run an arbitrary command. {kickoff} = path to the prompt file, # {session} = the tmux session name, {model} = resolved model. cmd = _render_template(backend["bin"], {"kickoff": str(kf), "session": session, "model": model}) log(f"starting {session} ({agent['backend']}, kind={agent['kind']}, phase={pid})") new_session(session, cwd, cmd, log_path)
else: # "arg": prompt passed as a CLI argument (claude-style) rid = resume_id(cfg, agent) if agent.get("resume") and backend.get("supports_resume") else None parts = [backend["bin"]] if rid: parts.append(_render_template(backend.get("resume_flag", "--resume '{id}'"), {"id": rid})) if backend.get("remote_control"): parts.append(_render_template(backend.get("remote_control_flag", "--remote-control '{session}'"), {"session": session})) if model: parts.append(_render_template(backend.get("model_flag", "--model '{model}'"), {"model": model})) if backend.get("flags"): parts.append(backend["flags"]) parts.append(f"\"$(cat '{kf}')\"") cmd = " ".join(p for p in parts if p) log(f"starting {session} ({agent['backend']}, kind={agent['kind']}, phase={pid}, " f"model={model or 'default'}{', resume' if rid else ''})") new_session(session, cwd, cmd, log_path)
def start_service(cfg, svc): session = svc["session"] if session_alive(session): log(f"{session} already running — leaving it") return log(f"starting service {session}") new_session(session, svc.get("dir", cfg["project_dir"]), svc["command"], str(_session_log_path(cfg, session)))
# ── usage-limit state machine ────────────────────────────────────────────────────
RESET_RE = re.compile(r"resets?\s*(?:at\s*)?(\d{1,2})(?::(\d{2}))?\s*(am|pm)?", re.I)
def _limit_state_path(cfg, session): return Path(cfg["state_dir"]) / f"limited-{session}.json"def _load_limit_state(cfg, session): try: return json.loads(_limit_state_path(cfg, session).read_text()) except Exception: return Nonedef _save_limit_state(cfg, session, st): _limit_state_path(cfg, session).write_text(json.dumps(st))def _clear_limit_state(cfg, session): try: _limit_state_path(cfg, session).unlink() except FileNotFoundError: pass
def _parse_reset_epoch(pane): matches = list(RESET_RE.finditer(pane)) if not matches: return None m = matches[-1] try: hour, minute = int(m.group(1)), int(m.group(2) or 0) ampm = (m.group(3) or "").lower() if ampm == "pm" and hour != 12: hour += 12 elif ampm == "am" and hour == 12: hour = 0 if hour > 23 or minute > 59: return None cand = datetime.now().replace(hour=hour, minute=minute, second=0, microsecond=0) if cand.timestamp() <= time.time(): cand += timedelta(days=1) return cand.timestamp() except Exception: return None
def _next_limit_until(cfg, pane, now): fallback = int(cfg["watchdog"].get("limit_probe_fallback", 300)) slack = int(cfg["watchdog"].get("limit_reset_slack", 45)) parsed = _parse_reset_epoch(pane) if parsed is not None and parsed - now <= 6 * 3600: return parsed + slack, True return now + fallback, False
def _limit_nudge_msg(kind): if kind in ("persistent", "orchestrator"): return ("watchdog probe: if the quota window has reset, RESUME now — re-check status " "and continue from where you stopped.") return ("watchdog probe: if the quota window has reset, RESUME your loop now — pull latest, " "re-read your phase STATUS/REVIEW files, and continue; re-arm your loop pacing.")
def limit_tick(cfg, agent, pane): """True while the agent is inside a usage-limit window — callers suppress all healing.""" session = agent["session"] backend = backend_of(cfg, agent) limit_re = _re(backend, "limit_re") submit = backend.get("submit_key", "Enter") fallback = int(cfg["watchdog"].get("limit_probe_fallback", 300)) state = _load_limit_state(cfg, session) limited_now = bool(limit_re and limit_re.search(pane))
if state is None: if not limited_now or pane_active(cfg, agent, pane, use_log=False): return False now = time.time() until, parsed = _next_limit_until(cfg, pane, now) if parsed: log(f"limit hit on {agent['name']} — banner says reset " f"{datetime.fromtimestamp(until):%a %H:%M}; holding (no reboots)") else: log(f"limit hit on {agent['name']} — reset unparsable; flat " f"{fallback//60}-min probe loop (no reboots)") _save_limit_state(cfg, session, {"until": until, "nudges": 0}) return True
if pane_active(cfg, agent, pane, use_log=False) or not limited_now: log(f"limit lifted on {agent['name']} — clearing limit state") _clear_limit_state(cfg, session) return False
now = time.time() if now < state.get("until", 0): return True
msg = _limit_nudge_msg(agent["kind"]) if msg[:28] in "\n".join(pane.splitlines()[-8:]): return True nudges = state.get("nudges", 0) + 1 log(f"limit probe #{nudges} on {agent['name']} — nudging to resume") ping_session(session, msg, submit_key=submit) time.sleep(3) pane2 = capture_pane(session, 40) if pane_active(cfg, agent, pane2, use_log=False) and not (limit_re and limit_re.search(pane2)): log(f"limit lifted on {agent['name']} — probe resumed it") _clear_limit_state(cfg, session) return True until, _ = _next_limit_until(cfg, pane2, now) if nudges == 3: log(f"WARNING: {agent['name']} still limited after {nudges} probes — flat probes; never rebooting") _save_limit_state(cfg, session, {"until": until, "nudges": nudges}) return True
# ── stall detection ──────────────────────────────────────────────────────────────
_idle_since: dict[str, float] = {}
def _last_nonempty_line(text): for line in reversed(text.splitlines()): if line.strip(): return line.strip() return ""
def _parse_waiting_until(cfg, agent, pane): if backend_of(cfg, agent).get("footer_ui"): line = _last_nonempty_line(pane) if not line.startswith("WAITING-UNTIL:"): return None m = re.search(r"WAITING-UNTIL:\s*(\S+)", line) else: m = re.search(r"WAITING-UNTIL:\s*(\S+)", pane) if not m: return None try: return datetime.fromisoformat(m.group(1).replace("Z", "+00:00")).timestamp() except Exception: return None
def stall_check_one(cfg, agent): session = agent["session"] if not session_alive(session): _idle_since[session] = 0.0 _clear_limit_state(cfg, session) return now = time.time() pane = capture_pane(session, 40) if limit_tick(cfg, agent, pane): _idle_since[session] = 0.0 return if pane_active(cfg, agent, pane): _idle_since[session] = 0.0 return since = _idle_since.get(session) or now _idle_since[session] = since idle = now - since grace = int(cfg["watchdog"].get("stall_grace", 180)) until = _parse_waiting_until(cfg, agent, pane) if until is not None: if now <= until + grace: return reason = f"past its WAITING-UNTIL by {int(now-until)}s — self-wake did not fire" else: stall_idle = int(backend_of(cfg, agent).get("stall_idle", 300)) if idle < stall_idle: return reason = f"idle {int(idle)}s with no WAITING-UNTIL marker" log(f"stall: {agent['name']} ({session}) {reason} — kill + reboot") start_agent(cfg, agent, force=True) _idle_since[session] = 0.0
# ── healing ──────────────────────────────────────────────────────────────────────
def backend_mismatch(cfg, agent): expected = _expected_proc(backend_of(cfg, agent)) if not expected: return False cmd = session_command(agent["session"]) known = {_expected_proc(b) for b in cfg["backends"].values() if _expected_proc(b)} # only a definite OTHER declared backend is a mismatch; a transient login shell during # startup (not a known backend process) is not. if cmd not in known: return False return cmd != expected
def heal_one(cfg, agent): session = agent["session"] backend = backend_of(cfg, agent) if not session_alive(session): log(f"{agent['name']} ({session}) gone — restarting") start_agent(cfg, agent) return if backend_mismatch(cfg, agent): log(f"{agent['name']} ({session}) is {session_command(session)!r}, expected " f"{_expected_proc(backend)} — kill + restart") start_agent(cfg, agent, force=True) return pane = capture_pane(session, 25) if pane_active(cfg, agent, pane): return if limit_tick(cfg, agent, pane): return fatal = _re(backend, "fatal_re") if fatal and fatal.search(pane): log(f"FATAL session-state error on {agent['name']} ({session}) — kill + restart") start_agent(cfg, agent, force=True)
# ── wake (persistent agents with a wake schedule) ────────────────────────────────
def wake_agent(cfg, agent): """Returns True when the wake landed (or is moot), False to retry next tick.""" wake = agent.get("wake") if not wake: return True session = agent["session"] if not session_alive(session): return False backend = backend_of(cfg, agent) if pane_active(cfg, agent, capture_pane(session, 25)): return False pf = wake.get("prompt_file") try: msg = " ".join((_resolve(cfg["project_dir"], pf)).read_text().split()) except (FileNotFoundError, TypeError): log(f"wake skipped for {agent['name']} — prompt file missing: {pf}") return True if not msg: return True log(f"waking {agent['name']} ({session}) for scheduled supervision pass") ping_session(session, msg, submit_key=backend.get("submit_key", "Enter")) return True
# ── handoff signalling (loop pair) ───────────────────────────────────────────────
_hand = {"sha": "", "adv_inbox": "", "builder_inbox": ""}
def handoff_reset(): _hand["sha"] = _hand["adv_inbox"] = _hand["builder_inbox"] = ""
def _git(repo, args): return subprocess.run(f"git -C {repo!r} {args}", shell=True, capture_output=True, text=True)
def _show_pushed(cfg, repo, path): sub = _state_subdir(cfg) for loc in (f"origin/main:{sub}/{path}", f"origin/main:{path}"): r = _git(repo, f"show {loc!r}") if r.returncode == 0: return r.stdout return ""
def _ping_agents(cfg, value, default, msg): """Ping one or more agents. `value` is an agent name, a LIST of names, or falsy (→ default). Each target is pinged in its own session with its own backend's submit key — so a handoff can notify multiple reviewers (e.g. claim_pings = ["correctness-adversary", "readability-adversary"]).""" names = value if isinstance(value, list) else [value or default] for name in names: agent = cfg["agents"].get(name) session = agent["session"] if agent and agent.get("session") else (cfg["session_prefix"] + str(name)) submit = backend_of(cfg, agent).get("submit_key", "Enter") if agent else "Enter" ping_session(session, msg, submit_key=submit)
def handoff_check(cfg): h = cfg["loop"].get("handoff") if not h: return repo = handoff_repo(cfg) claim_pat = h.get("claim_pattern", "^claim") review_pat = h.get("review_pattern", "^review") _git(repo, "fetch -q origin") head = _git(repo, "rev-parse origin/main").stdout.strip() if head: if not _hand["sha"]: _hand["sha"] = head elif head != _hand["sha"]: subjects = _git(repo, f"log --format=%s {_hand['sha']}..origin/main").stdout if re.search(claim_pat, subjects, re.M | re.I): log("handoff: claim commit → pinging reviewer(s)") _ping_agents(cfg, h.get("claim_pings", "adversary"), "adversary", "watchdog ping: the other loop pushed a gate CLAIM commit. " "Pull and verify the claimed gate now.") if re.search(review_pat, subjects, re.M | re.I): log("handoff: review commit → pinging builder") _ping_agents(cfg, h.get("review_pings", "builder"), "builder", "watchdog ping: the other loop pushed a verdict/finding commit. " "Pull the review file and act.") _hand["sha"] = head inboxes = h.get("inboxes", []) md5 = lambda s: hashlib.md5(s.encode()).hexdigest() sub_dir = _state_subdir(cfg) for fname, key, target in ( (inboxes[0] if len(inboxes) > 0 else None, "adv_inbox", h.get("claim_pings", "adversary")), (inboxes[1] if len(inboxes) > 1 else None, "builder_inbox", h.get("review_pings", "builder")), ): if not fname: continue content = _show_pushed(cfg, repo, fname) if content: hh = md5(content) if hh != _hand[key]: log(f"handoff: {fname} changed → pinging {target}") _ping_agents(cfg, target, target, f"watchdog ping: the other loop pushed {sub_dir}/{fname} — pull, read it, " f"act, then delete the file (commit + push) to mark it consumed.") _hand[key] = hh else: _hand[key] = ""
# ── phase advance (loop machine) ─────────────────────────────────────────────────
def loop_agents(cfg): return [a for a in cfg["agents"].values() if a["kind"] == "loop" and a.get("enabled", True)]
def stop_loops(cfg): for a in loop_agents(cfg): if session_alive(a["session"]): log(f"killing {a['session']}") kill_session(a["session"])
def start_loops(cfg): for a in loop_agents(cfg): start_agent(cfg, a)
# ── optional per-phase token + time logging (log_tokens) ──────────────────────────# When [watchdog].log_tokens (or [loop].log_tokens) is true, the watchdog records, for each phase,# how many tokens each agent used and how long the phase took, appended to <log_dir>/token-log.jsonl.# Tokens are summed from each agent's Claude Code session transcript, attributed by working dir — so# give each agent its OWN dir for accurate per-agent numbers (the Builder/Adversary loop pair already# uses separate clones). View with: agents.py tokens.
def log_tokens_enabled(cfg): return bool(cfg.get("watchdog", {}).get("log_tokens") or cfg.get("loop", {}).get("log_tokens"))
def _transcript_dir(workdir): name = str(workdir).rstrip("/").replace("/", "-").replace(".", "-") return Path(os.path.expanduser("~/.claude/projects")) / name
def _sum_tokens(workdir): t = {"input": 0, "output": 0, "cache_create": 0, "cache_read": 0} d = _transcript_dir(workdir) if d.is_dir(): for f in d.glob("*.jsonl"): try: for line in f.open(errors="ignore"): try: o = json.loads(line) except Exception: continue if o.get("type") == "assistant": u = (o.get("message", {}) or {}).get("usage", {}) or {} t["input"] += u.get("input_tokens", 0) or 0 t["output"] += u.get("output_tokens", 0) or 0 t["cache_create"] += u.get("cache_creation_input_tokens", 0) or 0 t["cache_read"] += u.get("cache_read_input_tokens", 0) or 0 except OSError: continue t["total"] = t["input"] + t["output"] + t["cache_create"] + t["cache_read"] return t
def _token_cumulative(cfg): """Cumulative tokens per agent so far, summed from each agent's transcript dir.""" return {a["name"]: _sum_tokens(a["dir"]) for a in cfg["agents"].values()}
_TOKEN_KEYS = ("input", "output", "cache_create", "cache_read", "total")def _token_state_path(cfg): return Path(cfg["state_dir"]) / "token-phase.json"def _token_log_path(cfg): return Path(cfg["log_dir"]) / "token-log.jsonl"def _tok_delta(cur, base): return {k: cur.get(k, 0) - base.get(k, 0) for k in _TOKEN_KEYS}
def token_phase_begin(cfg, phase_id): """Set the baseline (cumulative tokens + start time) for the phase now starting. Idempotent across watchdog restarts: keeps the original baseline if already tracking this phase.""" if not log_tokens_enabled(cfg): return sf = _token_state_path(cfg) try: if json.loads(sf.read_text()).get("phase_id") == phase_id: return except Exception: pass sf.write_text(json.dumps({"phase_id": phase_id, "started": datetime.now().isoformat(timespec="seconds"), "baseline": _token_cumulative(cfg)}))
def token_phase_flush(cfg, next_phase_id): """Close the current phase: append its per-agent + total token deltas and duration to the token-log, then re-baseline for next_phase_id (or finalize tracking if None).""" if not log_tokens_enabled(cfg): return sf = _token_state_path(cfg) try: st = json.loads(sf.read_text()) except Exception: return cur = _token_cumulative(cfg) base = st.get("baseline", {}) started = st.get("started") try: dur = round((datetime.now() - datetime.fromisoformat(started)).total_seconds(), 1) except Exception: dur = None per_agent = {n: _tok_delta(cur.get(n, {}), base.get(n, {})) for n in cur} total = {k: sum(per_agent[n][k] for n in per_agent) for k in _TOKEN_KEYS} rec = {"phase_id": st.get("phase_id"), "started": started, "ended": datetime.now().isoformat(timespec="seconds"), "duration_s": dur, "agents": per_agent, "total": total} with _token_log_path(cfg).open("a") as fh: fh.write(json.dumps(rec) + "\n") parts = ", ".join(f"{n}={per_agent[n]['total']:,}" for n in per_agent) log(f"[log_tokens] phase {rec['phase_id']}: {total['total']:,} tok in {dur}s ({parts})") if next_phase_id is not None: sf.write_text(json.dumps({"phase_id": next_phase_id, "started": datetime.now().isoformat(timespec="seconds"), "baseline": cur})) else: sf.unlink(missing_ok=True)
def phase_advance_check(cfg): """On heavy tick: if the current phase is DONE, advance (or finish the sequence).
Returns True only when it actually transitions/completes THIS tick (caller skips healing that one tick). Once the sequence is already marked complete it is idempotent (returns False, no re-log, no re-stop). Appending a phase after completion clears the stale marker and resumes the loops on the new phase.""" ps = phases(cfg) if not ps or not cfg["loop"].get("auto_advance", True): return False marker = Path(cfg["log_dir"]) / "SEQUENCE-COMPLETE" idx = cur_idx(cfg) ph = ps[idx] if not phase_done(cfg, ph["status"]): return False nxt = idx + 1 if nxt < len(ps): log(f"PHASE {ph['id']} DONE — auto-transitioning to {ps[nxt]['id']}") token_phase_flush(cfg, ps[nxt]["id"]) stop_loops(cfg) Path(phase_idx_file(cfg)).write_text(str(nxt)) if marker.exists(): marker.unlink() # resuming into a (freshly-appended) phase — clear stale completion handoff_reset() start_loops(cfg) return True # last phase is DONE → sequence complete if marker.exists(): return False # already handled — idempotent (no re-log, no re-stop) log(f"PHASE SEQUENCE COMPLETE (last phase {ph['id']} DONE) — stopping loops") token_phase_flush(cfg, None) stop_loops(cfg) ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") marker.write_text(f"phase sequence complete {ts}. Loops stopped; build finished.\n") oc = cfg["loop"].get("on_complete") if oc: trig = Path(cfg["log_dir"]) / oc.get("trigger_file", ".run-on-complete") if trig.exists(): trig.unlink() runname = oc.get("run") if runname and runname in cfg["agents"]: log(f"on_complete: launching task agent {runname!r}") start_agent(cfg, cfg["agents"][runname], force=True) return True
# ── watchdog loop ────────────────────────────────────────────────────────────────
def watched(cfg): return [a for a in cfg["agents"].values() if a.get("enabled", True) and a.get("watch", "none") != "none"]
def watchdog_loop(cfg_path): cfg = load_config(cfg_path) sig = int(cfg["watchdog"].get("signal_interval", 30)) heavy = int(cfg["watchdog"].get("heavy_interval", 300)) ps = phases(cfg) log(f"watchdog up — phase={cur_phase(cfg).get('id','-')} [{cur_idx(cfg)+1}/{len(ps)}] " f"signal={sig}s heavy={heavy}s, watching: {[a['name'] for a in watched(cfg)]}") elapsed = heavy # force a heavy check on first tick wake_elapsed = {a["name"]: 0 for a in cfg["agents"].values() if a.get("wake")} if log_tokens_enabled(cfg): token_phase_begin(cfg, cur_phase(cfg).get("id")) log("[log_tokens] enabled — per-phase token + time logging to token-log.jsonl") while True: cfg = load_config(cfg_path) # re-read every tick: config is authoritative, no env drift has_loops = bool(loop_agents(cfg)) seq_done = (Path(cfg["log_dir"]) / "SEQUENCE-COMPLETE").exists()
if has_loops: handoff_check(cfg) for a in watched(cfg): if a["watch"] == "heal+stall": stall_check_one(cfg, a) else: if session_alive(a["session"]): limit_tick(cfg, a, capture_pane(a["session"], 40))
if not seq_done: for name, el in list(wake_elapsed.items()): interval = int(cfg["agents"][name]["wake"].get("interval", 3600)) if el >= interval: if wake_agent(cfg, cfg["agents"][name]): wake_elapsed[name] = 0
if elapsed >= heavy: elapsed = 0 advanced = phase_advance_check(cfg) if has_loops else False if not advanced: for a in watched(cfg): if seq_done and a["kind"] == "loop": continue heal_one(cfg, a)
time.sleep(sig) elapsed += sig for k in wake_elapsed: wake_elapsed[k] += sig
# ── CLI commands ──────────────────────────────────────────────────────────────
def start_watchdog(cfg, cfg_path): session = cfg["session_prefix"] + "watchdog" if session_alive(session): log("watchdog already running") return log("starting watchdog") script = Path(__file__).resolve() new_session(session, cfg["project_dir"], f"exec >>'{cfg['log_dir']}/{session}.log' 2>&1; " f"python3 '{script}' watchdog --config '{Path(cfg_path).resolve()}'", str(_session_log_path(cfg, session)))
def cmd_up(cfg, cfg_path, names): if cfg["loop"].get("resume_phase") is False and not Path(phase_idx_file(cfg)).exists(): Path(phase_idx_file(cfg)).write_text("0") targets = ([cfg["agents"][n] for n in names if n in cfg["agents"]] if names else [a for a in cfg["agents"].values() if a.get("enabled", True)]) for a in targets: start_agent(cfg, a) if not names: for s in cfg["services"].values(): start_service(cfg, s) start_watchdog(cfg, cfg_path) else: for n in names: if n in cfg["services"]: start_service(cfg, cfg["services"][n]) if n == "watchdog": start_watchdog(cfg, cfg_path)
def cmd_down(cfg, names): sessions = [] if names: for n in names: if n in cfg["agents"]: sessions.append(cfg["agents"][n]["session"]) elif n in cfg["services"]: sessions.append(cfg["services"][n]["session"]) elif n == "watchdog": sessions.append(cfg["session_prefix"] + "watchdog") else: sessions = [a["session"] for a in cfg["agents"].values()] sessions += [s["session"] for s in cfg["services"].values()] sessions.append(cfg["session_prefix"] + "watchdog") for s in sessions: if session_alive(s): log(f"killing {s}") kill_session(s)
def cmd_status(cfg): idx, ps = cur_idx(cfg), phases(cfg) if ps: ph = ps[idx] done = "## DONE" if phase_done(cfg, ph["status"]) else "in progress" print(f" phase: {ph['id']} [{idx+1}/{len(ps)}] plan={ph.get('plan','-')} ({done})") print(f" {'AGENT':<14} {'KIND':<11} {'BACKEND':<9} {'MODEL':<20} {'WATCH':<10} STATE") for a in cfg["agents"].values(): st = "RUNNING" if session_alive(a["session"]) else "stopped" en = "" if a.get("enabled", True) else " (disabled)" rc = session_command(a["session"]) if st == "RUNNING" else "" print(f" {a['name']:<14} {a['kind']:<11} {a['backend']:<9} " f"{role_model(cfg,a) or 'default':<20} {a.get('watch','none'):<10} {st}{en}" + (f" [{rc}]" if rc else "")) for s in cfg["services"].values(): st = "RUNNING" if session_alive(s["session"]) else "stopped" print(f" {s['name']:<14} {'service':<11} {'-':<9} {'-':<20} {'-':<10} {st}") wd = cfg["session_prefix"] + "watchdog" print(f" {'watchdog':<14} {'service':<11} {'-':<9} {'-':<20} {'-':<10} " f"{'RUNNING' if session_alive(wd) else 'stopped'}")
def cmd_phase(cfg, args): ps = phases(cfg) if not ps: print("no [loop].phases configured"); return if not args or args[0] == "show": idx = cur_idx(cfg) print(f"phase {ps[idx]['id']} [{idx+1}/{len(ps)}] seq: {' '.join(p['id'] for p in ps)}") return if args[0] == "next": Path(phase_idx_file(cfg)).write_text(str(min(cur_idx(cfg)+1, len(ps)-1))) elif args[0] == "set" and len(args) > 1: Path(phase_idx_file(cfg)).write_text(str(int(args[1]))) print(f"phase idx now {cur_idx(cfg)} ({cur_phase(cfg).get('id')})")
def cmd_tokens(cfg): """Pretty-print <log_dir>/token-log.jsonl: per-phase tokens by agent + total + duration.""" p = _token_log_path(cfg) if not p.exists(): print(f"no token log at {p}\n(set [watchdog].log_tokens = true and run the loop)"); return recs = [] for line in p.read_text().splitlines(): try: recs.append(json.loads(line)) except Exception: pass if not recs: print("token log is empty"); return names = [] for r in recs: for n in r.get("agents", {}): if n not in names: names.append(n) w = max([7] + [len(n) for n in names]) hdr = f"{'phase':<10} {'dur(s)':>8} " + " ".join(f"{n:>{w}}" for n in names) + f" {'TOTAL':>13}" print(hdr); print("-" * len(hdr)) grand = {n: 0 for n in names} durtot = 0.0 for r in recs: ag = r.get("agents", {}) cells = " ".join(f"{ag.get(n,{}).get('total',0):>{w},}" for n in names) print(f"{str(r.get('phase_id')):<10} {str(r.get('duration_s')):>8} {cells} " f"{r.get('total',{}).get('total',0):>13,}") for n in names: grand[n] += ag.get(n,{}).get("total",0) durtot += r.get("duration_s") or 0 print("-" * len(hdr)) cells = " ".join(f"{grand[n]:>{w},}" for n in names) print(f"{'TOTAL':<10} {durtot:>8.0f} {cells} {sum(grand.values()):>13,}")
def cmd_selftest(): """Self-contained regression checks for the footer-UI activity detector. Needs no config.""" backend = { "active_re": "esc interrupt|thinking|inferring|running tool|preparing patch|reading|searching", "footer_ui": True, "log_grace": 180, } cfg = {"backends": {"tui": backend}, "log_dir": "/tmp"} a = {"name": "x", "backend": "tui", "session": "selftest-x", "kind": "loop"} idle = "\n ▣ Build · GPT-5.4 · 2m 19s\n 178.4K (17%) ctrl+p commands\n" active = "\n ~ Preparing patch...\n ⬝⬝⬝■■ esc interrupt 137.6K\n" checks = [ ("footer_ui idle footer is idle", not pane_active(cfg, a, idle, use_log=False)), ("footer_ui active footer is active", pane_active(cfg, a, active, use_log=False)), ("limit banner + idle footer is not active", not pane_active(cfg, a, idle + "\nYou've hit your weekly limit · resets Jun 16, 10pm\n", use_log=False)), ] bad = [n for n, ok in checks if not ok] for n, ok in checks: print(f" {'PASS' if ok else 'FAIL'}: {n}") sys.exit(1 if bad else 0)
INIT_TOML = """\# Starter agent-orchestrator config. See the README for the full schema.[defaults]session_prefix = "{prefix}-"log_dir = ".ao-state"backend = "claude"model = "claude-sonnet-4-6"watch = "heal"
[backend.claude]bin = "claude"flags = "--dangerously-skip-permissions"prompt_delivery = "arg"process_name = "claude"submit_key = "Enter"stall_idle = 300active_re = "esc to interrupt|Running tool|\\\\u00b7 \\\\d+"limit_re = "usage limit|limit reached|reached your .*limit"
[[agent]]name = "worker"kind = "persistent"prompt = "You are a worker agent. Wait for instructions.""""
def cmd_init(args): target = Path(args[0]) if args else Path.cwd() target.mkdir(parents=True, exist_ok=True) cfg_file = target / "agents.toml" if cfg_file.exists(): die(f"{cfg_file} already exists — refusing to overwrite") cfg_file.write_text(INIT_TOML.format(prefix=target.resolve().name or "proj")) (target / "prompts").mkdir(exist_ok=True) log(f"scaffolded {cfg_file} and {target/'prompts'}/ — edit, then `agents.py up --config {cfg_file}`")
# ── main ──────────────────────────────────────────────────────────────────────
def main(): argv = sys.argv[1:] if not argv or argv[0] in ("-h", "--help", "help"): print(__doc__); return cfg_path = _cfg_path(argv) argv = [a for i, a in enumerate(argv) if a != "--config" and (i == 0 or argv[i-1] != "--config")] cmd = argv[0] if argv else "status" rest = argv[1:]
if cmd == "selftest": cmd_selftest(); return if cmd == "init": cmd_init(rest); return
cfg = load_config(cfg_path) if cmd == "up": cmd_up(cfg, cfg_path, rest) elif cmd == "down": cmd_down(cfg, rest) elif cmd == "status": cmd_status(cfg) elif cmd == "watchdog": watchdog_loop(cfg_path) elif cmd == "phase": cmd_phase(cfg, rest) elif cmd == "tokens": cmd_tokens(cfg) elif cmd == "logs": if not rest: die("usage: agents.py logs <name>") sess = cfg["agents"].get(rest[0], {}).get("session") or (cfg["session_prefix"] + rest[0]) os.execvp("tail", ["tail", "-f", str(_session_log_path(cfg, sess))]) else: print(__doc__)
if __name__ == "__main__": main()