From b739504e1f2b9926cff38be632718164c180b881 Mon Sep 17 00:00:00 2001 From: mfowler Date: Sat, 4 Jul 2026 13:45:47 +0000 Subject: [PATCH] =?UTF-8?q?feat(watchdog):=20[pipeline]=20=E2=80=94=20sequ?= =?UTF-8?q?ential=20standalone-agent=20phases=20via=20completion=20markers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01UWTdUq2bsic7JZGqJp3nD6 --- agents.py | 76 +++++++++++++++++++++++++++++++++++++++++++++-- examples/IDEAS.md | 30 +++++++++++++++++++ 2 files changed, 103 insertions(+), 3 deletions(-) diff --git a/agents.py b/agents.py index 38f1195..c1751eb 100755 --- a/agents.py +++ b/agents.py @@ -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) diff --git a/examples/IDEAS.md b/examples/IDEAS.md index d1fb619..bc1bfb2 100644 --- a/examples/IDEAS.md +++ b/examples/IDEAS.md @@ -84,6 +84,36 @@ generations advance until fitness plateaus. --- +## Engine idea: unify the phase machine and per-agent pipelines (one apparatus) + +The `[loop]` **phase machine** advances an ordered plan for a *fixed* set of loop agents (Builder + +Adversary) that run **every** phase in **one** worktree, detecting completion via a `## DONE` +commit-subject in a status file. A common variant, though, is a sequence of **distinct** agents — +different prompt/model, sometimes a different worktree — handed off one at a time (e.g. build → review → +e2e → integrate, each a separate agent). That's *the same shape* — an ordered sequence of stages, each +with a completion signal and the right agent(s) running — so it shouldn't need a **separate pipeline +apparatus**; it should be **the same phase machine**, generalized. + +**The missing primitive is a per-stage `agent` + `dir`.** Agents already carry a `dir` (used for +per-agent token attribution). If a **phase entry** could optionally name the agent(s) that run it and +the directory they run in (plus a `done` marker as an alternative to `## DONE`), the one phase machine +covers both cases: +- **loop-style phase** (default/back-compat): no `agent` → the fixed loop agents run it; completion = + `## DONE` in the phase's status file, in the loop dir. +- **pipeline-style phase:** `agent = "fork-iroh"`, `dir = "work-fork"`, `done = + "work-fork/.ao-state/FORK-IROH-COMPLETE"` → advancing the phase stops the previous stage's agent and + starts this one in its dir; completion = the marker. + +Advancing a phase then means "stop the outgoing stage's agent(s), start the incoming stage's agent(s) in +their dir" — a superset of today's "re-kickoff the same loop agents." The `p-lichen-orchestrator` project +prototyped this as a standalone `[pipeline]` block (stages = `{agent, done}`, reconciled by the watchdog +each tick, markers as source of truth); the lesson is that **that logic belongs in the phase machine**, +not beside it — fold `[pipeline]` into `[loop].phases` via optional `agent`/`dir`/`done` per phase and +delete the parallel mechanism. (Bonus: `on_complete`, per-phase model overrides, and the DONE-nudge all +already live on the phase machine, so pipeline-style phases inherit them for free.) + +--- + ## Suggested next trio If picking three that cover the most new ground: **The Line** (pipeline), **The Incident Room**