recipe-maintainers/agent-orchestrator
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(watchdog): [pipeline] — sequential standalone-agent phases via completion markers

Browse Source 
Adds a [pipeline] block: an ordered sequence of DISTINCT agents (different prompt/model/dir) run one
at a time, advancing when each writes its completion marker — the standalone-agent analog of the
[loop] phase machine (which drives FIXED loop agents via ## DONE). The watchdog reconciles it every
tick, stateless (markers are the source of truth): exactly the first not-yet-complete stage runs,
earlier stages are retired, the active stage is stall/heal-watched. Pipeline agents are enabled=false
(the pipeline owns their lifecycle).

Also fixes a latent watchdog crash: wake_elapsed is built once at startup but config is re-read each
tick, so removing an agent's 'wake' mid-run (e.g. winding down a wake) hit agent['wake'] -> KeyError
and killed the whole watchdog silently (stalling phase advancement + stall recovery). Now skips agents
whose wake was removed.

IDEAS.md: note that [pipeline] and [loop].phases are the same shape and should be unified via an
optional per-phase agent/dir/done (fold pipeline into the phase machine, delete the parallel path).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UWTdUq2bsic7JZGqJp3nD6
This commit is contained in:
mfowler
2026-07-04 13:45:47 +00:00
parent 08bbb60343
commit b739504e1f
2 changed files with 103 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
agents.py
View File

@ -66,6 +66,7 @@ def load_config(path):
"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,
@ -951,6 +952,63 @@ 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_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:
log(f"pipeline: retire stage {ag['name']!r} (complete or not yet active)")
kill_session(ag["session"])
if active is None:
marker = Path(cfg["log_dir"]) / "PIPELINE-COMPLETE"
if not marker.exists():
log("pipeline: ALL STAGES COMPLETE")
marker.write_text("pipeline complete\n")
return active
def watchdog_loop(cfg_path):
cfg = load_config(cfg_path)
sig = int(cfg["watchdog"].get("signal_interval", 30))
@ -971,7 +1029,13 @@ def watchdog_loop(cfg_path):
if has_loops and not seq_done:
handoff_check(cfg)
gate_token_check(cfg)
for a in watched(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:
@ -979,7 +1043,13 @@ def watchdog_loop(cfg_path):
limit_tick(cfg, a, capture_pane(a["session"], 40))
for name, el in list(wake_elapsed.items()):
agent = cfg["agents"][name]
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).
@ -997,7 +1067,7 @@ def watchdog_loop(cfg_path):
if elapsed >= heavy:
elapsed = 0
if not advanced:
for a in watched(cfg):
for a in watch_list:
if seq_done and a["kind"] == "loop":
continue
heal_one(cfg, a)