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**