diff --git a/docs/architecture/orchestrator.md b/docs/architecture/orchestrator.md index 44f80fd19..ad88524e9 100644 --- a/docs/architecture/orchestrator.md +++ b/docs/architecture/orchestrator.md @@ -126,14 +126,22 @@ The implement-phase run loop is multi-slice (see [slice-dag.md](slice-dag.md)), | # | Observed state | Action | |---|----------------|--------| -| 1 | `IN_PROGRESS`, no commits on integration branch | No-op — scheduler re-yields the slice as `READY` and a fresh team spawns | -| 2 | `IN_PROGRESS` + commits + no consensus tracker | Mark the slice as already spawned so the scheduler does not re-yield it | +| 1 | `IN_PROGRESS`, no commits on integration branch | Reap any orphaned Job, then no scheduler action — the scheduler re-yields the slice as `READY` and a fresh team spawns | +| 2 | `IN_PROGRESS` + commits + no consensus tracker | Reap any orphaned Job, then take no scheduler action so the slice re-yields `READY` and the run loop re-drives it ([#3685](https://github.com/jwbron/egg/issues/3685)) | | 3 | `IN_PROGRESS` + commits + consensus reached but unrecorded | Mark `COMPLETE`; emit a louder-than-fresh-spawn audit warning | | 4 | `BLOCKED` | Preserve the BLOCKED state; if no pending HITL decision exists, escalate to HITL via `_escalate_blocked_slice_to_hitl` | | 5 | Corrupt / unclassifiable state | Escalate to HITL via `_escalate_corrupt_slice_to_hitl` | Cases 4 and 5 create an unresolved `Decision` on the contract (via the shared `_escalate_layer_c_hitl` helper) with three options — mark complete, restart slice, cancel — and pause the pipeline rather than silently re-yielding as `READY`. Per the plan task body: "silent classification error is worse than an operator pause." +**Why case 2 re-drives instead of resuming in place ([#3685](https://github.com/jwbron/egg/issues/3685)).** Case 2 originally called `scheduler.mark_spawned`, on the reasoning that the surviving agent pods (or a lazy spawn-on-need path) carried the slice forward. Post-[#3164](https://github.com/jwbron/egg/issues/3164) neither exists. The orchestrator-owned BRC event loop is the only thing that dispatches work for a slice; it is process-local, `ConcurrentPhaseExecutor.spawn_all` is the only thing that starts one, and `_run_one_slice` is the only caller of `scheduler.record_complete` for a running slice. `mark_spawned` at bootstrap therefore parked the slice in scheduler-`RUNNING` with no dispatcher and no completer: `iter_ready` never re-yields a `RUNNING` slice, it holds a `max_parallel_slices` slot, and `all_done()` can never turn true, so the pipeline reports `status: running` with zero pods indefinitely while every liveness signal (driver thread present, per-slice tracker reconstructed, slice classified `resumed`) reads healthy. The [#2914](https://github.com/jwbron/egg/issues/2914) live-pod guard only narrowed the wedge to "a pod was still alive at bootstrap", which post-#3164 means "a one-shot event Job that exits within minutes". Re-driving costs little and is what "resume" already means under orchestrator ownership: `spawn_all` spawns no agents up front, it starts the event loop, and the loop's one-shot Jobs pick up the commits already on the slice's integration branch (which resumes in place via `integration_base_sha`, [#2947](https://github.com/jwbron/egg/issues/2947)). Before re-driving, `_reap_orphaned_slice_jobs` force-removes any Job still live for the slice: its loop is gone, so nothing will observe its termination or derive its next event, and left in place it holds the role's worktree while the fresh cohort's Job attaches to the same checkout ([#3337](https://github.com/jwbron/egg/issues/3337)). Per-agent worktrees are deliberately not deleted; the re-driven slice wants them warm. The audit log's `redriven=` field replaces the old `resumed=` / `reclassified_fresh=` pair: a slice id under `redriven` that never appears in a subsequent spawn is a real anomaly, whereas `resumed` was indistinguishable from a wedge. One known cost: the per-slice tracker reconstructed at startup (#2409) is superseded by the fresh one `spawn_all` registers, so the slice replays a BRC round against commits already on its branch. + +**Where the reap runs, and why it is not conditioned on the classification.** `_reap_orphaned_slice_jobs` is called for **both** case 1 (`"fresh"`) and case 2 (`"resume"`), above the classification switch rather than inside one of its arms. Both leave the slice `READY` and therefore admissible, and the only thing separating them is whether commits reached the slice's integration branch on origin — a signal orthogonal to whether an agent Job is still live. An orchestrator recycle before the slice's first push, or a transient gateway probe failure (which `_classify_non_complete_slice` deliberately defaults to `"fresh"`), both classify `"fresh"` with the orphan still `Running`; conditioning the reap on `"resume"` would leave the #3337 race open on exactly those paths, and nothing downstream closes it (`_reap_superseded_siblings` documents that it cannot match a same-role key adopted across a restart). Classification decides what prompt the re-driven cohort gets, never whether orphan teardown happens. + +**Why the reap waits for observed teardown.** `remove_agent_container(force=True)` only *orders* a foreground delete — `KubernetesClient.remove_container` says in its own docstring that the GC removes pods asynchronously and they are not guaranteed gone when the call returns. Returning there and immediately admitting the slice would re-open the window the reap exists to close: the orphan is still `Running` inside its termination grace period with the role worktree mounted when the fresh cohort's Job attaches and `_clean_reused_worktree` runs `git reset --hard && git clean -fd` under it. So `_await_reaped_jobs_gone` waits on `k8s.wait_for_job_gone` per reaped Job, bounded by a shared `_REAP_TEARDOWN_WAIT_SECONDS` (20 s) deadline — the same "wait for the teardown you requested to be OBSERVED" contract `restart_agent` adopted in [#3597](https://github.com/jwbron/egg/issues/3597). An unobserved teardown (budget exhausted, waiter raised, Job still terminating, or a Pod-UID-only handle that `wait_for_job_gone` cannot address) is logged as such and reported via `teardown_confirmed=false` on the audit line rather than silently claimed; the field under-claims by design, so `false` means "not observed", never "the reap failed". Note this is deliberately **stricter** than `restart_phase` step 4, whose teardown is unwaited: step 4 goes on to delete the per-agent worktrees (step 4b, with salvage), so nothing remains for a lingering pod to corrupt. Here the worktrees are kept warm on purpose, which is exactly what makes the observed teardown load-bearing. + +**Why adoption cannot be relied on instead of reaping.** `compute_dedupe_key` *is* deterministic across orchestrator restarts, and that is what lets live-Job reconciliation recognise an in-flight event after a restart — so a matching key is possible, and where it matches the spawner's `_event_dedupe_key_live` adoption would collapse the duplicate on its own. What breaks the guarantee is narrower than "the new process has no tracker" (it does — `startup_reconciliation` reconstructs one, and the classifier consults it): `spawn_all` registers a *fresh* tracker that supersedes the reconstructed one, so the identity the new loop derives comes from zeroed round state. Where the orphan was mid-round, its key differs, adoption cannot see it, and two pods run against one worktree. The orchestrator cannot tell the two cases apart from the outside — it has no view of the dead loop's session or round state — so it reaps unconditionally and accepts that a matching-key orphan is killed slightly early, costing one replayed round. A redundant delete is recoverable; a clobbered worktree is not. + **Case-4 suppression scope.** The case-4 anomaly check (`_slice_has_pending_decision`) returns `True` whenever the contract carries *any* unresolved `Decision`, regardless of which slice or phase it originated in — the contract's `Decision` schema does not carry a structured `slice_id` tag today, so the classifier conservatively treats any unresolved HITL as "potentially the reason this slice is BLOCKED" and skips the missing-HITL escalation. Practical consequence operators should know: a single unrelated pending HITL on the pipeline (e.g. a refine-phase question) silently suppresses every BLOCKED-slice case-4 escalation until that decision resolves. The trade-off favours not double-alerting an operator who is already engaged with a HITL over surfacing a real cross-slice mismatch; a future schema bump that adds slice scoping to `Decision` can tighten this without changing call sites. The classifier itself is extracted to a module-level helper `_classify_non_complete_slice` so tests can fake the gateway probe and consensus-tracker lookup without spinning up the run loop. The classifier's probe-failure default is "fresh, re-yield `READY`" (the safer direction for the scheduler), while the resolver's probe-failure default is "derived parent" (the safer direction for the next push) — the asymmetry is deliberate and each picks the direction least likely to break its own caller. **Per-slice consensus tracker reconstruction at startup ([#2409](https://github.com/jwbron/egg/issues/2409)).** Documented in the [Startup reconciliation](#pipeline-state-persistence) section above. The signals route (`handle_consensus_confirmed_signal`) also no longer skips reconstruction when `slice_id` is supplied — the strict-equality filter in `reconstruct_tracker_from_messages` is the canonical scope mechanism, so slice-scoped reconstruction is safe at the CONFIRMED handler too. diff --git a/docs/reference/agent-recovery.md b/docs/reference/agent-recovery.md index 4d1ed0e0e..4fdcf39dc 100644 --- a/docs/reference/agent-recovery.md +++ b/docs/reference/agent-recovery.md @@ -329,7 +329,7 @@ Restarts are allowed when the pipeline is in `RUNNING`, `AWAITING_HUMAN`, `FAILE 6. Per-agent worktrees and their local branches are **deleted** — unpushed commits are salvaged to `egg/recovered/*` refs on a best-effort basis first (see [Salvaging Unpushed Local Commits](#salvaging-unpushed-local-commits); worktrees with a corrupted `.git` marker may be skipped without salvage). Fresh worktrees then re-fork from `origin/` tip, so only commits pushed to the shared work branch survive into respawned agents' trees. For per-worktree retention, use `restart_agent` instead (#3080). 7. All agents for the phase are respawned from scratch -**Resume-after-orchestrator-restart vs. operator-driven phase restart.** The clear above runs when an operator (or the overseer/HITL ladder) calls `restart_phase` to start the phase over. The orthogonal case — an orchestrator-pod recycle mid-phase that should **resume** the in-flight slice DAG rather than start it over — is handled by **Layer-C bootstrap reconciliation** on the next orchestrator startup. Layer C iterates non-`COMPLETE` slices, observes integration-branch commit counts and consensus tracker presence, and applies a 5-way classification: re-yield as `READY` (no commits yet), mark already-spawned (commits but no tracker), mark `COMPLETE` (commits + consensus reached but unrecorded), preserve `BLOCKED` (and escalate to HITL if no pending decision), or escalate corrupt state to HITL. Cases 4 and 5 create an unresolved `Decision` on the contract — silent classification error is worse than an operator pause. See [`Slice/phase restart hardening`](../architecture/orchestrator.md#slicephase-restart-hardening-closes-2409). +**Resume-after-orchestrator-restart vs. operator-driven phase restart.** The clear above runs when an operator (or the overseer/HITL ladder) calls `restart_phase` to start the phase over. The orthogonal case — an orchestrator-pod recycle mid-phase that should **resume** the in-flight slice DAG rather than start it over — is handled by **Layer-C bootstrap reconciliation** on the next orchestrator startup. Layer C iterates non-`COMPLETE` slices, observes integration-branch commit counts and consensus tracker presence, and applies a 5-way classification: re-yield as `READY` (no commits yet), re-yield as `READY` (commits but no tracker; [#3685](https://github.com/jwbron/egg/issues/3685): re-driving the slice is what starts its BRC event loop, and the loop is the only thing that dispatches its work), mark `COMPLETE` (commits + consensus reached but unrecorded), preserve `BLOCKED` (and escalate to HITL if no pending decision), or escalate corrupt state to HITL. Both re-yield cases first reap any Job still live for the slice and wait (bounded) for its teardown to be observed — the reap runs above the classification switch, because commits-on-origin is orthogonal to whether an orphaned Job still holds the role worktree. Cases 4 and 5 create an unresolved `Decision` on the contract — silent classification error is worse than an operator pause. See [`Slice/phase restart hardening`](../architecture/orchestrator.md#slicephase-restart-hardening-closes-2409). Startup reconciliation also rebuilds per-slice consensus trackers ([#2409](https://github.com/jwbron/egg/issues/2409) closure): for each slice in `contract.slices`, `reconstruct_tracker_from_messages(pipeline_id, graph, slice_id=)` rebuilds the nested `{pipeline_id}/{slice_id}` tracker from the message store, so a recycled orchestrator pod no longer loses in-flight slice consensus. diff --git a/orchestrator/routes/pipelines/__init__.py b/orchestrator/routes/pipelines/__init__.py index def669c2a..3e65d838e 100644 --- a/orchestrator/routes/pipelines/__init__.py +++ b/orchestrator/routes/pipelines/__init__.py @@ -1280,11 +1280,13 @@ def stream_pipeline(pipeline_id: str) -> Response: _teardown_phase_overseer, ) from ._pod_liveness import ( # noqa: E402,F401 + _REAP_TEARDOWN_WAIT_SECONDS, + _await_reaped_jobs_gone, _count_live_pods_for_pipeline, _get_spawner, _guard_live_pods_or_force, _live_event_agents, - _slice_agents_alive, + _reap_orphaned_slice_jobs, ) from ._populate import ( # noqa: E402,F401 _FOREST_REASON_TO_OUTCOME, diff --git a/orchestrator/routes/pipelines/_pod_liveness.py b/orchestrator/routes/pipelines/_pod_liveness.py index 967326517..5514b35c4 100644 --- a/orchestrator/routes/pipelines/_pod_liveness.py +++ b/orchestrator/routes/pipelines/_pod_liveness.py @@ -9,6 +9,16 @@ import routes.pipelines as _pkg # noqa: E402,F401 +# Budget for observing reaped slice Jobs actually gone (#3685). Job +# deletion is asynchronous even with ``force=True``, so the reap must +# wait for the teardown it requested before the run loop admits the +# slice — otherwise the fresh cohort's Job attaches to a worktree the +# orphan still holds. Mirrors ``_routes_restart._JOB_TEARDOWN_WAIT_SECONDS`` +# (20.0) and ``kubernetes_spawner._EVENT_JOB_TERMINATION_WAIT_S`` (15.0); +# the deadline is shared across every Job reaped for one slice so a +# pathological cohort cannot stall the bootstrap pass unboundedly. +_REAP_TEARDOWN_WAIT_SECONDS = 20.0 + def _get_spawner(): """Get the appropriate spawner for the current runtime. @@ -122,43 +132,237 @@ def _live_event_agents(pipeline_id: str, slice_id: str | None) -> list[dict[str, return entries -def _slice_agents_alive(spawner: _pkg.Any, pipeline_id: str, slice_id: str) -> bool: - """Check if any live agents exist for a slice (#2914). - - Returns ``True`` if at least one pod labeled with the pipeline and - slice IDs is in a live state (Pending/Creating/Running). Returns - ``False`` if zero live pods or if the label query fails — the - conservative default forces re-spawn rather than risking a wedge. - - Caller contract: callers must have already torn down stale cohorts - with foreground propagation (e.g. ``restart_phase`` step 4 calls - ``remove_agent_container(force=True)``). A pod whose Job is being - deleted but is still in its termination grace period still reports - ``phase=Running`` (``kubernetes_client.py`` maps Running → RUNNING - without a Terminating-specific status), so without foreground - teardown the helper can false-positive against terminating pods - and wedge again. The ``spawner`` is taken as a parameter (rather - than fetched via ``_get_spawner``) so tests can inject a stub - directly, paralleling how ``_classify_non_complete_slice`` +def _await_reaped_jobs_gone( + spawner: _pkg.Any, + pipeline_id: str, + slice_id: str, + job_names: list[str], +) -> bool: + """Block until the reaped Jobs are observed gone, bounded (#3685). + + ``remove_agent_container(force=True)`` only orders a foreground + delete: ``KubernetesClient.remove_container`` says so in its own + docstring — "the GC deletes the pods asynchronously — they are not + guaranteed gone by the time the call returns". Returning at that + point and immediately making the slice admissible re-opens the very + window the reap exists to close: the orphan is still ``Running`` + inside its termination grace period with the role worktree mounted + when the fresh cohort's Job attaches to the same checkout and + ``_clean_reused_worktree`` runs ``git reset --hard && git clean -fd`` + under it (#3337). + + The spawn-side pre-spawn wait (``kubernetes_spawner._events``) does + not cover this: it selects on ``LABEL_EVENT_DEDUPE={dedupe_key}``, + and the orphan's key is not necessarily the fresh loop's first key + (see ``_reap_orphaned_slice_jobs``). So the wait has to happen here, + on the same "wait for the teardown you requested to be OBSERVED" + contract ``restart_agent`` uses (#3597). + + Returns ``True`` only when every named Job was observed gone. + ``False`` means "not observed", never "the teardown failed" — it + under-claims by design, and each unobserved Job is logged so an + operator can see that the slice was admitted with the window + potentially still open. + """ + k8s = getattr(spawner, "k8s", None) + waiter = getattr(k8s, "wait_for_job_gone", None) + namespace = getattr(k8s, "namespace", None) + if waiter is None or not namespace: + # A backend without the wait helper (or without a namespace) is + # a single backend-capability fact, not N per-Job failures, so + # it earns one log line. Defensive: ``KubernetesClient`` + # implements ``wait_for_job_gone``, and ``ContainerSpawner`` is + # an alias of ``KubernetesSpawner``, so this is unreachable + # against the production spawner. + _pkg.logger.warning( + "Orphaned-Job reap could not observe teardown; slice admitted " + "with the worktree-handoff window potentially open (#3685)", + pipeline_id=pipeline_id, + slice_id=slice_id, + jobs=len(job_names), + reason="no_wait_helper", + ) + return False + + confirmed = True + deadline = _pkg.time.monotonic() + _REAP_TEARDOWN_WAIT_SECONDS + for name in job_names: + remaining = deadline - _pkg.time.monotonic() + if remaining <= 0: + confirmed = False + _pkg.logger.warning( + "Orphaned-Job reap could not observe teardown; slice admitted " + "with the worktree-handoff window potentially open (#3685)", + pipeline_id=pipeline_id, + slice_id=slice_id, + job_name=name, + reason="budget_exhausted", + ) + continue + try: + gone = bool(waiter(name, namespace, timeout_s=remaining)) + except Exception as wait_err: # noqa: BLE001 — the wait is best-effort + confirmed = False + _pkg.logger.warning( + "Orphaned-Job reap teardown wait raised; treating the teardown " + "as unconfirmed (#3685)", + pipeline_id=pipeline_id, + slice_id=slice_id, + job_name=name, + error=str(wait_err), + ) + continue + if not gone: + confirmed = False + _pkg.logger.warning( + "Orphaned slice Job still terminating after the teardown wait; " + "the re-driven cohort may attach to a worktree the orphan " + "still holds (#3685)", + pipeline_id=pipeline_id, + slice_id=slice_id, + job_name=name, + ) + return confirmed + + +def _reap_orphaned_slice_jobs(spawner: _pkg.Any, pipeline_id: str, slice_id: str) -> list[str]: + """Tear down agent Jobs that outlived their event loop (#3685). + + Layer-C bootstrap re-drives every non-COMPLETE slice through the run + loop, and the fresh cohort's one-shot event Jobs re-attach to the + per-role worktrees. Any Job still live at that point belongs to a + BRC event loop that no longer exists (the loop is process-local and + died with the previous orchestrator process, or with the previous + driver thread), so nothing will ever observe its termination or + re-derive its next event. Left in place it holds the role's worktree + while the new cohort's Job attaches to the same checkout — the #3337 + two-live-pods race. + + Why the spawner's live-key adoption is not enough to collapse that + duplicate: ``compute_dedupe_key`` IS deterministic across + orchestrator restarts, so a matching key is possible — but only when + the fresh loop happens to derive the same ``event_identity``, and it + derives that from the tracker ``spawn_all`` registers, which is a + fresh zeroed one superseding the reconstructed one (#2409). Where + the orphan was mid-round its key differs, adoption cannot see it, + and two pods run against one worktree. This process cannot tell the + two cases apart — it has no view of the dead loop's session or + round state — so it reaps unconditionally and accepts that a + matching-key orphan is killed slightly early, costing one replayed + round. That is the strictly safer direction: a redundant delete is + recoverable, a clobbered worktree is not. + + Reaping is foreground (``force=True``) and then WAITED ON: the + delete is asynchronous, so ``_await_reaped_jobs_gone`` observes the + teardown (bounded by :data:`_REAP_TEARDOWN_WAIT_SECONDS`) before the + caller makes the slice admissible. This is deliberately stricter + than ``restart_phase`` step 4, whose own teardown is unwaited — step + 4 goes on to DELETE the per-agent worktrees (step 4b, with salvage), + so nothing is left for a lingering pod to corrupt. Here the + worktrees are deliberately kept warm for the re-driven slice, which + is exactly what makes the observed teardown load-bearing. + + Returns the handles reaped — the Job name where the listing supplied + one, else the container id — empty when the slice has no live Jobs + (the common case after a full pod recycle). Entirely best-effort: a + failed label query or a failed removal is logged and swallowed, + because a reap failure must never block recovery of a slice that is + otherwise ready to re-drive. The ``spawner`` is taken as a parameter + (rather than fetched via ``_get_spawner``) so tests can inject a + stub directly, paralleling how ``_classify_non_complete_slice`` receives ``gateway``. """ try: - pods = spawner.backend.list_containers( - labels={ - _pkg.LABEL_PIPELINE_ID: pipeline_id, - _pkg.LABEL_SLICE_ID: slice_id, - }, - ) - live_count = sum(1 for p in pods if p.status in _pkg._LIVE_POD_STATUSES) - return live_count > 0 + # ``list_slice_jobs`` applies the same + # ``{LABEL_PIPELINE_ID, LABEL_SLICE_ID}`` pair, so the label + # scoping lives in one place rather than being re-derived here. + pods = spawner.list_slice_jobs(pipeline_id, slice_id) except Exception as e: # noqa: BLE001 _pkg.logger.warning( - "Slice liveness check failed; treating as not-alive to force re-spawn (#2914)", + "Orphaned-Job reap skipped: slice Job query failed (#3685)", pipeline_id=pipeline_id, slice_id=slice_id, error=str(e), ) - return False + return [] + + reaped: list[str] = [] + pending_waits: list[str] = [] + unaddressable = 0 + for pod in pods: + # Status-only liveness, unlike ``_job_is_live``, which also + # excludes terminating Jobs via ``deletion_timestamp`` (#3597). + # The asymmetry is deliberate, not an oversight: + # ``KubernetesClient.list_containers`` never populates + # ``deletion_timestamp``, and a Terminating pod still reports + # phase ``Running``, so this filter cannot see the distinction. + # It does not need to — the consequence is a redundant delete on + # a Job already on its way out (plus an entry in the ``reaped=`` + # audit list), whereas on the adoption path treating a corpse as + # live silently swallows a respawn. Waiting on it is correct + # either way: an already-terminating Job is precisely one whose + # worktree we must see released. + if pod.status not in _pkg._LIVE_POD_STATUSES: + continue + job_name = getattr(pod, "job_name", None) + handle = job_name or getattr(pod, "container_id", None) + if not handle: + continue + try: + # ``job_name`` first, matching ``cleanup_pipeline`` and + # ``restart_agent``. ``remove_agent_container`` resolves a + # Job name, Job UID or Pod UID equally well for the delete, + # but it also forwards the handle to + # ``delete_session_by_container``: event-mode gateway + # sessions are keyed by the stable base ``container_id`` + # (not the per-event Job name), so that call only ever + # reaches a legacy per-Job session here — the event-mode + # session is released by the phase/pipeline-end + # ``cleanup_pipeline`` → ``_teardown_session`` path. + spawner.remove_agent_container(handle, force=True, cleanup_session=True) + except Exception as e: # noqa: BLE001 + _pkg.logger.warning( + "Failed to reap orphaned slice Job during bootstrap re-drive (#3685)", + pipeline_id=pipeline_id, + slice_id=slice_id, + container_id=handle, + error=str(e), + ) + continue + reaped.append(handle) + if isinstance(job_name, str) and job_name: + pending_waits.append(job_name) + else: + # Without a Job name the only handle is a Pod UID, which + # ``wait_for_job_gone`` would normalize into a Job name that + # never existed — a 404 on the first read, reported as + # "gone" without observing anything. Count it as unobserved + # rather than claiming an observation we never made. + unaddressable += 1 + + if not reaped: + return [] + + teardown_confirmed = _await_reaped_jobs_gone(spawner, pipeline_id, slice_id, pending_waits) + if unaddressable: + teardown_confirmed = False + _pkg.logger.warning( + "Orphaned-Job reap could not observe teardown; slice admitted " + "with the worktree-handoff window potentially open (#3685)", + pipeline_id=pipeline_id, + slice_id=slice_id, + jobs=unaddressable, + reason="unaddressable", + ) + _pkg.logger.info( + "Reaped orphaned slice Jobs whose BRC event loop died with the " + "previous orchestrator process (#3685)", + pipeline_id=pipeline_id, + slice_id=slice_id, + reaped=reaped, + teardown_confirmed=teardown_confirmed, + ) + return reaped def _guard_live_pods_or_force( diff --git a/orchestrator/routes/pipelines/_run_implement.py b/orchestrator/routes/pipelines/_run_implement.py index 3c28b9d7e..6182969e0 100644 --- a/orchestrator/routes/pipelines/_run_implement.py +++ b/orchestrator/routes/pipelines/_run_implement.py @@ -350,12 +350,31 @@ def _bootstrap_check_one(slice_obj: _pkg.Any) -> tuple[str, bool]: # action; the scheduler will re-yield the slice as READY and # the run loop spawns fresh agents. # (2) IN_PROGRESS, commits on integration branch, consensus - # NOT reached → call ``scheduler.mark_spawned`` so the run - # loop does NOT respawn. Per-slice tracker reconstruction - # is handled at orchestrator boot by - # startup_reconciliation.py (slice-4 TASK-4-5); the - # producer pods (if alive) or the lazy spawn-on-need path - # carry the slice forward. + # NOT reached → reap any orphaned Jobs and take no scheduler + # action, so the slice re-yields READY and the run loop + # re-drives it (#3685). This branch used to call + # ``scheduler.mark_spawned`` on the theory that "the producer + # pods (if alive) or the lazy spawn-on-need path carry the + # slice forward". Neither exists: post-#3164 the ONLY thing + # that dispatches work for a slice is the orchestrator-owned + # BRC event loop, which ``spawn_all`` starts and which died + # with the previous process; and the ONLY caller of + # ``scheduler.record_complete`` for a running slice is + # ``_run_one_slice``. So ``mark_spawned`` here parked the + # slice in scheduler-RUNNING with no dispatcher and no + # completer: ``iter_ready`` never re-yields it, it holds a + # ``max_parallel_slices`` slot, ``all_done()`` never turns + # true, and the pipeline reports ``running`` with zero pods + # forever. Re-driving is cheap and is what "resume" already + # means post-#3164: ``spawn_all`` spawns no agents up front, + # it starts the event loop, and the loop's one-shot Jobs pick + # the slice up where its branch left off (the integration + # branch resumes in place via ``integration_base_sha``, + # #2947). The per-slice tracker reconstructed at boot + # (slice-4 TASK-4-5) is superseded by the fresh one + # ``spawn_all`` registers, so the slice replays one BRC round + # against commits already on its branch rather than making no + # progress at all. # (3) IN_PROGRESS, commits on integration branch, consensus # REACHED, slice PR NOT opened → mark COMPLETE so the # slice-PR opener path (with TASK-3-2 idempotency @@ -367,11 +386,10 @@ def _bootstrap_check_one(slice_obj: _pkg.Any) -> tuple[str, bool]: # (5) Unknown / corrupt state (impossible status enum value) # → surface an OVERSEER_ALERT instead of silently # re-yielding as READY. - bootstrap_resumed: list[str] = [] bootstrap_consensus_complete: list[str] = [] bootstrap_blocked: list[str] = [] bootstrap_corrupt: list[str] = [] - bootstrap_reclassified_fresh: list[str] = [] # resume-but-dead → fresh (#2914) + bootstrap_redriven: list[str] = [] # resume-classified → re-driven (#3685) layer_b_marked_complete = set(bootstrap_merged) for s in layer_b_candidates: if s.id in layer_b_marked_complete: @@ -386,6 +404,26 @@ def _bootstrap_check_one(slice_obj: _pkg.Any) -> tuple[str, bool]: gateway_mode=gateway_mode, consensus_tracker_lookup=_pkg._lookup_peer_consensus_tracker_or_none, ) + if classification in ("resume", "fresh"): + # Reap BEFORE the branch switch, not inside one of its arms + # (#3685). Cases 1 ("fresh") and 2 ("resume") both leave the + # slice READY, so the run loop admits it and ``spawn_all`` + # attaches a fresh cohort to the per-role worktrees — and + # the only thing separating the two classifications is + # whether commits reached the integration branch on origin, + # which is orthogonal to whether an agent Job is still live. + # An orchestrator recycle before the slice's first push, or + # a transient gateway probe failure (which + # ``_classify_non_complete_slice`` deliberately defaults to + # "fresh"), both yield "fresh" with the orphan still + # Running. Wiring the reap to "resume" alone would leave the + # #3337 two-live-pods race wide open on exactly those paths, + # and nothing downstream covers it — + # ``_reap_superseded_siblings`` documents that it cannot + # match keys adopted across a restart. Classification + # decides what prompt the re-driven cohort gets; it must + # never decide whether orphan teardown happens. + _pkg._reap_orphaned_slice_jobs(spawner, pipeline_id, s.id) if classification == "consensus_complete": # Case 3 — louder than fresh-spawn but quieter than # case-4/5 HITL. A warning here makes the non-trivial @@ -404,21 +442,12 @@ def _bootstrap_check_one(slice_obj: _pkg.Any) -> tuple[str, bool]: bootstrap_consensus_complete.append(s.id) continue if classification == "resume": - # Verify agents are actually live before marking as spawned (#2914). - # On restart_phase, agents were torn down but contract still shows - # IN_PROGRESS with commits — we must not mark_spawned when cohort - # is absent, or the pipeline wedges with no agents running. - if _pkg._slice_agents_alive(spawner, pipeline_id, s.id): - scheduler.mark_spawned(s.id) - bootstrap_resumed.append(s.id) - else: - _pkg.logger.warning( - "Layer-C resume classification but no live agents; " - "treating as fresh to force re-spawn (#2914)", - pipeline_id=pipeline_id, - slice_id=s.id, - ) - bootstrap_reclassified_fresh.append(s.id) + # Case 2: re-drive (#3685, supersedes the #2914 liveness + # guard). No scheduler call: leaving the slice READY is + # what makes the run loop admit it, which is the only path + # that starts its BRC event loop and can ever complete it. + # The orphan reap already ran above. + bootstrap_redriven.append(s.id) continue if classification == "blocked": bootstrap_blocked.append(s.id) @@ -426,7 +455,8 @@ def _bootstrap_check_one(slice_obj: _pkg.Any) -> tuple[str, bool]: if classification == "corrupt": bootstrap_corrupt.append(s.id) continue - # "fresh" → no Layer-C action, scheduler re-yields READY. + # "fresh" → no scheduler action beyond the reap above; the + # scheduler re-yields the slice as READY. # The bootstrap passes above persist with ``commit_to_branch=False`` # — one batched commit+push here covers every reconciled slice @@ -444,13 +474,7 @@ def _bootstrap_check_one(slice_obj: _pkg.Any) -> tuple[str, bool]: already_complete_on_contract=bootstrap_complete, detected_merged_on_origin=bootstrap_merged, ) - if ( - bootstrap_resumed - or bootstrap_consensus_complete - or bootstrap_blocked - or bootstrap_corrupt - or bootstrap_reclassified_fresh - ): + if bootstrap_consensus_complete or bootstrap_blocked or bootstrap_corrupt or bootstrap_redriven: # NOTE: include ``bootstrap_blocked`` in the gate (reviewer_code # v3 NACK fix) — a bootstrap pass whose only Layer-C activity is # BLOCKED slices was previously suppressing the audit-trail line @@ -458,19 +482,22 @@ def _bootstrap_check_one(slice_obj: _pkg.Any) -> tuple[str, bool]: # the structured "we saw a blocked slice" log to spot # pending-HITL backlogs without grepping for the side-effect. # - # Also include ``bootstrap_reclassified_fresh`` (#2914) — resume- - # classified slices that were re-verified against k8s and found - # to have no live agents. Surfacing the reclassification here - # gives operators a structured audit trail for the - # ``restart_phase``-recovery path. + # ``redriven`` replaces the old ``resumed`` / ``reclassified_fresh`` + # pair (#3685). The distinction those two drew (mark-spawned vs + # re-yielded) no longer exists: mark-spawning a slice at bootstrap + # could never make progress, so every case-2 slice now takes the + # re-drive path. Operators should read ``redriven`` as "these + # slices will be admitted again on a following tick"; a slice id + # here that never shows up in a subsequent spawn is a real + # anomaly, whereas the old ``resumed`` was indistinguishable + # from a wedge. _pkg.logger.info( "Slice bootstrap reconciliation classified non-COMPLETE slices (slice-4 TASK-4-4)", pipeline_id=pipeline_id, - resumed=bootstrap_resumed, + redriven=bootstrap_redriven, consensus_complete_unrecorded=bootstrap_consensus_complete, blocked=bootstrap_blocked, corrupt=bootstrap_corrupt, - reclassified_fresh=bootstrap_reclassified_fresh, ) # Case 5 — escalate via HITL so the pipeline pauses until the # operator picks an option (reviewer_contract / reviewer_code v1 diff --git a/orchestrator/routes/pipelines/_slice_state.py b/orchestrator/routes/pipelines/_slice_state.py index 464d809f7..445ff4211 100644 --- a/orchestrator/routes/pipelines/_slice_state.py +++ b/orchestrator/routes/pipelines/_slice_state.py @@ -609,9 +609,12 @@ def _classify_non_complete_slice( origin. No Layer-C action; the scheduler re-yields READY and the run loop spawns fresh agents. * ``"resume"`` — case (2) IN_PROGRESS with commits on origin and - consensus NOT reached. Caller calls - ``scheduler.mark_spawned(slice_id)`` so the run loop does NOT - respawn. + consensus NOT reached. Caller takes no scheduler action so the + slice re-yields READY and the run loop re-drives it, which is + what starts its BRC event loop (#3685); it reaps any orphaned + Job first. The label is kept as a description of the observed + state (the slice resumes on the commits already on its + integration branch), not as an instruction to mark it spawned. * ``"consensus_complete"`` — case (3) IN_PROGRESS with commits and ``tracker.evaluate()['is_complete']`` True. Caller marks the slice COMPLETE so the next loop iteration runs the slice-PR @@ -669,8 +672,14 @@ def _classify_non_complete_slice( except Exception as probe_err: # noqa: BLE001 # Probe failure (gateway down, transient HTTP). Conservative # default: treat as has_commits=False so the slice is - # re-yielded READY rather than silently mark-spawned with - # no agents alive. + # re-yielded READY. Both the "fresh" and "resume" outcomes + # reap any orphaned Job and then re-yield READY today + # (#3685) — the Layer-C loop runs the reap above the + # classification switch precisely so this default stays + # cheap — so the cost of guessing wrong here is only the + # prompt content of the re-driven cohort: not whether the + # slice runs at all, and not whether a Job that outlived its + # event loop is torn down first. # # NOTE on asymmetry vs. ``_resolve_slice_base_branch`` # (slice-4 TASK-4-3, ~line 10510): the resolver defaults diff --git a/orchestrator/tests/test_ble001_narrowing_audit.py b/orchestrator/tests/test_ble001_narrowing_audit.py index ae260fb49..c5f72a9c6 100644 --- a/orchestrator/tests/test_ble001_narrowing_audit.py +++ b/orchestrator/tests/test_ble001_narrowing_audit.py @@ -196,7 +196,16 @@ def test_audit_window_retains_documented_ble001_population() -> None: # moment the operator's answer was recorded. Broadened to match every other # best-effort block in that file — an observability write must never strand # the gate. - assert len(noqa_lines) <= 124, ( + # Raised 124 -> 126 by #3685, which added two audited sites to the + # orphaned-Job reap in ``_pod_liveness.py``: + # + the ``spawner.list_slice_jobs`` label query — a failed listing must + # degrade to "no orphans found" and log, never block recovery of a + # slice that is otherwise ready to re-drive; + # + the bounded teardown wait's ``waiter(...)`` call — a raising waiter + # degrades to ``confirmed = False`` (teardown unconfirmed) so the reap + # reports an open worktree-handoff window instead of propagating out of + # the admission path. + assert len(noqa_lines) <= 126, ( f"Found {len(noqa_lines)} ``# noqa: BLE001`` swallows in " f"the routes/pipelines package, well past the documented " f"population — a future PR appears to have re-introduced swallow-all " diff --git a/orchestrator/tests/test_slice_phase_restart_hardening.py b/orchestrator/tests/test_slice_phase_restart_hardening.py index 73369f3d8..d33bcd0cf 100644 --- a/orchestrator/tests/test_slice_phase_restart_hardening.py +++ b/orchestrator/tests/test_slice_phase_restart_hardening.py @@ -37,8 +37,11 @@ - ``"fresh"`` — IN_PROGRESS/PENDING + no commits on origin → no Layer-C action; scheduler re-yields READY. - - ``"resume"`` — IN_PROGRESS + commits + no consensus → - ``scheduler.mark_spawned``, no respawn. + - ``"resume"``: IN_PROGRESS + commits + no consensus → no scheduler + action, so the slice re-yields READY and the run loop re-drives it + (#3685: re-driving is what starts the slice's BRC event loop, and + ``_run_one_slice`` is the only caller that can complete it; the old + ``scheduler.mark_spawned`` parked the slice in RUNNING with neither). - ``"consensus_complete"`` — IN_PROGRESS + commits + consensus REACHED → mark COMPLETE so the slice-PR opener fires. - ``"blocked"`` — BLOCKED → preserve status; caller escalates via @@ -61,7 +64,7 @@ slice-id filter without one slice's ack/propose records bleeding into the other's tracker. -The classes below also cover ``_slice_agents_alive`` (#2914 / #2916), the +The classes below also cover ``_reap_orphaned_slice_jobs`` (#3685), the ``_escalate_layer_c_hitl`` HITL persistence helpers, and ``GatewayClient.merge_base`` strict SHA-shape + session-auth bootstrapping. """ @@ -919,8 +922,9 @@ def test_d1_in_progress_no_commits_classifies_fresh(self) -> None: def test_d2_in_progress_commits_no_consensus_classifies_resume(self) -> None: """(d2) IN_PROGRESS + commits pushed + consensus NOT - reached → ``"resume"``: caller marks the scheduler - ``mark_spawned`` so the run loop does NOT respawn.""" + reached → ``"resume"``: caller takes no scheduler action, so + the run loop re-drives the slice onto the commits already on + its integration branch (#3685).""" from routes.pipelines import _classify_non_complete_slice slice_obj = _make_slice("slice-1", status=SliceStatus.IN_PROGRESS) @@ -1061,8 +1065,8 @@ def test_repoless_pipeline_classifies_fresh(self) -> None: def test_probe_failure_classifies_fresh(self) -> None: """Adversarial probe: the gateway probe raises mid-call. The classifier MUST default to ``"fresh"`` rather than - ``"resume"`` so we don't silently mark-spawn a slice whose - true origin state is unknown. + ``"resume"`` so a slice whose true origin state is unknown + takes the plainest recovery path. """ from routes.pipelines import _classify_non_complete_slice @@ -1082,144 +1086,288 @@ def test_probe_failure_classifies_fresh(self) -> None: # --------------------------------------------------------------------------- -# #2914: _slice_agents_alive() — k8s probe for restart-phase resume guard +# #3685: _reap_orphaned_slice_jobs(), the foreground teardown of Jobs whose +# BRC event loop died with the previous orchestrator process # -# The fix for #2914 adds a runtime check that prevents the bootstrap -# reconciler from calling scheduler.mark_spawned() when no live agents -# exist. Without this, restart_phase on a sliced implement wedges the -# pipeline: the scheduler thinks the slice is RUNNING but no containers -# are present, so no signals can arrive and the slice never completes. +# Layer C re-drives every non-COMPLETE slice through the run loop (that +# is the only path that starts a slice's event loop, and the only one +# that can ever complete it). A Job that survived the restart holds the +# role's worktree and would race the fresh cohort's Job on the same +# checkout (#3337). ``compute_dedupe_key`` IS deterministic across +# restarts, so the spawner's live-key adoption *may* collapse the +# duplicate — but ``spawn_all`` registers a fresh, zeroed tracker that +# supersedes the reconstructed one (#2409), so a mid-round orphan's key +# differs and adoption cannot see it. This process cannot tell the two +# apart, so the re-drive reaps unconditionally: a redundant delete is +# recoverable, a clobbered worktree is not. # -# This helper must be defensive: -# - Returns False (force fresh re-spawn) on any k8s API error -# - Returns False when zero pods match the slice labels -# - Returns True only when at least one pod is in a live state -# - Filters by both pipeline_id AND slice_id labels (not just pipeline) +# The helper must be defensive: +# - Removes only live pods (Pending / Creating / Running) +# - Leaves terminal pods alone (they are already gone) +# - Scopes the query to both pipeline_id AND slice_id (delegated to +# ``spawner.list_slice_jobs``, which owns that label pair) +# - Swallows a failed query and a failed removal: a reap failure must +# never block recovery of a slice that is ready to re-drive +# - Waits (bounded) for each reaped Job to actually be gone before +# returning, since ``force=True`` only orders an async delete # --------------------------------------------------------------------------- -class TestSliceAgentsAlive: - """Exercise _slice_agents_alive() against a stubbed spawner backend. +class TestReapOrphanedSliceJobs: + """Exercise ``_reap_orphaned_slice_jobs`` against a stubbed spawner. The helper takes ``spawner`` as a parameter (paralleling how ``_classify_non_complete_slice`` takes ``gateway``) so tests inject a stub directly without patching ``routes.pipelines._get_spawner``. """ - @staticmethod - def _make_container_info(container_id: str, status): + # Sentinel so ``job_name=None`` means "the listing supplied no Job + # name" rather than "use the default". + _AUTO_JOB_NAME = object() + + @classmethod + def _make_container_info(cls, container_id: str, status, job_name=_AUTO_JOB_NAME): from models import ContainerInfo return ContainerInfo( container_id=container_id, container_name=f"egg-{container_id}", status=status, + job_name=(f"job-{container_id}" if job_name is cls._AUTO_JOB_NAME else job_name), ) - def _make_spawner(self, returned_pods): - """Build a spawner stub whose backend.list_containers yields - the given pods.""" - backend = MagicMock() - backend.list_containers.return_value = returned_pods + def _make_spawner(self, returned_pods, *, gone: bool = True): + """Build a spawner stub whose ``list_slice_jobs`` yields the + given pods and whose Job-teardown waiter reports ``gone``.""" spawner = MagicMock() - spawner.backend = backend + spawner.list_slice_jobs.return_value = returned_pods + spawner.k8s.namespace = "egg" + spawner.k8s.wait_for_job_gone.return_value = gone return spawner - def test_true_when_running_pod_exists(self): - """At least one RUNNING pod → slice is live, resume is safe.""" + def test_reaps_running_pod(self): + """A RUNNING pod is an orphan of the dead loop: force-remove it.""" from models import ContainerStatus - from routes.pipelines import _slice_agents_alive + from routes.pipelines import _reap_orphaned_slice_jobs - pods = [ - self._make_container_info("p1", ContainerStatus.RUNNING), - ] + spawner = self._make_spawner([self._make_container_info("p1", ContainerStatus.RUNNING)]) - spawner = self._make_spawner(pods) - assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is True + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == ["job-p1"] + spawner.remove_agent_container.assert_called_once_with( + "job-p1", force=True, cleanup_session=True + ) - def test_true_when_pending_pod_exists(self): - """PENDING pod (still scheduling) → slice is live, don't re-spawn.""" + def test_reaps_pending_and_creating_pods(self): + """PENDING / CREATING pods are live too (``LIVE_POD_STATUSES``): + a Job mid-spawn when the orchestrator went down is just as + orphaned as a Running one, and will attach to the same worktree + the moment its pod starts.""" from models import ContainerStatus - from routes.pipelines import _slice_agents_alive + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = self._make_spawner( + [ + self._make_container_info("p1", ContainerStatus.PENDING), + self._make_container_info("p2", ContainerStatus.CREATING), + ] + ) - pods = [ - self._make_container_info("p1", ContainerStatus.PENDING), + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == [ + "job-p1", + "job-p2", ] + assert spawner.remove_agent_container.call_count == 2 - spawner = self._make_spawner(pods) - assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is True + def test_skips_terminal_pods(self): + """EXITED / FAILED pods have already gone; removing them would + be a wasted API round-trip per bootstrap.""" + from models import ContainerStatus + from routes.pipelines import _reap_orphaned_slice_jobs - def test_true_when_creating_pod_exists(self): - """CREATING pod (Job→Pod transition) → slice is live, don't re-spawn. + spawner = self._make_spawner( + [ + self._make_container_info("p1", ContainerStatus.EXITED), + self._make_container_info("p2", ContainerStatus.FAILED), + ] + ) - ``_LIVE_POD_STATUSES`` (``models.LIVE_POD_STATUSES``) includes - CREATING because k8s Jobs pass through it on their way to - Running. Without this branch, a slice mid-spawn would be - misclassified as dead and double-spawned. (reviewer suggestion 2 - on #2916: same shape as the RUNNING/PENDING tests.) - """ + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == [] + spawner.remove_agent_container.assert_not_called() + + def test_no_pods_is_a_noop(self): + """The common case after a full pod recycle: nothing to reap.""" + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = self._make_spawner([]) + + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == [] + spawner.remove_agent_container.assert_not_called() + spawner.k8s.wait_for_job_gone.assert_not_called() + + def test_swallows_list_error(self): + """Defensive: a k8s API error must not abort the re-drive.""" + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = MagicMock() + spawner.list_slice_jobs.side_effect = RuntimeError("k8s unreachable") + + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == [] + + def test_swallows_removal_error_and_continues(self): + """One failed removal must not skip the remaining orphans, and + must not propagate: the slice still needs to be re-driven.""" from models import ContainerStatus - from routes.pipelines import _slice_agents_alive + from routes.pipelines import _reap_orphaned_slice_jobs - pods = [ - self._make_container_info("p1", ContainerStatus.CREATING), - ] + spawner = self._make_spawner( + [ + self._make_container_info("p1", ContainerStatus.RUNNING), + self._make_container_info("p2", ContainerStatus.RUNNING), + ] + ) + spawner.remove_agent_container.side_effect = [RuntimeError("boom"), None] + + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == ["job-p2"] + assert spawner.remove_agent_container.call_count == 2 + # Only the Job that was actually removed is waited on. + assert [c.args[0] for c in spawner.k8s.wait_for_job_gone.call_args_list] == ["job-p2"] + + def test_scopes_the_query_to_pipeline_and_slice(self): + """The query must carry both the pipeline and the slice so a + sibling slice's live cohort is never reaped by this slice's + bootstrap. ``list_slice_jobs`` owns the label pair, so the + assertion is on the arguments handed to it.""" + from models import ContainerStatus + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = self._make_spawner( + [self._make_container_info("p1", ContainerStatus.RUNNING)], + ) - spawner = self._make_spawner(pods) - assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is True + _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-2") - def test_false_when_no_pods(self): - """Zero pods → slice is dead, force fresh re-spawn.""" - from routes.pipelines import _slice_agents_alive + spawner.list_slice_jobs.assert_called_once_with("pipeline-x", "slice-2") - spawner = self._make_spawner([]) - assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is False + # -- teardown wait (#3685 review round 2) -------------------------------- + # ``remove_agent_container(force=True)`` only *orders* the delete; + # ``KubernetesClient.remove_container`` says so in its own docstring. + # Returning before the pod is gone re-opens the exact window the reap + # exists to close, so the reap waits for the teardown it requested to + # be observed — the #3597 contract. An unobserved teardown is reported + # honestly rather than claimed, and never blocks the re-drive. - def test_false_when_only_terminal_pods(self): - """Only EXITED/FAILED pods (post-restart_phase cleanup) → slice is dead.""" + def test_waits_for_each_reaped_job_to_be_gone(self): + """Every removed Job is waited on by name, in the cluster + namespace, with a positive remaining budget.""" from models import ContainerStatus - from routes.pipelines import _slice_agents_alive + from routes.pipelines import _reap_orphaned_slice_jobs - pods = [ - self._make_container_info("p1", ContainerStatus.EXITED), - self._make_container_info("p2", ContainerStatus.FAILED), - ] + spawner = self._make_spawner( + [ + self._make_container_info("p1", ContainerStatus.RUNNING), + self._make_container_info("p2", ContainerStatus.RUNNING), + ] + ) - spawner = self._make_spawner(pods) - assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is False + _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") - def test_false_on_k8s_api_error(self): - """Defensive: k8s API error → assume dead, force re-spawn.""" - from routes.pipelines import _slice_agents_alive + calls = spawner.k8s.wait_for_job_gone.call_args_list + assert [c.args[0] for c in calls] == ["job-p1", "job-p2"] + assert all(c.args[1] == "egg" for c in calls) + assert all(c.kwargs["timeout_s"] > 0 for c in calls) - backend = MagicMock() - backend.list_containers.side_effect = RuntimeError("k8s unreachable") - spawner = MagicMock() - spawner.backend = backend + def test_shares_one_budget_across_the_cohort(self): + """The deadline is per-slice, not per-Job: a pathological cohort + cannot stall the bootstrap pass for N x the budget.""" + from unittest.mock import patch + + from models import ContainerStatus + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = self._make_spawner( + [ + self._make_container_info("p1", ContainerStatus.RUNNING), + self._make_container_info("p2", ContainerStatus.RUNNING), + ] + ) + elapsed = {"t": 0.0} + + def _slow_wait(name, namespace, timeout_s): + elapsed["t"] += timeout_s + return True - assert _slice_agents_alive(spawner, "pipeline-x", "slice-1") is False + spawner.k8s.wait_for_job_gone.side_effect = _slow_wait - def test_filters_by_pipeline_and_slice_labels(self): - """Helper must query with both labels to avoid false-positive on - a different slice in the same pipeline.""" + with patch("routes.pipelines.time.monotonic", side_effect=[0.0, 0.0, 5.0]): + _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") + + # Second Job gets budget minus what the first consumed, not a + # fresh full budget. + second = spawner.k8s.wait_for_job_gone.call_args_list[1] + assert ( + second.kwargs["timeout_s"] + < spawner.k8s.wait_for_job_gone.call_args_list[0].kwargs["timeout_s"] + ) + + def test_unobserved_teardown_still_returns_the_reaped_handles(self): + """A Job still terminating after the budget is reported as + unconfirmed, but the reap still succeeded and the slice must + still be re-driven — an unobserved teardown is not a failure to + reap, and blocking recovery on it would be worse than the race.""" from models import ContainerStatus - from routes.pipelines import _slice_agents_alive + from routes.pipelines import _reap_orphaned_slice_jobs - backend = MagicMock() - backend.list_containers.return_value = [ - self._make_container_info("p1", ContainerStatus.RUNNING), - ] - spawner = MagicMock() - spawner.backend = backend + spawner = self._make_spawner( + [self._make_container_info("p1", ContainerStatus.RUNNING)], + gone=False, + ) + + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == ["job-p1"] - _slice_agents_alive(spawner, "pipeline-x", "slice-2") + def test_waiter_error_is_swallowed(self): + """The wait is best-effort: a raising waiter must not abort the + bootstrap pass.""" + from models import ContainerStatus + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = self._make_spawner([self._make_container_info("p1", ContainerStatus.RUNNING)]) + spawner.k8s.wait_for_job_gone.side_effect = RuntimeError("apiserver down") + + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == ["job-p1"] + + def test_pod_without_job_name_is_removed_but_not_waited_on(self): + """A listing that supplies no ``job_name`` leaves only the + container id to remove by. That id is a Pod name, and + ``wait_for_job_gone`` would normalize it into a Job that never + existed — a 404 reads as "gone" and would fake a confirmation. + So it is removed, counted as unobserved, and not waited on.""" + from models import ContainerStatus + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = self._make_spawner( + [self._make_container_info("p1", ContainerStatus.RUNNING, job_name=None)] + ) + + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == ["p1"] + spawner.remove_agent_container.assert_called_once_with( + "p1", force=True, cleanup_session=True + ) + spawner.k8s.wait_for_job_gone.assert_not_called() + + def test_missing_wait_helper_does_not_break_the_reap(self): + """A spawner without a usable ``wait_for_job_gone`` (older + backend, test double) degrades to the pre-#3685-review behaviour + rather than raising.""" + from models import ContainerStatus + from routes.pipelines import _reap_orphaned_slice_jobs + + spawner = MagicMock() + spawner.list_slice_jobs.return_value = [ + self._make_container_info("p1", ContainerStatus.RUNNING) + ] + spawner.k8s = None - # Verify the label selector included both pipeline and slice - call_kwargs = backend.list_containers.call_args.kwargs - assert "labels" in call_kwargs - labels = call_kwargs["labels"] - assert labels["egg.pipeline.id"] == "pipeline-x" - assert labels["egg.slice.id"] == "slice-2" + assert _reap_orphaned_slice_jobs(spawner, "pipeline-x", "slice-1") == ["job-p1"] class TestSliceHasPendingDecision: diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index 941d0a05f..8101ee83d 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -1950,41 +1950,50 @@ def test_concurrent_wave_no_op_collapse_elides_second_push(self) -> None: # --------------------------------------------------------------------------- -# #2914 — restart_phase resume guard: bootstrap must verify pods are alive -# before mark_spawned, or the pipeline wedges with no agents running. +# #3685: bootstrap must RE-DRIVE a resume-classified slice, never park it. # --------------------------------------------------------------------------- -class TestBootstrapResumeAliveGuard: - """#2914 — Layer-C resume classification must re-verify against k8s. +class TestBootstrapResumeRedrive: + """#3685: Layer-C ``"resume"`` must leave the slice admissible. The bootstrap classifier returns ``"resume"`` for IN_PROGRESS slices - with commits on the integration branch. Without the alive-guard, - that flips the scheduler slice to RUNNING via ``mark_spawned`` — - correct under normal restart, but wedging after ``restart_phase`` - tore the container cohort down (contract still shows IN_PROGRESS + - commits but no pods exist). The guard re-queries the spawner; if - no live pods are found the slice is treated as fresh so the run - loop re-yields READY and respawns a new cohort. - - These tests close the integration gap that - ``TestSliceAgentsAlive`` (unit) cannot — exercising the actual - call site at the Layer-C ``resume`` branch in - ``routes/pipelines.py:_run_implement_phase_slices`` end-to-end. + with commits on the integration branch. It used to answer that by + calling ``scheduler.mark_spawned``, on the theory that the surviving + agent pods carried the slice forward. Nothing does: post-#3164 the + orchestrator-owned BRC event loop is the only dispatcher and + ``spawn_all`` is the only thing that starts one, while + ``_run_one_slice`` is the only caller of ``scheduler.record_complete`` + for a running slice. ``mark_spawned`` therefore parked the slice in + scheduler-RUNNING with no dispatcher and no completer: ``iter_ready`` + never re-yielded it, it held a ``max_parallel_slices`` slot, and + ``all_done()`` could never turn true, so the pipeline reported + ``running`` with zero pods indefinitely. + + The tests assert the same invariant from the two sides of the old + liveness guard (#2914): live pods or not, the slice must reach + ``_run_concurrent_phase``. The live-pods cases additionally pin the + reap, so an orphaned Job does not race the fresh cohort on the + role's worktree (#3337) — and pin it for the ``"fresh"`` + classification too, since the reap runs above the classification + switch (see ``test_fresh_with_live_pods_also_reaps_orphans``). """ - def _make_spawner(self, *, live_pods: list[Any]) -> MagicMock: - """Build a spawner whose backend.list_containers returns the - given pods and whose gateway probe reports the integration - branch has commits on origin (classifier → ``"resume"``).""" + def _make_spawner(self, *, live_pods: list[Any], has_commits: bool = True) -> MagicMock: + """Build a spawner whose ``list_slice_jobs`` returns the given + pods. ``has_commits`` drives the gateway probe and therefore the + classification: a non-None SHA → ``"resume"``, ``None`` → + ``"fresh"``.""" spawner = MagicMock() spawner.gateway = MagicMock() spawner.gateway.create_slice_pr.return_value = "https://example/pr/1" spawner.gateway.is_slice_branch_merged_into_parent.return_value = False - # Non-None SHA → classifier sees commits on origin → "resume". - spawner.gateway.get_remote_branch_sha.return_value = "deadbeef" * 5 + spawner.gateway.get_remote_branch_sha.return_value = "deadbeef" * 5 if has_commits else None + spawner.list_slice_jobs.return_value = live_pods + spawner.k8s.namespace = "egg" + spawner.k8s.wait_for_job_gone.return_value = True backend = MagicMock() - backend.list_containers.return_value = live_pods + backend.list_containers.return_value = [] spawner.backend = backend return spawner @@ -1996,15 +2005,14 @@ def _make_container_info(container_id: str, status: Any) -> Any: container_id=container_id, container_name=f"egg-{container_id}", status=status, + job_name=f"job-{container_id}", ) - def test_resume_with_no_live_pods_respawns_fresh(self) -> None: - """Classifier → ``"resume"`` but ``list_containers`` returns [] - → slice must run through the regular path (i.e. - ``_run_concurrent_phase`` IS called). Without the guard, - ``mark_spawned`` would have flipped the slice to RUNNING and - ``iter_ready`` would never re-yield it, leaving the pipeline - wedged.""" + def _run(self, spawner: MagicMock) -> tuple[int, MagicMock, MagicMock]: + """Drive the bootstrap + run loop over one non-COMPLETE slice; + returns the exit code, the patched ``_run_concurrent_phase`` + mock, and the patched module logger (for audit-line assertions). + """ pipeline = _make_pipeline() slice1 = _make_slice("slice-1", tasks=[_make_task("task-1-1")]) slice1.status = SliceStatus.IN_PROGRESS @@ -2017,16 +2025,15 @@ def test_resume_with_no_live_pods_respawns_fresh(self) -> None: patch( "routes.pipelines._run_concurrent_phase", return_value=(0, "ok") ) as mock_run_phase, + patch("routes.pipelines.logger") as mock_logger, patch("orchestrator.peer_consensus.remove_peer_consensus_tracker"), - # No tracker → classifier's consensus_complete=False - # branch → "resume". + # No tracker → classifier's consensus_complete=False branch. patch( "routes.pipelines._lookup_peer_consensus_tracker_or_none", return_value=None, ), ): mock_start_recon.return_value = (MagicMock(), threading.Event()) - spawner = self._make_spawner(live_pods=[]) exit_code, _ = _run_implement_phase_slices( pipeline_id=pipeline.id, pipeline=pipeline, @@ -2039,22 +2046,113 @@ def test_resume_with_no_live_pods_respawns_fresh(self) -> None: certs_volume=None, worktree_repo_path=Path("/tmp/x"), ) + return exit_code, mock_run_phase, mock_logger + + @staticmethod + def _bootstrap_audit_kwargs(mock_logger: MagicMock) -> dict[str, Any]: + """Pull the Layer-C classification audit line's kwargs.""" + for call in mock_logger.info.call_args_list: + if call.args and "classified non-COMPLETE slices" in call.args[0]: + return call.kwargs + raise AssertionError("Layer-C classification audit line was never emitted") + + def test_resume_with_no_live_pods_is_redriven(self) -> None: + """Classifier → ``"resume"`` and no pods survived the restart + (the ordinary orchestrator-recycle case): the slice must run + through the regular path so its event loop starts.""" + spawner = self._make_spawner(live_pods=[]) + + exit_code, mock_run_phase, mock_logger = self._run(spawner) + assert exit_code == 0 - # The fix: with no live pods, the slice must still spawn fresh. invoked = {c.kwargs["slice_id"] for c in mock_run_phase.call_args_list} assert invoked == {"slice-1"}, ( - "resume-classified slice with no live pods must be respawned — " - "without the #2914 guard, mark_spawned would wedge the pipeline" + "resume-classified slice must be re-driven; mark_spawned would " + "park it in scheduler-RUNNING with no event loop and no completer" ) - # Liveness probe must have actually fired for the slice (label - # selector includes the slice id, not just the pipeline). - list_calls = spawner.backend.list_containers.call_args_list - slice_labelled_calls = [ - c for c in list_calls if c.kwargs.get("labels", {}).get("egg.slice.id") == "slice-1" - ] - assert slice_labelled_calls, ( - "_slice_agents_alive must call list_containers with the slice label" + # Nothing live → nothing to reap. + spawner.remove_agent_container.assert_not_called() + spawner.k8s.wait_for_job_gone.assert_not_called() + # The audit line is the operator's only handle on this path: + # ``redriven`` replaces the old ``resumed`` / ``reclassified_fresh`` + # pair, and a slice id here that never shows up in a subsequent + # spawn is the wedge signature #3685 fixed. + audit = self._bootstrap_audit_kwargs(mock_logger) + assert audit["redriven"] == ["slice-1"] + assert audit["consensus_complete_unrecorded"] == [] + assert audit["blocked"] == [] + assert audit["corrupt"] == [] + + def test_resume_with_live_pods_is_redriven_and_reaps_orphans(self) -> None: + """A pod that outlived the restart does NOT mean the slice is + progressing: its event loop died with the previous process, so + no one will observe its termination or derive its next event. + The slice is still re-driven, and the orphan is force-reaped + first so it does not race the fresh cohort on the role's + worktree (#3337).""" + from models import ContainerStatus + + spawner = self._make_spawner( + live_pods=[self._make_container_info("orphan-1", ContainerStatus.RUNNING)] + ) + + exit_code, mock_run_phase, mock_logger = self._run(spawner) + + assert exit_code == 0 + invoked = {c.kwargs["slice_id"] for c in mock_run_phase.call_args_list} + assert invoked == {"slice-1"}, ( + "a live pod is not a live dispatcher: the slice must still be " + "re-driven so a BRC event loop exists for it (#3685)" + ) + assert self._bootstrap_audit_kwargs(mock_logger)["redriven"] == ["slice-1"] + spawner.remove_agent_container.assert_called_once_with( + "job-orphan-1", force=True, cleanup_session=True + ) + # The reap query must be slice-scoped, not pipeline-wide. + assert [c.args[1] for c in spawner.list_slice_jobs.call_args_list] == ["slice-1"] + # ``force=True`` only orders the delete; the bootstrap must + # observe the Job actually gone before admitting the slice, + # otherwise the orphan is still holding the worktree the fresh + # cohort is about to reset (#3685 review round 2). + spawner.k8s.wait_for_job_gone.assert_called_once() + assert spawner.k8s.wait_for_job_gone.call_args.args[0] == "job-orphan-1" + + def test_fresh_with_live_pods_also_reaps_orphans(self) -> None: + """Regression guard for the reap's placement. + + Wiring the reap into the ``"resume"`` arm alone left the #3337 + worktree race wide open on the ``"fresh"`` path — and ``"fresh"`` + is not the rare case it looks like: an orchestrator recycle + before the slice's first push lands there, and so does any + transient gateway probe failure, which + ``_classify_non_complete_slice`` deliberately defaults to + ``has_commits=False``. Both leave the slice READY with the orphan + still Running. Commits-on-origin is orthogonal to Job liveness, + so the reap runs above the classification switch and this test + pins it there. + """ + from models import ContainerStatus + + spawner = self._make_spawner( + live_pods=[self._make_container_info("orphan-1", ContainerStatus.RUNNING)], + has_commits=False, + ) + + exit_code, mock_run_phase, mock_logger = self._run(spawner) + + assert exit_code == 0 + invoked = {c.kwargs["slice_id"] for c in mock_run_phase.call_args_list} + assert invoked == {"slice-1"} + spawner.remove_agent_container.assert_called_once_with( + "job-orphan-1", force=True, cleanup_session=True ) + spawner.k8s.wait_for_job_gone.assert_called_once() + # "fresh" is not a re-drive of recorded work, so the slice must + # NOT appear under ``redriven`` — the reap fired without the + # classification changing. + for call in mock_logger.info.call_args_list: + if call.args and "classified non-COMPLETE slices" in call.args[0]: + assert call.kwargs["redriven"] == [] # ---------------------------------------------------------------------------