Bail out of the HITL gate when the operator cancels

orchestrator/routes/pipelines/__init__.py

@@ -1391,6 +1391,7 @@ def stream_pipeline(pipeline_id: str) -> Response:
     _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

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
@@ -280,6 +305,25 @@ 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``.
+            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):

orchestrator/routes/pipelines/_ledger.py

@@ -555,9 +555,21 @@ def _queue_and_await_contract_decisions(
     pipeline_id: str,
     pipeline_identifier: int | str,
     phase: _pkg.PipelinePhase,
+    cancelled: _pkg.Callable[[], bool] | None = None,
 ) -> int:
     """Promote unresolved contract decisions/feedback into the orchestrator queue.
 
+    ``cancelled`` is an optional predicate the caller supplies to answer "has
+    the operator cancelled this pipeline?". It is consulted after each
+    blocking wait and stops the remaining waits when it returns True. A cancel
+    sweeps only the decisions already *in* the queue, so a cancel that lands
+    while pass 1 is still queueing this batch leaves the later entries PENDING
+    with nobody to cancel them — without this, the next
+    ``wait_for_decision`` would block for the process lifetime (#3633
+    review). Callers must still re-check the cancel themselves once this
+    returns: the early return is deliberately indistinguishable from a
+    zero-resolution round.
+
     Returns the number of contract decisions/feedback this call surfaced and
     the operator *resolved* this round — the converge-before-advance signal
     (#3392). Decisions that were surfaced but came back non-RESOLVED (e.g. the
@@ -717,6 +729,14 @@ def _save_contract_update(mutator: _pkg.Callable[[_pkg.Any], bool]) -> None:
     resolved_count = 0
     for contract_id, queued in queued_decisions:
         resolved = dq.wait_for_decision(queued.id)
+        if cancelled is not None and cancelled():
+            _pkg.logger.info(
+                "Contract decision bridge abandoned: pipeline cancelled (#3633)",
+                pipeline_id=pipeline_id,
+                phase=phase_value,
+                resolved_count=resolved_count,
+            )
+            return resolved_count
         if resolved.status != _pkg.DecisionStatus.RESOLVED:
             continue
         resolved_count += 1
@@ -737,6 +757,14 @@ def _apply(latest: _pkg.Any, _cd_id: str = contract_id, _res: str = resolution_s
     feedback_resolved = False
     if queued_feedback is not None and pending_feedback is not None:
         resolved = dq.wait_for_decision(queued_feedback.id)
+        if cancelled is not None and cancelled():
+            _pkg.logger.info(
+                "Contract feedback bridge abandoned: pipeline cancelled (#3633)",
+                pipeline_id=pipeline_id,
+                phase=phase_value,
+                resolved_count=resolved_count,
+            )
+            return resolved_count
         if resolved.status == _pkg.DecisionStatus.RESOLVED:
             feedback_resolved = True
             answers: dict[str, str] = {}
@@ -924,6 +952,27 @@ def _set_status(status: _pkg.PipelineStatus) -> _pkg.Pipeline:
             )
 
         resolved = dq.wait_for_decision(decision.id)
+        # ...unless what unblocked the wait was the operator cancelling the
+        # pipeline, which sweeps this decision to CANCELLED. Restoring RUNNING
+        # here would overwrite the persisted CANCELLED that every driver work
+        # loop keys on, re-admitting the run the operator just stopped
+        # (#3633 review). Bail before the status write, not after it.
+        #
+        # Skipping the write is necessary but not sufficient: ``gated`` is the
+        # same value an ordinary gating returns, so the *caller* has to re-read
+        # the cancel and stop the driver. ``_run_implement_advance`` does, and
+        # must keep doing so — IMPLEMENT is terminal, so a driver that merely
+        # falls through here writes COMPLETE over the CANCELLED this bail just
+        # preserved (#3633 review round 2).
+        if _pkg._pipeline_cancelled(store, pipeline_id):
+            _pkg.logger.info(
+                "Unresolved-gap gate: pipeline cancelled while awaiting the "
+                "operator — leaving the persisted CANCELLED intact (#3633)",
+                pipeline_id=pipeline_id,
+                phase=phase.value,
+            )
+            return gated
+
         # Restore RUNNING now the gate cleared (re-set to AWAITING_HUMAN
         # above on the next loop if gaps remain).
         _set_status(_pkg.PipelineStatus.RUNNING)

orchestrator/routes/pipelines/_run_concurrent.py

@@ -309,8 +309,11 @@ def _live_pipeline_phase() -> str:
     # Last chance to notice a cancel before minting a cohort (#3633 review).
     # The slice loop's guard runs at the top of its tick; everything between
     # there and here — contract load, per-role prompt building (draft reads,
-    # BRC history, git diffs), gateway session + worktree setup, integration
-    # branch creation — takes tens of seconds. A cancel landing in that window
+    # BRC history, git diffs), gateway session + worktree setup — takes tens
+    # of seconds. (Integration-branch creation is in that window too, but it
+    # happens back in ``_run_implement`` before this function is called, so
+    # this guard cannot prevent it — only the slice loop's own guard can.)
+    # A cancel landing in that window
     # runs the route's teardown BEFORE these Jobs exist, so nothing would reap
     # them: no reconciler acts on CANCELLED, and ``cleanup_pipeline`` only
     # re-runs on an operator DELETE. Re-read the status here so the cohort is