diff --git a/orchestrator/routes/pipelines/__init__.py b/orchestrator/routes/pipelines/__init__.py index 333faebb0..098e68122 100644 --- a/orchestrator/routes/pipelines/__init__.py +++ b/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 diff --git a/orchestrator/routes/pipelines/_alerts.py b/orchestrator/routes/pipelines/_alerts.py index 2bb70e20c..4e66a89dd 100644 --- a/orchestrator/routes/pipelines/_alerts.py +++ b/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): diff --git a/orchestrator/routes/pipelines/_ledger.py b/orchestrator/routes/pipelines/_ledger.py index 98304eeff..5efecd1d2 100644 --- a/orchestrator/routes/pipelines/_ledger.py +++ b/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) diff --git a/orchestrator/routes/pipelines/_run_concurrent.py b/orchestrator/routes/pipelines/_run_concurrent.py index 06241ac4c..c5dcaa8e6 100644 --- a/orchestrator/routes/pipelines/_run_concurrent.py +++ b/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 diff --git a/orchestrator/routes/pipelines/_run_hitl_gate.py b/orchestrator/routes/pipelines/_run_hitl_gate.py index 0cbfa785f..f0bb0d9d3 100644 --- a/orchestrator/routes/pipelines/_run_hitl_gate.py +++ b/orchestrator/routes/pipelines/_run_hitl_gate.py @@ -10,6 +10,63 @@ import routes.pipelines as _pkg # noqa: E402,F401 +def _gate_wait_cancelled(store, pipeline_id: str) -> bool: + """True when a gate's ``wait_for_decision`` was unblocked by an operator cancel (#3633). + + ``wait_for_decision`` returns on two very different events: a human + resolved the decision, or a cancel route called ``cancel_decision`` on it — + which the pipeline cancel does deliberately, under the comment "cancel any + pending decisions so ``wait_for_decision()`` unblocks" + (``_routes_crud.py``). Downstream the two were indistinguishable: a + cancelled decision carries no ``resolution``, and the empty string is a + member of ``_APPROVE_KEYWORDS``, so the gate read the operator's cancel as + an approval, took the "Approved — resume and advance" branch, wrote + ``RUNNING`` over the persisted ``CANCELLED``, and let the driver advance + into the next phase and mint a fresh cohort. + + That is the #3633 symptom reached through the one family of paths the + persisted-status layers cannot see: every gate re-reads a status the gate + block itself has already overwritten with ``AWAITING_HUMAN`` / + ``RUNNING``, so the loop-head check alone never fires. Each blocking wait + in the gate therefore has to ask this question on its own. + + **Why the returned decision's own status is deliberately not consulted.** + It reads like the sharper signal, and an earlier revision keyed on it, but + it over-fires in two ways that a persisted-status check does not: + + - ``routes/decisions/_lifecycle.py`` exposes a standalone endpoint that + cancels *one* decision without touching the pipeline. Treating that as a + pipeline cancel makes the driver exit and strands a live pipeline at + ``AWAITING_HUMAN`` with no waiter. + - the PATCH route sweeps pending decisions on ``FAILED`` too, so a + ``container_monitor`` false-positive ``FAILED`` — the case #1273 + deliberately carves out of ``_pipeline_cancelled`` — would bail here + anyway, bypassing that carve-out. + + Not bailing on ``FAILED`` is not the same as ignoring it, and the cost is + worth stating plainly: a ``FAILED`` PATCH sweeps the gate's decision, the + wait returns a decision with no ``resolution``, and the empty string is in + ``_APPROVE_KEYWORDS`` — so the gate reads it as an approval and advances + the phase. That is the lesser of the two evils (a false-positive + ``FAILED`` recovers to RUNNING under #1273, and stopping the driver on one + would strand a live pipeline), and no writer PATCHing ``status=FAILED`` + through that route is known today. If one ever appears, the fix belongs in + the resolution parsing — an unset ``resolution`` on a swept decision is not + an approval — not in a ``FAILED`` bail here. + + Both *real* cancel paths (the PATCH route in ``_routes_crud.py`` and + ``_cancel_pipeline_in_process`` in ``routes/decisions/_handlers.py``) + persist ``CANCELLED`` **before** sweeping the queue, so by the time a + swept wait returns the status this reads is already authoritative. Keep + that ordering if either route is ever restructured — it is what makes the + persisted status sufficient here. + + Delegates to ``_pipeline_cancelled``, inheriting its FAILED carve-out + (#1273) and its best-effort store-hiccup tolerance. + """ + return _pkg._pipeline_cancelled(store, pipeline_id) + + def _run_hitl_gate_converge( pipeline, *, @@ -166,6 +223,14 @@ def _run_hitl_gate_converge( _pkg._emit_pipeline_event(pipeline, "decision.created") _backstop_resolved = dq.wait_for_decision(_backstop.id) + if _pkg._gate_wait_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled while awaiting the decision-ledger " + "backstop — exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" _backstop_resolution = str( getattr(_backstop_resolved, "resolution", None) or "" ).strip() @@ -250,6 +315,26 @@ def _run_hitl_gate_converge( ) if _rerun_requested: return pipeline, "continue" # Re-enter outer loop → re-run phase + # The attestation gate blocks in its own ``wait_for_decision`` + # and, on any non-RESOLVED terminal status, deliberately "fails + # open to the phase gate" — a safe posture when a cancelled + # attestation only meant one more operator prompt. It is not safe + # once the cancel is the operator stopping the run: falling + # through writes ``AWAITING_HUMAN`` over the persisted + # ``CANCELLED`` and then queues a *fresh* phase_gate decision, + # minted after the cancel route already swept the queue, so + # nothing will ever cancel it and the wait below never returns. + # That leaks the driver thread for the process lifetime and skips + # the ``finally`` entirely — no cleanup, no worktree preservation + # (#3633 review). + if _pkg._gate_wait_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled while awaiting the explicit-none " + "attestation — exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" # Check for an existing pending phase_gate decision for this # phase. A prior agent-exit event may @@ -262,6 +347,55 @@ def _run_hitl_gate_converge( for d in pipeline.decisions ) + # Read the draft up front, before the reuse-vs-create branch below. + # Both ``draft_content`` and ``phase_label`` are read unconditionally + # further down — the follow-up decision in the "bare request changes" + # path uses them for its question and context — but they used to be + # bound only inside the ``else:`` arm. Resuming onto an existing + # pending gate (an orchestrator restart or a driver respawn while + # parked at a refine/plan gate) and then answering with a bare + # "request changes" therefore raised ``UnboundLocalError`` before the + # follow-up could be queued (#3633 review). Binding both here costs + # one draft read on the reuse path and makes the follow-up reachable + # from either branch. + phase_label = "analysis" if current_phase.value == "refine" else current_phase.value + draft_content = _pkg._read_phase_draft( + worktree_repo_path, + current_phase.value, + issue_number=pipeline.issue_number, + pipeline_id=pipeline_id, + branch=pipeline.branch, + ) + # Warn if draft is missing — the agent may not have written + # it to the expected path. See #1016. The warning stays scoped to + # the create arm, which is the arm that renders the draft into a new + # gate comment: the reuse arm never read the draft before the hoist + # above, so warning there would be new operator-facing noise about a + # draft nothing is about to show (#3633 review round 2). The + # placeholder is still bound on both arms — the follow-up prompt uses + # it as decision context regardless of which arm queued the gate. + if draft_content is None: + if existing_pending_gate: + _pkg.logger.debug( + "HITL gate: draft not found on work branch (reusing an " + "existing pending gate; not rendering a draft)", + pipeline_id=pipeline_id, + phase=current_phase.value, + worktree_path=str(worktree_repo_path), + ) + else: + _pkg.logger.warning( + "HITL gate: draft not found on work branch", + pipeline_id=pipeline_id, + phase=current_phase.value, + worktree_path=str(worktree_repo_path), + ) + draft_content = ( + f"**Warning**: No {phase_label} draft was found on the " + f"work branch. The agent may not have written the output " + f"to the expected path." + ) + if existing_pending_gate: _pkg.logger.info( "HITL gate: reusing existing pending phase_gate decision", @@ -278,30 +412,6 @@ def _run_hitl_gate_converge( and d.status == _pkg.DecisionStatus.PENDING ) else: - draft_content = _pkg._read_phase_draft( - worktree_repo_path, - current_phase.value, - issue_number=pipeline.issue_number, - pipeline_id=pipeline_id, - branch=pipeline.branch, - ) - phase_label = "analysis" if current_phase.value == "refine" else current_phase.value - - # Warn if draft is missing — the agent may not have written - # it to the expected path. See #1016. - if draft_content is None: - _pkg.logger.warning( - "HITL gate: draft not found on work branch", - pipeline_id=pipeline_id, - phase=current_phase.value, - worktree_path=str(worktree_repo_path), - ) - draft_content = ( - f"**Warning**: No {phase_label} draft was found on the " - f"work branch. The agent may not have written the output " - f"to the expected path." - ) - question = ( f"The {current_phase.value} phase has completed. " f"Please review the {phase_label} and approve to continue, " @@ -403,6 +513,26 @@ def _run_hitl_gate_converge( # Check resolution — did the human approve or request changes? resolved_decision = dq.get_decision(decision.id) + + # ...or did neither happen, because the operator cancelled the + # pipeline and the route cancelled this decision to unblock the wait + # above? Bail before any of the resolution parsing below: an unset + # resolution reads as an approval (``"" in _APPROVE_KEYWORDS``), and + # the approve branch rewrites the operator's CANCELLED to RUNNING and + # advances the phase — the #3633 spawn, through the one path the + # persisted-status layers cannot see. "break" leaves the driver loop + # the same way its own CANCELLED check at the loop head does, so the + # ``finally`` observes CANCELLED and preserves the worktrees + # ``restart_phase`` resumes from (#1725). + if _pkg._gate_wait_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled while awaiting the phase gate — " + "exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" + resolution = (resolved_decision.resolution or "").strip() # JSON-first resolution parsing: try structured payload before @@ -495,6 +625,14 @@ def _run_hitl_gate_converge( ) dq.wait_for_decision(followup.id) resolved_followup = dq.get_decision(followup.id) + if _pkg._gate_wait_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled while awaiting gate follow-up " + "specifics — exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" followup_resolution = (resolved_followup.resolution or "").strip() # Parse follow-up resolution (also JSON-first) @@ -592,6 +730,7 @@ def _run_hitl_gate_converge( pipeline_id, _pkg._pipeline_identifier(pipeline.issue_number, pipeline_id), current_phase, + cancelled=lambda: _pkg._gate_wait_cancelled(store, pipeline_id), ) except Exception as bridge_err: _pkg.logger.warning( @@ -601,6 +740,23 @@ def _run_hitl_gate_converge( error=str(bridge_err), ) + # The bridge is the *longest* human-latency window in the gate — one + # blocking wait per contract question, answered sequentially — and a + # cancel landing in it unblocks every one of them with no + # ``resolution``. The bridge only counts RESOLVED answers, so + # ``_decisions_resolved_this_round`` stays 0, the converge branch + # below is skipped, and control falls straight through to "Approved — + # resume and advance", which writes RUNNING over the operator's + # CANCELLED and advances the phase: #3633 verbatim (#3633 review). + if _pkg._gate_wait_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled while bridging contract decisions — " + "exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" + # Converge-before-advance (#3392): if the operator just # resolved one or more decisions, re-run the phase so the # documents reflect those resolutions and any decision the diff --git a/orchestrator/routes/pipelines/_run_phase_blocks.py b/orchestrator/routes/pipelines/_run_phase_blocks.py index 1e0b3e4e7..0cf77ced1 100644 --- a/orchestrator/routes/pipelines/_run_phase_blocks.py +++ b/orchestrator/routes/pipelines/_run_phase_blocks.py @@ -21,7 +21,9 @@ def _run_implement_advance( store, worktree_repo_path, ): - """IMPLEMENT-phase advance loop block (extracted verbatim; pure fall-through).""" + """IMPLEMENT-phase advance loop block. Returns ``(pipeline, action)``; + ``action == "break"`` -> the operator cancelled inside the gap gate and the + caller must leave the driver loop instead of advancing.""" if current_phase == _pkg.PipelinePhase.IMPLEMENT: try: gap_gated = _pkg._await_unresolved_gap_gate( @@ -34,6 +36,25 @@ def _run_implement_advance( pipeline.config.hitl_gates, ) pipeline = store.load_pipeline(pipeline_id) + # The gate bails out of its wait without restoring RUNNING when the + # operator cancelled mid-wait (#3633) — but it reports that through + # the same ``gated`` boolean an ordinary gating returns, so nothing + # downstream can tell the two apart. Re-read the persisted status + # here and hand the driver an explicit stop: without it the cancel + # is silently defaulted away, the commit+push below mutates the + # remote branch of a pipeline the operator just stopped, and + # IMPLEMENT being terminal (``PHASE_TRANSITIONS[IMPLEMENT] == []``) + # takes _run_pipeline straight into its "pipeline complete" branch, + # overwriting CANCELLED with COMPLETE and deleting the worktrees + # restart_phase resumes from (#3633 review round 2). + if _pkg._pipeline_cancelled(store, pipeline_id): + _pkg.logger.info( + "Unresolved-gap gate: pipeline cancelled while awaiting the " + "operator — stopping the driver without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" # The gate ran after the statefile commit+push above, so # when it changed the contract (operator resolved a gap, # or the override audit landed) the resolution is still @@ -88,7 +109,7 @@ def _run_implement_advance( phase=current_phase.value, error=str(gap_gate_err), ) - return pipeline + return pipeline, None def _run_plan_advance( diff --git a/orchestrator/routes/pipelines/_run_pipeline.py b/orchestrator/routes/pipelines/_run_pipeline.py index 6b843a140..73decf3f3 100644 --- a/orchestrator/routes/pipelines/_run_pipeline.py +++ b/orchestrator/routes/pipelines/_run_pipeline.py @@ -744,7 +744,10 @@ def _hook() -> None: # discarded — the local commits remain pinned under the backup # ref for offline recovery. ``pre_event_hook`` tears down the # per-phase overseer under its own lock before the public - # ``pipeline.failed`` event, matching the prior ordering. + # ``pipeline.failed`` event, matching the prior ordering. The + # helper also fields the operator-cancelled-during-the-pause case + # (#3633), where it tears down but leaves CANCELLED persisted + # instead of pinning FAILED; the ``break`` below is right for both. if post_phase_sync_aborted and post_phase_sync_outcome is not None: _pkg._fail_pipeline_after_divergence_abort( pipeline_id, @@ -901,7 +904,7 @@ def _hook() -> None: # not block — both options need a human, so blocking would # stall the pipeline indefinitely; the reactive CI check stays # the backstop there. - pipeline = _pkg._run_implement_advance( + pipeline, _gap_gate_action = _pkg._run_implement_advance( pipeline, current_phase=current_phase, gateway_mode=gateway_mode, @@ -911,6 +914,13 @@ def _hook() -> None: store=store, worktree_repo_path=worktree_repo_path, ) + if _gap_gate_action == "break": + # The operator cancelled while the gap gate was blocked in + # ``wait_for_decision`` (#3633). Leave the loop the same way the + # CANCELLED check at the loop head does. Falling through instead + # would reach the terminal-phase branch below — IMPLEMENT has no + # successor — and write COMPLETE over the operator's CANCELLED. + break # --- HITL gate: pause for human approval --- # Refine/plan are gated by the converge-before-advance loop @@ -941,6 +951,16 @@ def _hook() -> None: ) if _hitl_gate_action == "continue": continue + if _hitl_gate_action == "break": + # The operator cancelled while the gate was blocked in one of + # its ``wait_for_decision`` calls (#3633). Leave the loop the + # same way the CANCELLED check at the loop head does rather + # than advancing the phase: the gate parks at AWAITING_HUMAN + # and writes RUNNING back on its way out, so by the time + # control returns here the loop-head check would be reading a + # status the gate itself had already overwritten. It has to + # bail from inside. + break # ---------------------------------------------------------- # #2777 (cq-4, TASK-1-2) — inline ``_run_pipeline`` @@ -1008,7 +1028,29 @@ def _hook() -> None: ) if not next_phases: - # Terminal phase — pipeline complete + # Terminal phase — pipeline complete. + # + # ...unless the operator cancelled somewhere between the loop + # head and here. This branch is the last unguarded status write + # in the loop, and it is the one every park-and-resume block in + # a terminal phase falls into: the block bails without writing + # RUNNING back, the driver reads that as "nothing left to do", + # and CANCELLED becomes COMPLETE plus a "completed + # successfully" broadcast — after which the ``finally`` sees a + # non-CANCELLED status, leaves ``skip_cleanup`` False, and + # deletes the worktrees ``restart_phase`` resumes from (#3633). + # Re-read the persisted status here so the guard holds for + # future blocks too, not just today's gap gate. + if _pkg._pipeline_cancelled(store, pipeline_id): + _pkg.logger.info( + "Terminal phase reached on a cancelled pipeline — " + "leaving the persisted CANCELLED intact and skipping " + "the completion broadcast (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + break + with _pkg.get_pipeline_state_lock(pipeline_id): pipeline = store.load_pipeline(pipeline_id) pipeline.status = _pkg.PipelineStatus.COMPLETE diff --git a/orchestrator/routes/pipelines/_run_pipeline_setup.py b/orchestrator/routes/pipelines/_run_pipeline_setup.py index 2381af8e7..a2b3fdc93 100644 --- a/orchestrator/routes/pipelines/_run_pipeline_setup.py +++ b/orchestrator/routes/pipelines/_run_pipeline_setup.py @@ -690,7 +690,10 @@ def _resolve_worktree_repo( # Operator aborted the manual reconcile (or the pause # budget was exhausted). Fail the pipeline; the local # commits remain pinned under the backup ref for offline - # recovery — nothing was discarded. + # recovery — nothing was discarded. Also covers an operator + # cancel landing during the pause (#3633): the helper is a + # status no-op there, leaving CANCELLED persisted, and the + # ``return pipeline, True`` below stops the driver either way. _pkg._fail_pipeline_after_divergence_abort( pipeline_id, store, diff --git a/orchestrator/routes/pipelines/_run_support.py b/orchestrator/routes/pipelines/_run_support.py index e03cc7fd4..f4adecb66 100644 --- a/orchestrator/routes/pipelines/_run_support.py +++ b/orchestrator/routes/pipelines/_run_support.py @@ -92,11 +92,33 @@ def _pipeline_cancelled(store, pipeline_id: str) -> bool: """True if ``pipeline_id`` is persisted as CANCELLED (#3633). The operator's cancel is the authoritative "stop driving this run" - signal, and it is the one signal that survives every in-process - mechanism (a stop event a thread never checks, an event loop the cancel - route could not reach). Driver work loops re-read it so a cancelled - pipeline stops admitting slices and spawning agents rather than walking - the rest of its DAG against an operator who believes it is stopped. + signal, and the persisted status outlives the in-process mechanisms a + cancel cannot reach (a stop event a thread never checks, an event loop + the cancel route did not know about). Driver work loops re-read it so a + cancelled pipeline stops admitting slices and spawning agents rather + than walking the rest of its DAG against an operator who believes it is + stopped. + + It is not inviolable, and callers should not treat it as such. Every + operator-blocking gate parks the pipeline at ``AWAITING_HUMAN`` and writes + ``RUNNING`` back once its wait returns, so a loop that only re-reads the + status *after* such a gate sees the gate's own write, not the operator's + cancel. That is why each of them re-checks from inside, before the write: + the refine/plan HITL gate (``_gate_wait_cancelled``, at all five of its + blocking waits), the unresolved-gap gate (``_await_unresolved_gap_gate``), + and the divergence-reconcile pause (``_alerts.py``). Any new + park-and-resume block belongs on that list. + + Re-checking is only half of it. A block that merely *skips* its RUNNING + write leaves the persisted CANCELLED intact and then hands its caller the + same value an ordinary gating returns — so the driver keeps walking the + DAG and the next unguarded status write clobbers the cancel anyway. Each + block therefore has to propagate a stop the driver acts on: the four + ``_run_hitl_gate.py`` sites and the gap gate's caller + (``_run_implement_advance``) return ``"break"``, and the + divergence-reconcile pause returns ``aborted=True``. The terminal-phase + "pipeline complete" branch in ``_run_pipeline`` re-checks as a backstop for + the case a future block forgets. FAILED is deliberately not included: ``container_monitor`` reconciliation can mark a live pipeline FAILED mid-poll, and the diff --git a/orchestrator/tests/test_cancel_stops_driver.py b/orchestrator/tests/test_cancel_stops_driver.py index d1ca8dda7..a2eb36396 100644 --- a/orchestrator/tests/test_cancel_stops_driver.py +++ b/orchestrator/tests/test_cancel_stops_driver.py @@ -7,7 +7,7 @@ its arms and spawned again: ``issue-3596-v2`` was cancelled at 20:48Z and spawned slice-3 agents at 22:55Z, complete with a fresh integration branch. -These tests pin the four layers of the fix: +These tests pin the five layers of the fix: 1. the cancel route stops every live BRC event loop, and does it *before* container cleanup (cleanup that races a live loop is removing pods the @@ -15,13 +15,17 @@ 2. a loop stopped mid-tick refuses the spawn it was about to request; 3. the concurrent-phase poll loop re-reads the persisted status and bails (without escalating, and without rewriting CANCELLED to FAILED); -4. the implement-phase slice loop refuses to admit another slice. +4. the implement-phase slice loop refuses to admit another slice; +5. the refine/plan HITL gate — the one path that *overwrites* the persisted + status the four layers above key on — bails on its own signal (a cancelled + decision) instead of reading the operator's cancel as an approval. """ from __future__ import annotations import threading from datetime import UTC, datetime, timedelta +from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -35,6 +39,8 @@ ) from flask import Flask from models import ( + DecisionStatus, + HITLDecision, Pipeline, PipelinePhase, PipelineStatus, @@ -550,8 +556,12 @@ def test_slice_loop_admits_nothing_after_a_cancel(): certs_volume=None, worktree_repo_path=pipelines_pkg.Path("/tmp/does-not-matter"), # Production (``_run_phase.py``) always threads the owning - # thread's epoch through; pass it so this covers the real - # configuration rather than the ``None`` default. + # thread's epoch through, so pass it rather than leaning on the + # ``None`` default. It is not what this test exercises: layer 4 + # keys on ``_pipeline_cancelled`` alone and bails before + # ``_run_concurrent_phase_with_impasse_retry``, where the epoch + # arm lives. Supersession is covered by + # ``test_phase_bail_reason_still_reports_supersession``. run_epoch=cancelled.run_epoch or cancelled.created_at, ) @@ -614,7 +624,535 @@ def _all_done() -> bool: store=store, certs_volume=None, worktree_repo_path=pipelines_pkg.Path("/tmp/does-not-matter"), + # As above: production-faithful, but not the epoch arm under + # test — ``all_done()`` short-circuits before the retry wrapper. run_epoch=running.run_epoch or running.created_at, ) assert scheduler.iter_ready_calls >= 1, "the guard stopped a RUNNING pipeline" + + +# --------------------------------------------------------------------------- +# Layer 5 — the HITL gate bails instead of reading a cancel as an approval +# --------------------------------------------------------------------------- +# +# The four layers above all key on the *persisted* status. The refine/plan +# gate is the one path that overwrites that status before they read it: it +# parks at AWAITING_HUMAN, blocks in ``wait_for_decision``, and the operator's +# cancel unblocks it by cancelling the decision — leaving no ``resolution``. +# An unset resolution reads as an approval (``"" in _APPROVE_KEYWORDS``), so +# the gate took its "Approved — resume and advance" branch, wrote RUNNING over +# the operator's CANCELLED, and let the driver advance into the next phase and +# mint a fresh cohort: #3633 verbatim, through the one door the +# persisted-status layers cannot watch. + + +def _gate_pipeline(status=PipelineStatus.AWAITING_HUMAN) -> Pipeline: + """A pipeline parked at the plan gate with a pending phase_gate decision. + + The pending decision routes the gate down its ``existing_pending_gate`` + branch, so the test reaches ``wait_for_decision`` without touching draft + reads or the decision queue's create path. + """ + pipeline = Pipeline( + id=PIPELINE_ID, + issue_number=3633, + repo="owner/repo", + branch=f"egg/{PIPELINE_ID}/work", + status=status, + current_phase=PipelinePhase.PLAN, + ) + pipeline.decisions = [ + HITLDecision( + id="gate-1", + question="Approve the plan?", + decision_type="phase_gate", + phase=PipelinePhase.PLAN, + status=DecisionStatus.PENDING, + ) + ] + return pipeline + + +def _gate_decision(status, resolution=None) -> HITLDecision: + return HITLDecision( + id="gate-1", + question="Approve the plan?", + decision_type="phase_gate", + phase=PipelinePhase.PLAN, + status=status, + resolution=resolution, + ) + + +class _StatusCell: + """The pipeline status the fake store persists, mutable mid-run. + + A cancel is not a value the gate is handed — it is a write that lands on + the store while the gate is parked in ``wait_for_decision``. Tests for the + later bail sites flip this from inside a hook to model exactly that. + """ + + def __init__(self, status): + self.status = status + + def cancel(self): + self.status = PipelineStatus.CANCELLED + + +def _run_gate( + resolved_decision, + *, + persisted_status=None, + cell=None, + pipeline=None, + ledger_status=("", False, None, {}), + attestation=None, + bridge=None, + on_wait=None, +): + """Drive ``_run_hitl_gate_converge`` to its phase-gate wait and back. + + ``load_pipeline`` hands out a *fresh* object per call, as the real store + does: the gate mutates what it loads, and a shared MagicMock return value + would let its own ``AWAITING_HUMAN`` write mask the persisted status the + bail re-reads. + + The hooks steer the gate onto whichever branch's bail is under test: + ``ledger_status`` is the ``_collect_decision_ledger_status`` 4-tuple (set + ``missing=True`` for the backstop, ``explicit_none`` for the attestation + gate), ``attestation`` / ``bridge`` stand in for the two helpers that block + on their own waits, and ``on_wait`` fires on every ``wait_for_decision``. + Each receives the ``_StatusCell`` so it can cancel from inside the wait. + """ + cell = cell if cell is not None else _StatusCell(persisted_status) + saved: list[PipelineStatus] = [] + store = MagicMock() + store.load_pipeline.side_effect = lambda _pid: _gate_pipeline(status=cell.status) + store.save_pipeline.side_effect = lambda p, *a, **k: saved.append(p.status) + + waits: list[str] = [] + + def _wait(decision_id): + waits.append(decision_id) + if on_wait is not None: + on_wait(len(waits), cell) + return resolved_decision + + dq = MagicMock() + dq.wait_for_decision.side_effect = _wait + dq.get_decision.return_value = resolved_decision + + def _attestation(**kwargs): + if attestation is not None: + attestation(cell) + return False, "", kwargs["pipeline"] + + def _bridge(*args, **kwargs): + if bridge is not None: + return bridge(cell) + return 0 + + with ( + patch.object( + pipelines_pkg, + "_collect_decision_ledger_status", + return_value=ledger_status, + ), + patch.object(pipelines_pkg, "_persist_decision_ledger_summary", return_value=None), + patch.object( + pipelines_pkg, + "_handle_explicit_none_attestation_gate", + side_effect=_attestation, + ), + patch.object(pipelines_pkg, "get_decision_queue", return_value=dq), + patch.object(pipelines_pkg, "get_pipeline_state_lock"), + patch.object(pipelines_pkg, "report_pipeline_status"), + patch.object(pipelines_pkg, "_emit_pipeline_event"), + patch.object(pipelines_pkg, "_read_phase_draft", return_value="draft body"), + patch.object(pipelines_pkg, "_read_human_phase_draft", return_value=None), + patch.object(pipelines_pkg, "_queue_and_await_contract_decisions", side_effect=_bridge), + patch.object(pipelines_pkg, "_persist_phase_gate_resolution"), + patch.object(pipelines_pkg, "_commit_statefiles_to_worktree"), + ): + _pipeline, action = pipelines_pkg._run_hitl_gate_converge( + pipeline if pipeline is not None else _gate_pipeline(), + current_phase=PipelinePhase.PLAN, + gateway_mode="public", + pipeline_id=PIPELINE_ID, + repo_path=Path("/repo"), + spawner=MagicMock(), + store=store, + worktree_repo_path=Path("/tmp/egg-worktree"), + ) + return action, saved + + +def test_gate_bails_when_the_cancel_route_cancels_its_decision(): + """The headline regression: cancelling a pipeline parked at a HITL gate + must not resurrect it to RUNNING and advance the phase.""" + # What ``cancel_decision`` leaves behind: CANCELLED, no resolution. + action, saved = _run_gate( + _gate_decision(DecisionStatus.CANCELLED), + persisted_status=PipelineStatus.CANCELLED, + ) + + assert action == "break", "the gate must exit the driver loop on a cancel" + assert PipelineStatus.RUNNING not in saved, ( + "the gate rewrote the operator's CANCELLED back to RUNNING" + ) + + +def test_gate_bails_on_a_cancelled_pipeline_even_if_the_decision_resolved(): + """A cancel that lands after the decision-queue sweep — or one racing an + operator who resolved the gate — must still bail: the persisted status is + checked alongside the decision's own.""" + action, saved = _run_gate( + _gate_decision(DecisionStatus.RESOLVED, "approve"), + persisted_status=PipelineStatus.CANCELLED, + ) + + assert action == "break" + assert PipelineStatus.RUNNING not in saved + + +def test_gate_still_advances_a_genuine_approval(): + """Control: a real approval on a live pipeline still advances. Without + this, ``return "break"`` unconditionally would pass the two above.""" + action, saved = _run_gate( + _gate_decision(DecisionStatus.RESOLVED, "approve"), + persisted_status=PipelineStatus.AWAITING_HUMAN, + ) + + assert action is None, "an approved gate must fall through and advance" + assert PipelineStatus.RUNNING in saved + + +def test_gate_still_advances_when_only_the_decision_was_cancelled(): + """A *lone* decision cancel on a live pipeline is not a pipeline cancel. + + ``routes/decisions/_lifecycle.py`` exposes a standalone endpoint that + cancels one decision without touching the pipeline. Keying the bail on the + returned decision's status would read that as a stop, exit the driver, and + strand a live pipeline at AWAITING_HUMAN with no waiter — which is why the + bail consults the persisted pipeline status only. + """ + action, saved = _run_gate( + _gate_decision(DecisionStatus.CANCELLED), + persisted_status=PipelineStatus.AWAITING_HUMAN, + ) + + assert action is None, "a lone decision cancel must not break the driver loop" + assert PipelineStatus.RUNNING in saved + + +def test_gate_bails_when_cancelled_at_the_ledger_backstop(): + """The decision-ledger backstop (#3390) blocks on its own wait before the + phase gate is ever queued. A cancel landing there must not fall through to + the gate — the ``proceed`` branch treats a non-RESOLVED backstop as an + operator override and walks straight into the phase gate below.""" + action, saved = _run_gate( + _gate_decision(DecisionStatus.CANCELLED), + persisted_status=PipelineStatus.AWAITING_HUMAN, + ledger_status=("no ledger", True, None, {}), + on_wait=lambda n, cell: cell.cancel(), + ) + + assert action == "break" + assert PipelineStatus.RUNNING not in saved + + +def test_gate_bails_when_cancelled_at_the_attestation_gate(): + """The explicit-none attestation gate (#3462) "fails open to the phase + gate" on any non-RESOLVED status — safe for a cancelled attestation, not + safe for a cancelled pipeline. Falling through would queue a *fresh* + phase_gate decision minted after the cancel route already swept the queue, + so nothing would ever cancel it and the wait below would never return.""" + action, saved = _run_gate( + _gate_decision(DecisionStatus.RESOLVED, "approve"), + persisted_status=PipelineStatus.AWAITING_HUMAN, + ledger_status=("attested none", False, ("coder", "abc1234", []), {}), + attestation=lambda cell: cell.cancel(), + ) + + assert action == "break" + assert PipelineStatus.RUNNING not in saved + + +def test_gate_bails_when_cancelled_at_the_followup_specifics(): + """A bare "request changes" queues a follow-up asking for specifics and + blocks on it. That wait is swept by a cancel exactly like the gate's own, + and an unset follow-up resolution reads as an approval.""" + action, saved = _run_gate( + _gate_decision(DecisionStatus.RESOLVED, "request changes"), + persisted_status=PipelineStatus.AWAITING_HUMAN, + # Wait 1 is the phase gate (still live); wait 2 is the follow-up. + on_wait=lambda n, cell: cell.cancel() if n == 2 else None, + ) + + assert action == "break" + assert PipelineStatus.RUNNING not in saved + + +def test_gate_bails_when_cancelled_bridging_contract_decisions(): + """The contract-decision bridge (#1889) is the longest human-latency + window in the gate — one blocking wait per contract question. A cancel + there leaves every answer unresolved, so the converge branch is skipped and + control falls through to "Approved — resume and advance".""" + action, saved = _run_gate( + _gate_decision(DecisionStatus.RESOLVED, "approve"), + persisted_status=PipelineStatus.AWAITING_HUMAN, + bridge=lambda cell: (cell.cancel(), 0)[1], + ) + + assert action == "break" + assert PipelineStatus.RUNNING not in saved, ( + "the gate advanced the phase after the operator cancelled mid-bridge" + ) + + +def test_bare_request_changes_on_a_reused_gate_reaches_the_followup(): + """Regression for the reuse path: ``draft_content`` / ``phase_label`` used + to be bound only on the create-a-new-gate arm, so resuming onto an existing + pending gate and answering with a bare "request changes" raised + ``UnboundLocalError`` before the follow-up could be queued.""" + waits: list[int] = [] + action, saved = _run_gate( + _gate_decision(DecisionStatus.RESOLVED, "request changes"), + persisted_status=PipelineStatus.AWAITING_HUMAN, + on_wait=lambda n, cell: waits.append(n), + ) + + assert waits == [1, 2], "the follow-up asking for specifics was never queued" + # The follow-up came back bare too, which the gate reads as an approval. + assert action is None + assert PipelineStatus.RUNNING in saved + + +def test_gate_wait_cancelled_helper(): + """Unit coverage for the seam itself, including the store-hiccup + tolerance it inherits from ``_pipeline_cancelled``.""" + live = _store_returning(_cancellable_pipeline()) + dead = _store_returning(_cancellable_pipeline(status=PipelineStatus.CANCELLED)) + + # A cancelled pipeline bails. Both real cancel paths persist CANCELLED + # before sweeping the decision queue, so a swept wait always sees it. + assert pipelines_pkg._gate_wait_cancelled(dead, PIPELINE_ID) is True + # A live pipeline proceeds. + assert pipelines_pkg._gate_wait_cancelled(live, PIPELINE_ID) is False + # FAILED is not a cancel (#1273): container_monitor can mark a live + # pipeline FAILED mid-gate and the consensus-complete path recovers it. + assert ( + pipelines_pkg._gate_wait_cancelled( + _store_returning(_cancellable_pipeline(status=PipelineStatus.FAILED)), + PIPELINE_ID, + ) + is False + ) + # A store hiccup must never invent a cancel and strand an approved gate. + broken = MagicMock() + broken.load_pipeline.side_effect = RuntimeError("state branch locked") + assert pipelines_pkg._gate_wait_cancelled(broken, PIPELINE_ID) is False + + +# --------------------------------------------------------------------------- +# The pre-spawn guard sits between executor construction and spawn_all +# --------------------------------------------------------------------------- + + +def test_pre_spawn_guard_runs_before_spawn_all(): + """A cancel landing in the prompt-build/session-setup window must stop the + cohort being minted at all. + + That window is tens of seconds wide, and a cancel inside it runs the + route's teardown BEFORE these Jobs exist — nothing would reap them, since + no reconciler acts on CANCELLED and ``cleanup_pipeline`` only re-runs on + an operator DELETE. So the guard has to be the last thing before + ``spawn_all``, not merely present somewhere in the function. + """ + cancelled = _cancellable_pipeline(status=PipelineStatus.CANCELLED) + cancelled.base_branch = "main" + cancelled.current_phase = PipelinePhase.PLAN + store = MagicMock() + store.load_pipeline.return_value = cancelled + + executor = MagicMock() + spawner = MagicMock() + + with ( + patch("concurrent_executor.ConcurrentPhaseExecutor", return_value=executor), + patch.object(pipelines_pkg, "_build_agent_prompt", return_value="prompt"), + ): + exit_code, logs = pipelines_pkg._run_concurrent_phase( + PIPELINE_ID, + cancelled, + "plan", + spawner, + {}, + "public", + ["owner/repo"], + {}, + store, + None, + Path("/tmp/egg-worktree"), + ) + + assert exit_code == 1 + assert "pipeline_cancelled" in logs + executor.spawn_all.assert_not_called() + # The executor owns a live event loop the moment it is constructed, so + # bailing without stopping it would leak the very thing layer 2 stops. + executor.stop_event_loop.assert_called_once() + + +# --------------------------------------------------------------------------- +# Layer 6 — the unresolved-gap gate's bail has to reach the driver +# --------------------------------------------------------------------------- +# +# The gap gate is the one park-and-resume block whose caller, not the block +# itself, decides the pipeline's fate. Skipping the RUNNING write leaves the +# operator's CANCELLED intact for exactly as long as it takes control to +# return to ``_run_pipeline``: IMPLEMENT is terminal +# (``PHASE_TRANSITIONS[IMPLEMENT] == []``), so an unstopped driver walks +# straight into its "pipeline complete" branch, writes COMPLETE over the +# cancel, broadcasts "Pipeline completed successfully", and — now that the +# ``finally`` no longer sees CANCELLED — deletes the worktrees ``restart_phase`` +# resumes from. The gate returning ``gated=True`` cannot be told apart from an +# ordinary gating, so the stop has to be propagated explicitly (#3633 review +# round 2). These tests drive the containing functions, not the gate alone. + + +def _gap_contract(*, resolved: bool): + """A contract carrying one tester→coder gap, resolved or not.""" + from egg_contracts.models import Contract + + return Contract( + pipeline_id=PIPELINE_ID, + slices=[ + { + "id": "slice-1", + "name": "n", + "tasks": [ + { + "id": "task-1-2", + "description": "d", + "gaps": [ + { + "id": "gap-1", + "from_role": "tester", + "to_role": "coder", + "description": "no error-path test", + "resolved": resolved, + } + ], + } + ], + } + ], + ) + + +def _implement_pipeline(status=PipelineStatus.RUNNING) -> Pipeline: + pipeline = Pipeline( + id=PIPELINE_ID, + issue_number=3633, + repo="owner/repo", + branch=f"egg/{PIPELINE_ID}/work", + base_branch="main", + status=status, + current_phase=PipelinePhase.IMPLEMENT, + ) + return pipeline + + +def _run_implement_advance(*, cancel_during_wait: bool, resolution: str | None): + """Drive ``_run_implement_advance`` over the *real* gap gate. + + ``cancel_during_wait`` models the operator's cancel as a store write that + lands while the gate is parked in ``wait_for_decision`` — the shape the + PATCH route produces (persist CANCELLED, then sweep the queue). + """ + cell = _StatusCell(PipelineStatus.RUNNING) + saved: list[PipelineStatus] = [] + + decision = HITLDecision( + id="gap-gate-1", + question="Resolve the gap?", + decision_type="phase_gate", + phase=PipelinePhase.IMPLEMENT, + status=(DecisionStatus.CANCELLED if resolution is None else DecisionStatus.RESOLVED), + resolution=resolution, + ) + + def _wait(_decision_id): + if cancel_during_wait: + cell.cancel() + return decision + + dq = MagicMock() + dq.queue_decision.return_value = decision + dq.wait_for_decision.side_effect = _wait + + def _load(_pipeline_id): + # A fresh object per load, as the real store does — a shared one would + # let the gate's own AWAITING_HUMAN write mask the persisted cancel. + return _implement_pipeline(status=cell.status) + + store = MagicMock() + store.load_pipeline.side_effect = _load + store.save_pipeline.side_effect = lambda p, *a, **k: saved.append(p.status) + + spawner = MagicMock() + + with ( + patch.object(pipelines_pkg, "get_decision_queue", return_value=dq), + patch.object(pipelines_pkg, "get_pipeline_state_lock"), + patch.object(pipelines_pkg, "report_pipeline_status"), + patch.object(pipelines_pkg, "_emit_event", None), + patch.object(pipelines_pkg, "_commit_statefiles_to_worktree", return_value=True), + patch( + "egg_contracts.loader.load_contract", + side_effect=[_gap_contract(resolved=False), _gap_contract(resolved=True)], + ), + ): + pipeline, action = pipelines_pkg._run_implement_advance( + _implement_pipeline(), + current_phase=PipelinePhase.IMPLEMENT, + gateway_mode="public", + pipeline_id=PIPELINE_ID, + repo_path=Path("/repo"), + spawner=spawner, + store=store, + worktree_repo_path=Path("/tmp/egg-worktree"), + ) + return action, saved, spawner + + +def test_implement_advance_stops_the_driver_on_a_cancel_at_the_gap_gate(): + """The gate's bail is only half the fix — its caller has to stop the driver. + + Returning ``gated`` alone is indistinguishable from an ordinary gating, so + ``_run_pipeline`` fell through to the terminal-phase branch and overwrote + the operator's CANCELLED with COMPLETE. + """ + action, saved, spawner = _run_implement_advance(cancel_during_wait=True, resolution=None) + + assert action == "break", ( + "a cancel inside the gap gate must stop the driver, not just skip the " + "gate's own RUNNING write" + ) + assert PipelineStatus.RUNNING not in saved, ( + "the gap gate rewrote the operator's CANCELLED back to RUNNING" + ) + # A cancelled pipeline must not keep mutating the remote work branch. + spawner.gateway.push_worktree_branch.assert_not_called() + + +def test_implement_advance_still_advances_after_a_genuine_gap_resolution(): + """The bail must not swallow the ordinary path: an operator who resolves + the gap and approves gets the post-gate commit+push and a fall-through.""" + action, saved, spawner = _run_implement_advance(cancel_during_wait=False, resolution="approve") + + assert action is None, "a resolved gap gate must let the driver advance" + assert PipelineStatus.RUNNING in saved + spawner.gateway.push_worktree_branch.assert_called_once() diff --git a/orchestrator/tests/test_contract_decision_bridge.py b/orchestrator/tests/test_contract_decision_bridge.py index b4bbca258..2f55d062d 100644 --- a/orchestrator/tests/test_contract_decision_bridge.py +++ b/orchestrator/tests/test_contract_decision_bridge.py @@ -273,6 +273,112 @@ def test_bridge_excludes_cancelled_decision_from_convergence_count( assert cancelled_d["resolved"] is False +def test_bridge_abandons_remaining_waits_when_the_pipeline_is_cancelled( + tmp_path: Path, +) -> None: + """A cancel mid-bridge must stop the remaining waits (#3633). + + The cancel route sweeps only the decisions already *in* the queue. Pass 1 + queues this whole batch up front, so a cancel that lands while pass 2 is + part-way through has already swept them all — but a cancel arriving + *between* pass 1 and pass 2, or a bridge re-entered afterwards, would leave + later entries PENDING with nobody left to cancel them, and the next + ``wait_for_decision`` would block for the process lifetime. + """ + from routes.pipelines import _queue_and_await_contract_decisions + + identifier = "issue-3633" + _make_contract_file( + tmp_path, + identifier, + decisions=[ + { + "id": f"decision-{n}", + "question": f"Question {n}?", + "type": "hitl", + "phase": "refine", + "options": [], + "resolved": False, + "resolution": None, + "resolved_by": None, + "resolved_at": None, + "debounce_until": None, + } + for n in (1, 2, 3) + ], + ) + + class _CountingQueue(_FakeQueue): + def __init__(self, resolutions): + super().__init__(resolutions) + self.waits = 0 + + def wait_for_decision(self, decision_id: str) -> HITLDecision: + self.waits += 1 + return super().wait_for_decision(decision_id) + + dq = _CountingQueue(resolutions=["a", "b", "c"]) + + resolved_count = _queue_and_await_contract_decisions( + dq, + tmp_path, + "pipeline-id", + identifier, + PipelinePhase.REFINE, + # The operator cancels while the second question is being answered. + cancelled=lambda: dq.waits >= 2, + ) + + assert dq.waits == 2, "the bridge kept waiting after the pipeline was cancelled" + # Only the pre-cancel answer counts — the cancel check runs before the + # count, so the round the cancel landed in contributes nothing. + assert resolved_count == 1 + + data = json.loads((tmp_path / f".egg-state/contracts/{identifier}.json").read_text()) + resolved_ids = [d["id"] for d in data["decisions"] if d["resolved"]] + assert resolved_ids == ["decision-1"] + + +def test_bridge_without_a_cancelled_predicate_answers_every_decision( + tmp_path: Path, +) -> None: + """Control: ``cancelled`` is optional and defaults to never firing, so the + existing callers that omit it keep answering the whole batch.""" + from routes.pipelines import _queue_and_await_contract_decisions + + identifier = "issue-3633-control" + _make_contract_file( + tmp_path, + identifier, + decisions=[ + { + "id": f"decision-{n}", + "question": f"Question {n}?", + "type": "hitl", + "phase": "refine", + "options": [], + "resolved": False, + "resolution": None, + "resolved_by": None, + "resolved_at": None, + "debounce_until": None, + } + for n in (1, 2, 3) + ], + ) + dq = _FakeQueue(resolutions=["a", "b", "c"]) + + resolved_count = _queue_and_await_contract_decisions( + dq, + tmp_path, + "pipeline-id", + identifier, + PipelinePhase.REFINE, + ) + + assert resolved_count == 3 + + def test_bridge_promotes_unsubmitted_feedback(tmp_path: Path) -> None: from routes.pipelines import _queue_and_await_contract_decisions diff --git a/orchestrator/tests/test_hard_reset_recovery.py b/orchestrator/tests/test_hard_reset_recovery.py index 47c4eea66..b14753c8e 100644 --- a/orchestrator/tests/test_hard_reset_recovery.py +++ b/orchestrator/tests/test_hard_reset_recovery.py @@ -332,6 +332,37 @@ def _hook() -> None: mock_report.assert_called_once() assert mock_report.call_args.kwargs["event_type"] == "pipeline.failed" + def test_cancelled_pipeline_keeps_cancelled_and_skips_the_broadcast(self): + """An operator cancel that unblocked the reconcile pause routes here + too (#3633). Pinning FAILED would overwrite the persisted CANCELLED + every driver work loop keys on, and report a failure the operator never + caused. The teardown hook still runs — it is wanted on either exit.""" + pipeline = MagicMock() + pipeline.status = PipelineStatus.CANCELLED + store = MagicMock() + store.load_pipeline.return_value = pipeline + order: list[str] = [] + + with ( + patch("routes.pipelines.get_pipeline_state_lock"), + patch("routes.pipelines.report_pipeline_status") as mock_report, + patch("routes.pipelines._emit_pipeline_event") as mock_emit, + ): + mock_emit.side_effect = lambda *a, **k: order.append("event") + _fail_pipeline_after_divergence_abort( + "pipe-1", + store, + phase=PipelinePhase.PLAN, + backup_ref="refs/egg-backup/sync-recovery/pipe-1/9", + local_only_commit_shas=("abc1234 foo",), + pre_event_hook=lambda: order.append("hook"), + ) + + assert pipeline.status == PipelineStatus.CANCELLED + store.save_pipeline.assert_not_called() + assert order == ["hook"], "pipeline.failed was broadcast for a cancel" + mock_report.assert_not_called() + class TestSyncWorktreeReconcilingDivergence: """The in-loop pause→reconcile→resume / abort loop (#2979).""" @@ -428,6 +459,63 @@ def test_abort_returns_aborted(self): assert result.diverged_unreconciled is True mock_sync.assert_called_once() + def test_cancel_during_the_pause_stops_without_restoring_running(self): + """The pause is a park-and-resume block: it writes AWAITING_HUMAN, + blocks, and writes RUNNING back once the wait returns. A cancel sweeps + that decision with no resolution, so without an inside check the RUNNING + write would land on top of the operator's CANCELLED and re-admit the + run (#3633). ``aborted=True`` is how the two callers spell "stop".""" + # ``load_pipeline`` hands out a fresh object per call, as the real + # store does — the pause mutates what it loads, so a shared return + # value would let its own AWAITING_HUMAN write mask the cancel. + persisted = {"status": PipelineStatus.RUNNING} + saved: list[PipelineStatus] = [] + store = MagicMock() + + def _load(_pipeline_id): + loaded = MagicMock() + loaded.status = persisted["status"] + return loaded + + store.load_pipeline.side_effect = _load + store.save_pipeline.side_effect = lambda p, *a, **k: saved.append(p.status) + dq = MagicMock() + + def _wait(_decision_id): + # The operator cancels while the pause is blocked here. + persisted["status"] = PipelineStatus.CANCELLED + + dq.wait_for_decision.side_effect = _wait + # What the cancel route leaves behind: swept, no resolution. + dq.get_decision.return_value = MagicMock(resolution=None) + lock_p, persist_p, report_p, emit_p = self._patch_ctx() + with ( + patch( + "routes.pipelines._sync_worktree_with_remote", + return_value=_diverged_outcome(), + ) as mock_sync, + patch("routes.pipelines.get_decision_queue", return_value=dq), + lock_p, + persist_p, + report_p, + emit_p, + ): + _result, aborted = _sync_worktree_reconciling_divergence( + MagicMock(), + "pipe-1", + store, + Path("/repo"), + worktree_repo_path=Path("/wt"), + phase=PipelinePhase.PLAN, + ) + + assert aborted is True + assert PipelineStatus.RUNNING not in saved, ( + "the reconcile pause rewrote the operator's CANCELLED to RUNNING" + ) + # Bailed before the resolution read, so no second sync attempt. + mock_sync.assert_called_once() + def test_reconcile_budget_exhausted_aborts(self): """If every resume re-diverges, the bounded budget eventually aborts rather than pausing forever.""" diff --git a/orchestrator/tests/test_pipeline_failure_path.py b/orchestrator/tests/test_pipeline_failure_path.py index f5abfe85c..d3c8fdad6 100644 --- a/orchestrator/tests/test_pipeline_failure_path.py +++ b/orchestrator/tests/test_pipeline_failure_path.py @@ -484,6 +484,162 @@ def test_worktree_cleanup_skipped_on_failure( # delete_worktrees should NOT be called since pipeline is FAILED mock_gateway.delete_worktrees.assert_not_called() + @patch(_COMMON_PATCHES[7]) + @patch(_COMMON_PATCHES[6]) + @patch(_COMMON_PATCHES[5]) + @patch(_COMMON_PATCHES[4]) + @patch(_COMMON_PATCHES[3]) + @patch(_COMMON_PATCHES[2]) + @patch(_COMMON_PATCHES[1]) + @patch(_COMMON_PATCHES[0]) + def test_worktree_cleanup_skipped_on_cancellation( + self, + mock_emit, + mock_get_spawner, + mock_get_store, + mock_spawn_wait, + mock_state_lock, + mock_build_prompt, + mock_read_draft, + mock_report, + ): + """CANCELLED joins FAILED on the ``skip_cleanup`` arm (#3633 review). + + ``restart_phase`` allowlists CANCELLED precisely so a ``cancel_task`` + run can be resumed without a full resubmission (#1725), and the PATCH + cancel route already passes ``preserve_worktrees=(status == + "cancelled")``. Before this, the driver's own ``finally`` contradicted + both by deleting the worktrees the operator was told they could resume + from — and the #3633 layers make that teardown land seconds after the + cancel rather than at the next consensus timeout, so the two policies + have to agree. The whole justification lives in a comment, which makes + a "tidy-up" back to FAILED-only a silent regression; pin it. + """ + from routes.pipelines import _run_pipeline + + pipeline = _make_running_pipeline() + pipeline.status = PipelineStatus.CANCELLED + mock_store, mock_gateway = _setup_mocks( + mock_report, + mock_read_draft, + mock_build_prompt, + mock_state_lock, + mock_spawn_wait, + mock_get_store, + mock_get_spawner, + mock_emit, + pipeline, + ) + mock_spawner = mock_get_spawner.return_value + mock_spawner.cleanup_pipeline.return_value = 0 + + with ( + patch.dict(os.environ, {"EGG_HOST_REPO_MAP": '{"repo": "/host/repo"}'}, clear=False), + patch("pathlib.Path.exists", return_value=True), + # The ``finally`` reaches the spawner through ``_get_spawner`` + # (runtime-dependent), not the ``get_container_spawner`` seam the + # rest of the harness patches. + patch("routes.pipelines._get_spawner", return_value=mock_spawner), + ): + _run_pipeline("issue-42", Path("/repo")) + + assert pipeline.status == PipelineStatus.CANCELLED + mock_gateway.delete_worktrees.assert_not_called() + + # The safety-net container sweep must still run — only the worktrees + # are preserved, and it has to be told so explicitly. + assert mock_spawner.cleanup_pipeline.call_args is not None, ( + "safety-net container cleanup must still run for a cancelled run" + ) + assert mock_spawner.cleanup_pipeline.call_args.kwargs["preserve_worktrees"] is True + + @patch(_COMMON_PATCHES[7]) + @patch(_COMMON_PATCHES[6]) + @patch(_COMMON_PATCHES[5]) + @patch(_COMMON_PATCHES[4]) + @patch(_COMMON_PATCHES[3]) + @patch(_COMMON_PATCHES[2]) + @patch(_COMMON_PATCHES[1]) + @patch(_COMMON_PATCHES[0]) + def test_terminal_phase_does_not_complete_a_cancelled_pipeline( + self, + mock_emit, + mock_get_spawner, + mock_get_store, + mock_spawn_wait, + mock_state_lock, + mock_build_prompt, + mock_read_draft, + mock_report, + ): + """The terminal-phase branch must not write COMPLETE over a cancel. + + It is the last unguarded status write in the driver loop, and every + park-and-resume block in a terminal phase lands on it: the block bails + without restoring RUNNING (leaving CANCELLED persisted), the driver + reads "no next phase" as success, and the operator's cancel becomes + COMPLETE plus a "Pipeline completed successfully" broadcast — after + which the ``finally`` no longer sees CANCELLED, so ``skip_cleanup`` + stays False and the worktrees ``restart_phase`` resumes from are + deleted (#3633 review round 2). + + The bail is modelled at ``_run_implement_advance`` — the gap gate's + caller, and today's only such block — but the guard is deliberately on + the branch rather than the block, so a future one cannot reintroduce + this by forgetting to propagate. + """ + from routes.pipelines import _run_pipeline + + pipeline = _make_running_pipeline() + pipeline.current_phase = PipelinePhase.IMPLEMENT + execution = pipeline.get_phase_execution(PipelinePhase.IMPLEMENT) + execution.status = PipelineStatus.RUNNING + execution.started_at = datetime.now(UTC) + + mock_store, mock_gateway = _setup_mocks( + mock_report, + mock_read_draft, + mock_build_prompt, + mock_state_lock, + mock_spawn_wait, + mock_get_store, + mock_get_spawner, + mock_emit, + pipeline, + ) + mock_spawner = mock_get_spawner.return_value + mock_spawner.cleanup_pipeline.return_value = 0 + + def _bail_without_propagating(pl, **_kwargs): + """A park-and-resume block that skips its RUNNING write and stops + there — what the gap gate did before its caller learned to stop + the driver.""" + pipeline.status = PipelineStatus.CANCELLED + return pl, None + + with ( + patch.dict(os.environ, {"EGG_HOST_REPO_MAP": '{"repo": "/host/repo"}'}, clear=False), + patch("pathlib.Path.exists", return_value=True), + patch("routes.pipelines._run_concurrent_phase", return_value=(0, "")), + patch( + "routes.pipelines._run_implement_advance", + side_effect=_bail_without_propagating, + ), + patch("routes.pipelines._get_spawner", return_value=mock_spawner), + ): + _run_pipeline("issue-42", Path("/repo")) + + assert pipeline.status == PipelineStatus.CANCELLED, ( + "the terminal-phase branch overwrote the operator's CANCELLED" + ) + completed = [ + c + for c in mock_emit.call_args_list + if len(c.args) >= 2 and c.args[1] == "pipeline.completed" + ] + assert completed == [], "a cancelled pipeline must not broadcast completion" + mock_gateway.delete_worktrees.assert_not_called() + class TestWorktreeCreationFailure: """Verify pipeline fails when worktree creation returns empty worktrees.""" diff --git a/orchestrator/tests/test_unresolved_gap_gate.py b/orchestrator/tests/test_unresolved_gap_gate.py index e0ae6852a..c87a1e779 100644 --- a/orchestrator/tests/test_unresolved_gap_gate.py +++ b/orchestrator/tests/test_unresolved_gap_gate.py @@ -267,3 +267,60 @@ def test_unresolved_decision_does_not_spin() -> None: gated, _, _ = _run_gate(dq, load_side_effect=[_contract(resolved=False)]) assert gated is True assert len(dq.queued) == 1 + + +def test_pipeline_cancelled_during_the_gate_keeps_cancelled() -> None: + """This gate is a park-and-resume block: AWAITING_HUMAN, block, then + RUNNING back once the wait returns. When what unblocked the wait was the + operator cancelling the pipeline, that RUNNING write lands on top of the + persisted CANCELLED every driver work loop keys on, re-admitting the run + the operator just stopped (#3633). Bail before the status write.""" + from routes.pipelines import _await_unresolved_gap_gate + + persisted = {"status": PipelineStatus.RUNNING} + saved: list[PipelineStatus] = [] + + class _CancellingQueue(_FakeQueue): + def wait_for_decision(self, decision_id: str): + # The operator cancels while the gate is blocked here; the cancel + # route sweeps this decision, leaving no resolution. + persisted["status"] = PipelineStatus.CANCELLED + return super().wait_for_decision(decision_id) + + dq = _CancellingQueue(resolutions=[None]) + + # A fresh object per load, as the real store does — a shared one would let + # the gate's own AWAITING_HUMAN write mask the cancel the bail re-reads. + def _load(_pipeline_id): + pipeline = Pipeline( + id="issue-42", issue_number=42, repo="owner/repo", branch="egg/issue-42" + ) + pipeline.current_phase = PipelinePhase.IMPLEMENT + pipeline.status = persisted["status"] + return pipeline + + store = MagicMock() + store.load_pipeline.side_effect = _load + store.save_pipeline.side_effect = lambda p, *a, **k: saved.append(p.status) + + with ( + patch("routes.pipelines.get_decision_queue", return_value=dq), + patch("routes.pipelines.get_pipeline_state_lock", return_value=nullcontext()), + patch("routes.pipelines.report_pipeline_status"), + patch("routes.pipelines._emit_event", None), + patch("egg_contracts.loader.load_contract", side_effect=[_contract(resolved=False)]), + ): + gated = _await_unresolved_gap_gate( + store, + "issue-42", + Path("/repo"), + Path("/worktree"), + 42, + PipelinePhase.IMPLEMENT, + True, + ) + + assert gated is True + assert PipelineStatus.RUNNING not in saved, ( + "the gap gate rewrote the operator's CANCELLED back to RUNNING" + )