Stop the driver when a pipeline is cancelled (#3633)

orchestrator/CLAUDE.md

@@ -278,7 +278,7 @@ Pure refactor, no behaviour change: the 27 method bodies are AST-identical to th
 | Prompt building: `_prompt_phase.py` (1,407), `_prompt_agent.py` (1,346), `_prompt_review.py` (765), `_prompt_reviewer.py` (593) | Phase-prompt, agent-prompt, review-prompt and reviewer-prep prompt assembly | `build_phase_prompt`, `build_agent_prompt`, `build_review_prompt`, `build_reviewer_prompt` |
 | Readers / synthesis: `_criteria.py` (961), `_drafts.py` (758), `_reviews.py` (186), `_context_pr.py` (1,220), `_brc_history.py` (961), `_populate.py` (1,460) | Review-criteria builders, draft-path + source-branch-artifact readers, review-verdict readers, context-PR composition, BRC-history readers, plan-draft synthesis + contract population | `build_review_criteria`, `read_draft_path`, `read_review_verdicts`, `compose_context_pr`, `populate_contract` |
 | State / lifecycle: `_slice_state.py` (1,094), `_statefiles.py` (613), `_worktree_sync.py` (1,393), `_slice_completion.py` (135), `_lifecycle_helpers.py` (339), `_status_view.py` (391), `_status_wait.py` (147) | Slice-DAG state helpers, statefile read/write, worktree-sync, slice-completion, lifecycle helpers, status view + long-poll wait | `slice_state`, `sync_worktree`, `complete_slice`, `status_view`, `wait_for_status` |
-| Decisions / overseer: `_ledger.py` (1,364), `_decision_candidates.py` (159), `_decisions.py` (259), `_resolve.py` (196), `_hitl_rerun.py` (337), `_overseer.py` (740), `_alerts.py` (1,277), `_pod_liveness.py` (228), `_first_principles.py` (247) | Decision-ledger + gap-gate + apply-handoff, considered-candidate rendering + refine→plan deferral handoff + ledger-summary persistence (#3526), HITL + divergence-reconcile decisions, decision resolution, HITL rerun, overseer detection-plane, divergence/alert/timeout emission, live-pod guarding, first-principles review seed | `register_decision`, `resolve_decision`, `rerun_hitl`, `detect_divergence`, `guard_live_pods` |
+| Decisions / overseer: `_ledger.py` (1,364), `_contract_bridge.py` (299), `_decision_candidates.py` (159), `_decisions.py` (259), `_resolve.py` (196), `_hitl_rerun.py` (337), `_overseer.py` (740), `_alerts.py` (1,277), `_pod_liveness.py` (228), `_first_principles.py` (247) | Decision-ledger + gap-gate + apply-handoff, the contract-decision bridge (#1889 — split out of `_ledger.py` to stay under the file-size cap), considered-candidate rendering + refine→plan deferral handoff + ledger-summary persistence (#3526), HITL + divergence-reconcile decisions, decision resolution, HITL rerun, overseer detection-plane, divergence/alert/timeout emission, live-pod guarding, first-principles review seed | `register_decision`, `_queue_and_await_contract_decisions`, `resolve_decision`, `rerun_hitl`, `detect_divergence`, `guard_live_pods` |
 | PR / drivers / salvage: `_drivers.py` (263), `_stacked_pr.py` (251), `_salvage.py` (71) | Pipeline-driver lifecycle helpers, stacked-PR assembly, agent-output salvage | `pipeline_drivers`, `build_stacked_pr`, `salvage_agent_output` |
 
 Pure refactor, no behaviour change: every route handler body, helper, and constant is AST-identical to the pre-split file (modulo the sanctioned `_pkg.`-prefixing and the decorator relocation onto thin wrappers). Patch seams preserved: the 16 `@pipelines_bp.route` decorators stay in the barrel so the URL rule → handler map registers identically; the private submodules reach the test-patched module globals via `import routes.pipelines as _pkg`, and the barrel re-exports every externally-referenced symbol across the dominant back-compat import surface (~137 referencing files repo-wide; the audited ~57 distinct `patch("routes.pipelines.<name>")` targets), so both `from routes.pipelines import X` and `patch("routes.pipelines.X")` resolve unchanged — the existing dense seam coverage (`test_consensus_polling`, `test_brc_nack_iteration`, `test_concurrent_*`, `test_advance_phase_*`) stays green. `_run_pipeline` becomes a thin loop delegating to per-phase handlers with no transition-ordering change. **Packaging-neutral:** `orchestrator/routes/` is already shipped by the recursive `COPY orchestrator/routes/ ./routes/` (Dockerfile:45), so the new submodules are auto-included — no Dockerfile change. `pipelines.py`'s allowlist entry — the **LAST** in the program — is dropped, so `scripts/file-size-allowlist.yaml`'s `files:` map is now **EMPTY**: the terminal acceptance criterion of the file-size decomposition program, closing #3312.

orchestrator/event_loop/__init__.py

@@ -372,10 +372,13 @@ class EventDecision:
     ``timing`` is a structured mapping for the slice-4 latency budget on a
     fresh spawn, ``None`` otherwise. ``blocked`` (#3496) names why a
     spawn-action decision did not spawn when the block is operator-relevant:
-    ``"exhausted"`` (terminal retry-budget exhaustion) or ``"parked"``
-    (#3548 — a no-op-park that only the retry heartbeat will release), so
-    the all-arms wedge detections can distinguish them from the benign
-    not-spawned shapes (dedupe, backoff).
+    ``"exhausted"`` (terminal retry-budget exhaustion), ``"parked"``
+    (#3548 — a no-op-park that only the retry heartbeat will release), or
+    ``"stopped"`` (#3633 — the loop was stopped mid-tick, typically by a
+    cancel), so the all-arms wedge detections can distinguish them from the
+    benign not-spawned shapes (dedupe, backoff). Only ``"exhausted"`` and
+    ``"parked"`` count toward a wedge; ``"stopped"`` is a teardown, not a
+    condition an operator can resolve.
     """
 
     role: str

orchestrator/event_loop/_loop.py

@@ -616,6 +616,22 @@ def _handle_role(self, role: str) -> EventDecision:
         # wait / unknown — nothing to spawn.
         return EventDecision(role=role, action=action)
 
+    # #3633: never spawn once the loop has been stopped. ``run()`` checks the
+    # stop event between ticks, but ``stop()`` is also called from outside the
+    # loop's own thread — the cancel route stops every live loop for a
+    # pipeline synchronously — so a stop that lands mid-tick would otherwise
+    # still get one final cohort of one-shot Jobs out the door. Re-checking
+    # immediately before the spawn decision closes that window.
+    if self._stop.is_set():
+        logger.info(
+            "event-loop: spawn blocked, loop is stopping",
+            pipeline_id=self.pipeline_id,
+            slice_id=self.slice_id,
+            role=role,
+            action=action,
+        )
+        return EventDecision(role=role, action=action, spawned=False, blocked="stopped")
+
     identity = _pkg.event_identity(action, payload)
     key = _pkg.compute_dedupe_key(
         self.pipeline_id, self.slice_id, self.phase, role, action, identity

orchestrator/routes/decisions/_handlers.py

@@ -1060,11 +1060,19 @@ def _read_redirect_seed_from_contract(pipeline_id: str, decision_id: Any) -> str
 
 
 def _cancel_pipeline_in_process(pipeline_id: str, *, reason: str) -> None:
-    """Cancel a pipeline from a resolution hook (status CANCELLED + cleanup).
+    """Cancel a pipeline from a resolution hook (status CANCELLED + loop stop).
 
     Mirrors the inline cancel pattern used elsewhere: flip status under the
-    pipeline state lock, emit ``PIPELINE_CANCELLED``, and cancel pending
-    decisions so any ``wait_for_decision`` unblocks.
+    pipeline state lock, stop the pipeline's live BRC event loops, emit
+    ``PIPELINE_CANCELLED``, and cancel pending decisions so any
+    ``wait_for_decision`` unblocks.
+
+    This is the second place CANCELLED originates (the PATCH route is the
+    first), so it needs the same #3633 loop teardown: without it the live
+    loops keep deriving arms and requesting one-shot Jobs, and only the
+    driver loops' own CANCELLED re-reads — a poll interval later — would
+    catch it. Note this hook does NOT tear down containers; the driver's
+    cancel bail and the operator's own cleanup own that half.
     """
     from models import PipelineStatus
     from state_store import get_pipeline_state_lock
@@ -1077,6 +1085,19 @@ def _cancel_pipeline_in_process(pipeline_id: str, *, reason: str) -> None:
         pipeline.error = reason
         store.update_pipeline(pipeline_id, pipeline.model_dump(mode="json"))
 
+    try:
+        # Deferred import: routes.pipelines is too heavy to bind at module
+        # import time (same reason as the other hooks in this file).
+        from routes.pipelines import _stop_pipeline_event_loops
+
+        _stop_pipeline_event_loops(pipeline_id, reason="pipeline_cancelled")
+    except Exception:
+        logger.warning(
+            "Failed to stop BRC event loops after first-principles cancel",
+            pipeline_id=pipeline_id,
+            exc_info=True,
+        )
+
     try:
         _pkg.emit_event(
             EventType.PIPELINE_CANCELLED,

orchestrator/routes/pipelines/__init__.py

@@ -1168,6 +1168,9 @@ def stream_pipeline(pipeline_id: str) -> Response:
     _refresh_context_pr_body,
     _repos_with_slices,
 )
+from ._contract_bridge import (  # noqa: E402,F401
+    _queue_and_await_contract_decisions,
+)
 from ._criteria import (  # noqa: E402,F401
     _get_agent_design_criteria,
     _get_code_review_criteria,
@@ -1246,7 +1249,6 @@ def stream_pipeline(pipeline_id: str) -> Response:
     _ledger_attestation_rerun_directive,
     _next_phases_for_epic,
     _persist_phase_gate_resolution,
-    _queue_and_await_contract_decisions,
     _sync_pipeline_decisions_to_contract,
     _unwrap_choice_resolution,
     _write_apply_phase_handoff,
@@ -1258,6 +1260,7 @@ def stream_pipeline(pipeline_id: str) -> Response:
     _compute_gateway_mode,
     _mark_pipeline_records_terminated,
     _normalize_submission_repos,
+    _stop_pipeline_event_loops,
     _sync_contract_phase_to_pipeline,
 )
 from ._overseer import (  # noqa: E402,F401
@@ -1382,14 +1385,15 @@ def stream_pipeline(pipeline_id: str) -> Response:
 )
 from ._run_concurrent_support import (  # noqa: E402,F401
     _latest_proposal_ts_impl,
+    _phase_bail_reason_impl,
     _record_container_exit_impl,
     _record_spawned_agents_impl,
     _retry_transient_spawn_failures_impl,
     _stop_running_containers_impl,
-    _superseded_by_restart_impl,
     _update_agents_complete_impl,
 )
 from ._run_hitl_gate import (  # noqa: E402,F401
+    _gate_wait_cancelled,
     _run_hitl_gate_converge,
 )
 from ._run_implement import (  # noqa: E402,F401
@@ -1431,8 +1435,9 @@ def stream_pipeline(pipeline_id: str) -> Response:
 )
 from ._run_support import (  # noqa: E402,F401
     _clear_stale_impasses_for_producers,
+    _park_at_gate_unless_cancelled,
     _parse_resolution,
-    _pipeline_superseded_by_restart,
+    _pipeline_cancelled,
     _spawn_and_wait,
 )
 from ._salvage import (  # noqa: E402,F401

orchestrator/routes/pipelines/_alerts.py

@@ -111,6 +111,10 @@ def _fail_pipeline_after_divergence_abort(
     ``pre_event_hook`` runs after the FAILED-write but before the public
     ``pipeline.failed`` broadcast (the post-phase site uses it to tear down
     the per-phase overseer container).
+
+    On an already-CANCELLED pipeline the status write and the broadcast are
+    both skipped — ``pre_event_hook`` still runs — so an operator cancel that
+    unblocked the reconcile pause is not rewritten as a failure (#3633).
     """
     phase_label = phase.value if phase is not None else "current phase"
     reason = (
@@ -124,6 +128,25 @@ def _fail_pipeline_after_divergence_abort(
         f"preserved under {backup_ref or '(backup ref write failed)'} "
         f"({len(local_only_commit_shas)} commit(s))."
     )
+    # ``_sync_worktree_reconciling_divergence`` also returns ``aborted=True``
+    # when what unblocked its ``wait_for_decision`` was the operator
+    # cancelling the pipeline (#3633).  That is a stop, not a failure: the
+    # FAILED write below would overwrite the persisted CANCELLED every driver
+    # work loop keys on, and the ``pipeline.failed`` broadcast would report a
+    # failure the operator never caused.  Run ``pre_event_hook`` anyway — the
+    # per-phase overseer teardown it carries is wanted on either exit — then
+    # return, leaving CANCELLED intact for the caller to stop on.
+    if _pkg._pipeline_cancelled(store, pipeline_id):
+        _pkg.logger.info(
+            "Divergence reconcile ended on an operator cancel — leaving the "
+            "persisted CANCELLED intact instead of pinning FAILED (#3633)",
+            pipeline_id=pipeline_id,
+            phase=phase_label,
+        )
+        if pre_event_hook is not None:
+            pre_event_hook()
+        return
+
     with _pkg.get_pipeline_state_lock(pipeline_id):
         pipeline = store.load_pipeline(pipeline_id)
         if phase is not None:
@@ -173,11 +196,13 @@ def _sync_worktree_reconciling_divergence(
     nothing discarded.
 
     Returns ``(outcome, aborted)``.  ``aborted`` is True when the operator
-    chose "Abort pipeline" or the reconcile-pause budget was exhausted; the
-    caller should fail the pipeline via
-    :func:`_fail_pipeline_after_divergence_abort`.  When ``aborted`` is
-    False the worktree is reconciled (or never diverged) and the caller
-    proceeds normally.
+    chose "Abort pipeline", the reconcile-pause budget was exhausted, or the
+    operator cancelled the pipeline while this was blocked on the pause
+    (#3633); the caller should stop driving the phase via
+    :func:`_fail_pipeline_after_divergence_abort`, which pins FAILED for the
+    first two and is a status no-op for the third so the persisted CANCELLED
+    survives.  When ``aborted`` is False the worktree is reconciled (or never
+    diverged) and the caller proceeds normally.
 
     Only call this from inside the ``_run_pipeline`` loop thread, which is
     allowed to block; route handlers that cannot block use
@@ -212,30 +237,56 @@ def _sync_worktree_reconciling_divergence(
         # Persist the reconcile HITL and flip to AWAITING_HUMAN under the
         # (reentrant) state lock so a reader never sees AWAITING_HUMAN
         # without the pending decision.
+        #
+        # The status is re-read from inside that lock and a cancel short-
+        # circuits both writes. Two things go wrong otherwise: the park
+        # clobbers the operator's CANCELLED (so the post-wait check below
+        # reads back our own AWAITING_HUMAN and resumes the run), and the HITL
+        # is minted after the cancel route's one-time pending sweep, leaving
+        # ``wait_for_decision`` — an unbounded poll — blocking for the process
+        # lifetime (#3633 review round 3). The state lock is what makes this
+        # airtight: ``StateStore.update_pipeline``, the cancel route's
+        # persistence path, takes the same one.
+        _park_cancelled = False
+        decision = None
         with _pkg.get_pipeline_state_lock(pipeline_id):
             pipeline = store.load_pipeline(pipeline_id)
-            pipeline.status = PipelineStatus.AWAITING_HUMAN
-            if phase is not None:
-                phase_execution = pipeline.get_phase_execution(phase)
-                if phase_execution is not None:
-                    phase_execution.status = PipelineStatus.AWAITING_HUMAN
-            store.save_pipeline(pipeline)
-            decision = _pkg._persist_hitl_decision(
-                pipeline_id,
-                pipeline,
-                store,
-                question=_pkg._divergence_reconcile_hitl_question(
-                    pipeline_id=pipeline_id,
+            if pipeline.status == PipelineStatus.CANCELLED:
+                _park_cancelled = True
+            else:
+                pipeline.status = PipelineStatus.AWAITING_HUMAN
+                if phase is not None:
+                    phase_execution = pipeline.get_phase_execution(phase)
+                    if phase_execution is not None:
+                        phase_execution.status = PipelineStatus.AWAITING_HUMAN
+                store.save_pipeline(pipeline)
+                decision = _pkg._persist_hitl_decision(
+                    pipeline_id,
+                    pipeline,
+                    store,
+                    question=_pkg._divergence_reconcile_hitl_question(
+                        pipeline_id=pipeline_id,
+                        phase=phase,
+                        backup_ref=outcome.backup_ref,
+                        local_only_commit_shas=outcome.local_only_commit_shas,
+                        rebase_category=outcome.rebase_category,
+                        rebase_detail=outcome.rebase_detail,
+                    ),
+                    options=list(_pkg._DIVERGENCE_RECONCILE_HITL_OPTIONS),
                     phase=phase,
-                    backup_ref=outcome.backup_ref,
-                    local_only_commit_shas=outcome.local_only_commit_shas,
-                    rebase_category=outcome.rebase_category,
-                    rebase_detail=outcome.rebase_detail,
-                ),
-                options=list(_pkg._DIVERGENCE_RECONCILE_HITL_OPTIONS),
-                phase=phase,
-                context=_pkg._DIVERGENCE_RECONCILE_HITL_CONTEXT,
+                    context=_pkg._DIVERGENCE_RECONCILE_HITL_CONTEXT,
+                )
+        if _park_cancelled:
+            # Checked before the ``decision is None`` arm below: nothing was
+            # persisted on this path, and a cancel is not a persist failure.
+            _pkg.logger.info(
+                "Divergence reconcile pause: pipeline cancelled before the "
+                "pause was persisted — leaving the persisted CANCELLED "
+                "intact (#3633)",
+                pipeline_id=pipeline_id,
+                phase=phase_label,
             )
+            return outcome, True
         if decision is None:
             # Could not persist the HITL — fail closed rather than spin on
             # a pause the operator can never see.
@@ -280,6 +331,37 @@ def _sync_worktree_reconciling_divergence(
 
             dq.wait_for_decision(decision.id)
 
+            # The wait also returns when the operator cancels the pipeline —
+            # the cancel route sweeps every pending decision, this one
+            # included, with no resolution.  Bail before the RUNNING write
+            # below: restoring RUNNING would overwrite the persisted
+            # CANCELLED the driver keys on and re-admit the run the operator
+            # just stopped (#3633).  ``aborted=True`` is how the two callers
+            # spell "stop driving this phase"; the FAILED pin they route to
+            # is suppressed on a cancelled pipeline in
+            # ``_fail_pipeline_after_divergence_abort``.
+            #
+            # The phase box is deliberately left at AWAITING_HUMAN. Once the
+            # park above has landed, restoring it would mean a second write to
+            # a pipeline the operator has already stopped, and the only status
+            # that would be honest to write is the one already on the pipeline
+            # record. Leaving it records *where* the run stopped — parked at
+            # the reconcile gate — which is what an operator reading the phase
+            # timeline of a cancelled pipeline wants. This is uniform with
+            # every other gate: ``_gate_wait_cancelled`` and the gap /
+            # attestation gates all break without restoring their phase box
+            # either, so a cancelled pipeline consistently renders the gate it
+            # was sitting at (#3633 review round 4).
+            if _pkg._pipeline_cancelled(store, pipeline_id):
+                _pkg.logger.info(
+                    "Divergence reconcile pause: pipeline cancelled while "
+                    "awaiting the operator — leaving the persisted CANCELLED "
+                    "intact (#3633)",
+                    pipeline_id=pipeline_id,
+                    phase=phase_label,
+                )
+                return outcome, True
+
             resolved = dq.get_decision(decision.id)
             resolution = (resolved.resolution or "") if resolved is not None else ""
             if _pkg._divergence_reconcile_is_abort(resolution):