fix(pipeline): PIPELINE-COMPLETE mirrors live stage list, not a write-once latch

pipeline_check wrote PIPELINE-COMPLETE the first tick all-then-configured stage markers existed, and
never cleared it. Appending stages to an already-completed pipeline (as the orchestrator did — adding
rust-linecov-e2e then two -realpds stages after the fork sequence finished) left a stale 'done' sentinel:
the new stages ran, but the file lied that the pipeline was complete, misleading operators reading state.

Now recompute the sentinel every tick from ALL current stages' markers: write it only when every marker
exists, and CLEAR a stale one the moment any stage is incomplete — so appending stages self-corrects.
Gated on the markers directly, not 'active is None' (which is also None when a stage names a missing agent).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UWTdUq2bsic7JZGqJp3nD6
This commit is contained in:
2026-07-07 19:25:12 +00:00
parent 08fd58ccc0
commit 234d6a054e

View File

@ -1025,11 +1025,21 @@ def pipeline_check(cfg):
_pipeline_push_on_retire(ag)
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"
# 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):