notplants
b73af35792
Found in production: rust-mutation emitted WAITING-UNTIL 3h11m out for a cargo-mutants run over the whole workspace (a genuinely multi-hour job, cargo running the whole time), but the marker branch skipped the build-aware check entirely and went straight to the cap — so the 7200s cap would have killed a live run and thrown away hours of work. The cap exists to catch a session that PARKED itself and is stuck; a build still running under the session is proof it is not. Now the cap only fires when idle exceeds it AND no build is running. The stated deadline remains the hard bound either way, so a runaway still cannot park forever. Tests: cap-reboots-when-no-build, cap-yields-to-a-live-build, past-deadline-reboots-even-with-a-build. 68 pass.
1384 lines
64 KiB
Python
Executable File
1384 lines
64 KiB
Python
Executable File
#!/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. Every
|
|
agent — a Builder/Adversary loop pair, a persistent supervisor, a one-shot task — is declared
|
|
in 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, the
|
|
handoff 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, tomllib
|
|
from datetime import datetime, timedelta
|
|
from 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", {}),
|
|
"pipeline": raw.get("pipeline", {}),
|
|
"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 None
|
|
def _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] = {}
|
|
_done_nudged: dict[str, bool] = {} # per-session: sent the one-time "write the done marker" nudge this phase
|
|
_build_deferred: set = set() # per-session: currently deferring a stall because a real build is running
|
|
|
|
# Default set of process names (comm) that mean "genuine long-running work is happening" — a compile,
|
|
# coverage run, mutation run, or a real e2e test driving the server/browser. If one of these is a
|
|
# descendant of the agent's tmux pane, a silent pane is a running build, NOT a stall. High-signal names
|
|
# only (no bare python/node/bash — those match the orchestrator's own engine + would false-positive).
|
|
DEFAULT_BUILD_PROCS_RE = (r"^(cargo|cargo-llvm-cov|cargo-mutants|cargo-nextest|nextest|rustc|rustdoc|"
|
|
r"cc1|cc1plus|collect2|lld|ld\.lld|llvm-cov|llvm-profdata|"
|
|
r"lichen-server|lichen-cms|lichen-shell|lichen-cli|chromium|chrome|playwright)$")
|
|
|
|
def _proc_descendants(roots):
|
|
"""All descendant PIDs of the given root PIDs (BFS via pgrep -P), EXCLUDING the roots themselves —
|
|
the pane root is the claude process whose args embed the prompt (which mentions cargo/rustc/…), so
|
|
matching it would false-positive; we only inspect its real child processes."""
|
|
roots = [p for p in roots if str(p).isdigit()]
|
|
seen, stack = set(), list(roots)
|
|
while stack:
|
|
pid = stack.pop()
|
|
if pid in seen:
|
|
continue
|
|
seen.add(pid)
|
|
c = subprocess.run(f"pgrep -P {pid}", shell=True, capture_output=True, text=True)
|
|
stack += [p for p in c.stdout.split() if p.isdigit()]
|
|
return seen - set(roots)
|
|
|
|
def _build_running(cfg, agent):
|
|
"""True if a real build/coverage/test process is running under the agent's tmux session, so a long
|
|
silent pane isn't mistaken for a stall. Matches process comm (never the claude root's args). Bounded
|
|
by stall_idle_max upstream so a genuinely hung build still gets rebooted eventually."""
|
|
session = agent["session"]
|
|
r = subprocess.run(f"tmux list-panes -t {session!r} -F '#{{pane_pid}}'",
|
|
shell=True, capture_output=True, text=True)
|
|
kids = _proc_descendants(r.stdout.split())
|
|
if not kids:
|
|
return False
|
|
try:
|
|
rx = re.compile(cfg["watchdog"].get("build_procs_re", DEFAULT_BUILD_PROCS_RE))
|
|
except re.error:
|
|
rx = re.compile(DEFAULT_BUILD_PROCS_RE)
|
|
ps = subprocess.run("ps -o comm= -p " + ",".join(sorted(kids)),
|
|
shell=True, capture_output=True, text=True)
|
|
return any(rx.match(c.strip()) for c in ps.stdout.splitlines() if c.strip())
|
|
|
|
def _done_nudge_msg(cfg, ph):
|
|
"""The DONE-nudge: prompts a stalled loop agent to finalize a built-but-unmarked phase."""
|
|
dm = cfg["loop"].get("done_marker", "## DONE")
|
|
pid = ph.get("id", "")
|
|
status = ph.get("status", f"STATUS-{pid}.md")
|
|
sub = _state_subdir(cfg)
|
|
return (f"watchdog nudge: you've stalled in phase '{pid}', which is NOT yet marked '{dm}'. Resume now "
|
|
f"— pull any pending review/inbox and continue. If (and ONLY if) every DoD item has a fresh "
|
|
f"PASS from BOTH adversaries with no standing veto, write '{dm}' to {sub}/{status} and push, "
|
|
f"so the phase settles and auto-advances. Do not stay idle.")
|
|
|
|
def _pane_last_active(session):
|
|
"""Unix timestamp of the tmux window's last activity (last output change), or None.
|
|
Seeds idle-duration from the agent's REAL last activity rather than `now`, so stalls are
|
|
detected regardless of when the watchdog process started — restarting the watchdog no longer
|
|
resets every agent's stall clock."""
|
|
r = subprocess.run(f"tmux display-message -p -t {session!r} '#{{window_activity}}'",
|
|
shell=True, capture_output=True, text=True)
|
|
try:
|
|
return float(r.stdout.strip())
|
|
except (ValueError, AttributeError):
|
|
return None
|
|
|
|
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):
|
|
# Only consulted once the pane is already idle (see stall_check_one), so scanning the whole
|
|
# capture and taking the MOST-RECENT marker is safe — and it's the only thing that works for a
|
|
# footer_ui backend (claude/opencode), whose input-box footer always renders BELOW the agent's
|
|
# final message. The footer never contains the marker, so the last match is the agent's own
|
|
# signal, whether or not a status footer follows it.
|
|
matches = re.findall(r"WAITING-UNTIL:\s*(\S+)", pane)
|
|
if not matches:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(matches[-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
|
|
_build_deferred.discard(session)
|
|
return
|
|
# Seed from the pane's real last-activity (not `now`), so a watchdog that just (re)started still
|
|
# sees an already-idle pane as idle-for-its-true-duration instead of resetting the clock.
|
|
since = _idle_since.get(session) or _pane_last_active(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:
|
|
# An agent that starts a long remote/async run (remote cargo build, terraform apply, VM
|
|
# provision, long ssh) prints `WAITING-UNTIL: <ISO8601>` so the watchdog holds off instead
|
|
# of killing it mid-run. Cap how far out it can push its own reboot, so a runaway can't park
|
|
# itself forever ("some max no matter what").
|
|
wu_max = int(cfg["watchdog"].get("waiting_until_max", 7200))
|
|
# The cap guards against a session that parked itself and is genuinely stuck. A build still
|
|
# running under the session (cargo-mutants / a coverage run / a remote ssh) is proof of life —
|
|
# those legitimately run for hours — so it defers to the agent's stated deadline instead of
|
|
# being guillotined by the cap. Past the deadline (+grace) it reboots either way, so a runaway
|
|
# can never park forever.
|
|
if wu_max and idle > wu_max and not _build_running(cfg, agent):
|
|
reason = (f"WAITING-UNTIL exceeded the {wu_max}s cap (idle {int(idle)}s, no build running) "
|
|
f"— rebooting regardless")
|
|
elif now <= until + grace:
|
|
return
|
|
else:
|
|
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
|
|
# Build-aware: a silent pane with a real compile/coverage/test process running is NOT a stall —
|
|
# defer the reboot until it finishes, but never past stall_idle_max (hard cap for a hung build).
|
|
stall_idle_max = int(backend_of(cfg, agent).get("stall_idle_max", 1800))
|
|
if idle < stall_idle_max and _build_running(cfg, agent):
|
|
if session not in _build_deferred:
|
|
log(f"stall-defer: {agent['name']} ({session}) idle {int(idle)}s but a build/test is "
|
|
f"running — waiting (hard cap {stall_idle_max}s)")
|
|
_build_deferred.add(session)
|
|
return
|
|
reason = (f"idle {int(idle)}s past build-aware hard cap {stall_idle_max}s — rebooting regardless"
|
|
if idle >= stall_idle_max else
|
|
f"idle {int(idle)}s with no WAITING-UNTIL marker and no build running")
|
|
# Ceremony-lag guard: a loop agent idling in a phase that's built but NOT marked done won't let the
|
|
# phase advance (the recurring "all gates PASS but no ## DONE written" stall). Nudge it ONCE per phase
|
|
# to finalize (write the done marker if the DoD is met) before falling back to the blunt kill+reboot.
|
|
if (cfg["loop"].get("done_nudge", True) and agent.get("kind") == "loop" and phases(cfg)
|
|
and not phase_done(cfg, cur_phase(cfg).get("status", "")) and not _done_nudged.get(session)):
|
|
log(f"stall: {agent['name']} ({session}) {reason} — DONE-nudge (phase built but not marked done)")
|
|
ping_session(session, _done_nudge_msg(cfg, cur_phase(cfg)),
|
|
submit_key=backend_of(cfg, agent).get("submit_key", "Enter"))
|
|
_done_nudged[session] = True
|
|
_idle_since[session] = now # fresh idle window to act on the nudge before reboot escalates
|
|
return
|
|
log(f"stall: {agent['name']} ({session}) {reason} — kill + reboot")
|
|
start_agent(cfg, agent, force=True)
|
|
_idle_since[session] = 0.0
|
|
_build_deferred.discard(session)
|
|
|
|
# ── 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"]
|
|
# A one-shot `task` is "woken" by RE-RUNNING it fresh — it has no persistent REPL to re-prompt — so
|
|
# scheduled work (e.g. a coverage audit) recurs autonomously on its interval, no operator needed.
|
|
# Skip only while its previous run is still going; otherwise kill + restart for a clean re-run.
|
|
if agent.get("kind") == "task":
|
|
if session_alive(session) and pane_active(cfg, agent, capture_pane(session, 25)):
|
|
return False
|
|
log(f"wake: re-running task {agent['name']} ({session})")
|
|
start_agent(cfg, agent, force=True)
|
|
return True
|
|
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)
|
|
|
|
# ── token logging granularity: per phase, or also per GATE (log_tokens) ────────────
|
|
# Phases are tracked per phase (token_phase_begin/flush). With token_granularity="gate" (the default)
|
|
# tokens are ALSO attributed to each gate. A "gate" is a claimed unit — any `claim(<label>)` commit on
|
|
# the work repo's origin/main (e.g. claim(D1-D5), claim(feat:multi-file)); a leading "feat:" is
|
|
# stripped for readability. A change in the most-recently-claimed label is a boundary; on each
|
|
# boundary the previous gate's per-agent token delta + duration is appended to token-log.jsonl tagged
|
|
# phase_id="<phase>:<label>", so `agents.py tokens` lists it as its own row. The per-phase rollup
|
|
# record is written either way; "phase" granularity logs only that.
|
|
_gate_claim_re = re.compile(r"^claim\(\s*([^)]+?)\s*\)", re.I)
|
|
|
|
def token_granularity(cfg):
|
|
"""'gate' (default; per claimed gate, plus the per-phase rollup) or 'phase' (per phase only)."""
|
|
g = (cfg.get("watchdog", {}).get("token_granularity")
|
|
or cfg.get("loop", {}).get("token_granularity") or "gate")
|
|
return g if g in ("gate", "phase") else "gate"
|
|
|
|
def _token_gate_state_path(cfg):
|
|
return Path(cfg["state_dir"]) / "token-gate.json"
|
|
|
|
def _latest_claimed_gate(cfg):
|
|
"""Label of the most-recent `claim(<label>)` subject on the work repo's origin/main, or None.
|
|
A leading 'feat:' is stripped so feature gates read as their bare name."""
|
|
r = _git(handoff_repo(cfg), "log -1 --format=%s --grep 'claim(' origin/main")
|
|
m = _gate_claim_re.match((r.stdout or "").strip())
|
|
if not m:
|
|
return None
|
|
label = m.group(1).strip()
|
|
return label[5:].strip() if label.lower().startswith("feat:") else label
|
|
|
|
def _write_token_delta(cfg, phase_id, st):
|
|
"""Append one per-agent token delta record (vs the baseline in st) to token-log.jsonl."""
|
|
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": 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")
|
|
return rec, per_agent, total
|
|
|
|
def gate_token_flush(cfg):
|
|
"""Close out the currently-tracked gate (if any): write its token delta, then drop the state."""
|
|
sf = _token_gate_state_path(cfg)
|
|
try:
|
|
st = json.loads(sf.read_text())
|
|
except Exception:
|
|
return
|
|
gate = st.get("gate")
|
|
if not gate:
|
|
return
|
|
phase_id = f"{st['phase']}:{gate}" if st.get("phase") else gate
|
|
rec, per_agent, total = _write_token_delta(cfg, phase_id, st)
|
|
parts = ", ".join(f"{n}={per_agent[n]['total']:,}" for n in per_agent)
|
|
log(f"[log_tokens] gate {phase_id}: {total['total']:,} tok in {rec['duration_s']}s ({parts})")
|
|
try:
|
|
sf.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
def gate_token_check(cfg):
|
|
"""When token_granularity=='gate', detect gate boundaries and flush per-gate token deltas."""
|
|
if not log_tokens_enabled(cfg) or token_granularity(cfg) != "gate":
|
|
return
|
|
current = _latest_claimed_gate(cfg)
|
|
if not current:
|
|
return
|
|
sf = _token_gate_state_path(cfg)
|
|
try:
|
|
tracked = json.loads(sf.read_text()).get("gate")
|
|
except Exception:
|
|
tracked = None
|
|
if tracked == current:
|
|
return
|
|
if tracked:
|
|
gate_token_flush(cfg) # close out the previous gate before starting the next
|
|
sf.write_text(json.dumps({"gate": current, "phase": cur_phase(cfg).get("id"),
|
|
"started": datetime.now().isoformat(timespec="seconds"),
|
|
"baseline": _token_cumulative(cfg)}))
|
|
log(f"[log_tokens] tracking gate: {current}")
|
|
|
|
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
|
|
if log_tokens_enabled(cfg) and token_granularity(cfg) == "gate":
|
|
gate_token_flush(cfg) # close out the last in-flight gate before leaving the phase
|
|
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()
|
|
_done_nudged.clear() # fresh DONE-nudge budget for the new phase
|
|
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"]
|
|
|
|
# ── standalone-agent pipeline (sequential phases via completion markers) ──────────
|
|
# A [pipeline] block runs a sequence of DISTINCT agents (different prompts / models / dirs) one at a
|
|
# time, advancing when each writes its completion marker — the standalone-agent analog of the loop
|
|
# phase machine. Stateless: the markers (which agents write in their OWN cwd, e.g. work-fork/.ao-state/)
|
|
# are the source of truth, so the watchdog reconciles the desired state every tick — exactly one stage
|
|
# runs, earlier (done) stages are retired, later stages wait. Declare pipeline agents enabled=false: the
|
|
# pipeline (not `up`/watched()) owns their lifecycle; the ACTIVE stage is still stall/heal-watched.
|
|
#
|
|
# [pipeline]
|
|
# enabled = true
|
|
# stages = [
|
|
# { agent = "fork-e2e", done = "work-fork/.ao-state/FORK-E2E-COMPLETE" },
|
|
# { agent = "fork-iroh", done = "work-fork/.ao-state/FORK-IROH-COMPLETE" },
|
|
# ]
|
|
|
|
def pipeline_stages(cfg):
|
|
pl = cfg.get("pipeline") or {}
|
|
if not pl.get("enabled", True):
|
|
return []
|
|
return pl.get("stages", [])
|
|
|
|
def _pipeline_marker(cfg, stage):
|
|
p = Path(stage["done"])
|
|
return p if p.is_absolute() else Path(cfg["project_dir"]) / stage["done"]
|
|
|
|
def _pipeline_push_on_retire(agent):
|
|
"""Best-effort `git push` in a completing stage's dir before it's retired, so its final commits
|
|
aren't stranded locally if it wrote its completion marker before its last push landed (a race:
|
|
the pipeline retires a stage the instant its marker appears). Never raises; logs the outcome."""
|
|
d = agent.get("dir")
|
|
if not d or not (Path(d) / ".git").exists():
|
|
return
|
|
try:
|
|
r = subprocess.run(["git", "-C", d, "push", "origin", "HEAD:main"],
|
|
env={**os.environ, "GIT_SSH_COMMAND": "ssh -o BatchMode=yes"},
|
|
capture_output=True, text=True, timeout=90)
|
|
if r.returncode == 0:
|
|
log(f"pipeline: pushed {agent['name']!r} before retiring (no stranded commits)")
|
|
elif "up-to-date" not in (r.stderr + r.stdout).lower():
|
|
log(f"pipeline: push-on-retire for {agent['name']!r} nonzero (non-fatal): "
|
|
f"{(r.stderr or r.stdout).strip().splitlines()[-1:] }")
|
|
except Exception as e:
|
|
log(f"pipeline: push-on-retire for {agent['name']!r} failed (non-fatal): {e}")
|
|
|
|
def pipeline_check(cfg):
|
|
"""Reconcile the [pipeline]: run the first stage whose completion marker is absent, retire every
|
|
other pipeline agent. Returns the ACTIVE stage's agent dict (so the watchdog stall/heal-watches it),
|
|
or None if there's no pipeline or it's fully complete."""
|
|
stages = pipeline_stages(cfg)
|
|
if not stages:
|
|
return None
|
|
active = None
|
|
for st in stages:
|
|
if not _pipeline_marker(cfg, st).exists():
|
|
active = cfg["agents"].get(st.get("agent"))
|
|
break
|
|
active_session = active["session"] if active else None
|
|
for st in stages:
|
|
ag = cfg["agents"].get(st.get("agent"))
|
|
if not ag:
|
|
continue
|
|
alive = session_alive(ag["session"])
|
|
if ag["session"] == active_session:
|
|
if not alive:
|
|
log(f"pipeline: advancing → start stage {ag['name']!r}")
|
|
start_agent(cfg, ag, force=True)
|
|
elif alive:
|
|
# Retiring a non-active stage. If it's a COMPLETED stage (its marker exists), push its repo
|
|
# first so final commits written just before the marker aren't stranded locally.
|
|
if _pipeline_marker(cfg, st).exists():
|
|
_pipeline_push_on_retire(ag)
|
|
log(f"pipeline: retire stage {ag['name']!r} (complete or not yet active)")
|
|
kill_session(ag["session"])
|
|
# PIPELINE-COMPLETE must MIRROR the current stage list, not latch on first completion. Recompute it
|
|
# from every stage's marker each tick: write it only when ALL current markers exist, and CLEAR a stale
|
|
# one the moment any stage is incomplete. Without the clear, appending new stages to a pipeline that
|
|
# had already completed leaves a lying "done" sentinel (the appended stages run, but the file says the
|
|
# pipeline finished). `active is None` is not a safe proxy here — it's also None when a stage names a
|
|
# missing agent — so gate on the markers directly.
|
|
marker = Path(cfg["log_dir"]) / "PIPELINE-COMPLETE"
|
|
all_complete = all(_pipeline_marker(cfg, st).exists() for st in stages)
|
|
if all_complete:
|
|
if not marker.exists():
|
|
log("pipeline: ALL STAGES COMPLETE")
|
|
marker.write_text("pipeline complete\n")
|
|
elif marker.exists():
|
|
log("pipeline: reopened — a stage is incomplete; clearing stale PIPELINE-COMPLETE")
|
|
marker.unlink()
|
|
return active
|
|
|
|
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(f"[log_tokens] enabled (granularity={token_granularity(cfg)}) — 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 and not seq_done:
|
|
handoff_check(cfg)
|
|
gate_token_check(cfg)
|
|
# Reconcile the standalone-agent pipeline (start/advance/retire stages by their markers), and
|
|
# fold its ACTIVE stage into the stall/heal watch list so it auto-recovers like any watched agent.
|
|
active_pipe = pipeline_check(cfg)
|
|
watch_list = list(watched(cfg))
|
|
if active_pipe and active_pipe["name"] not in {x["name"] for x in watch_list}:
|
|
watch_list.append(active_pipe)
|
|
for a in watch_list:
|
|
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))
|
|
|
|
for name, el in list(wake_elapsed.items()):
|
|
agent = cfg["agents"].get(name)
|
|
# Config is re-read every tick, but wake_elapsed was built once at startup. If an agent's
|
|
# `wake` was removed (or the agent deleted) mid-run — e.g. an operator winding down a wake —
|
|
# skip it instead of KeyError-crashing the whole watchdog. (This bug once killed the watchdog
|
|
# silently, stalling phase advancement.)
|
|
if not agent or "wake" not in agent:
|
|
continue
|
|
# After the phase sequence completes, quiet the loop-tied wakes (e.g. the on-demand
|
|
# auditor) — but a PERSISTENT agent (the operator-facing supervisor) keeps waking, so its
|
|
# hourly supervision survives SEQUENCE-COMPLETE and can drive follow-on work (a second build).
|
|
if seq_done and agent.get("kind") != "persistent":
|
|
continue
|
|
interval = int(agent["wake"].get("interval", 3600))
|
|
if el >= interval:
|
|
if wake_agent(cfg, agent):
|
|
wake_elapsed[name] = 0
|
|
|
|
# Auto-advance is checked EVERY tick (not just the heavy tick) so a completed phase advances
|
|
# within signal_interval of its `## DONE` landing, instead of idling up to heavy_interval.
|
|
advanced = phase_advance_check(cfg) if has_loops else False
|
|
|
|
if elapsed >= heavy:
|
|
elapsed = 0
|
|
if not advanced:
|
|
for a in watch_list:
|
|
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 = 300
|
|
active_re = "esc to interrupt|Running tool"
|
|
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()
|