diff --git a/orchestrator/CLAUDE.md b/orchestrator/CLAUDE.md index 1a3145f69..f374edcd0 100644 --- a/orchestrator/CLAUDE.md +++ b/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.")` 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. diff --git a/orchestrator/routes/pipelines/__init__.py b/orchestrator/routes/pipelines/__init__.py index 098e68122..aec437505 100644 --- a/orchestrator/routes/pipelines/__init__.py +++ b/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, @@ -1433,6 +1435,7 @@ 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_cancelled, _spawn_and_wait, diff --git a/orchestrator/routes/pipelines/_alerts.py b/orchestrator/routes/pipelines/_alerts.py index 4e66a89dd..79df05bd5 100644 --- a/orchestrator/routes/pipelines/_alerts.py +++ b/orchestrator/routes/pipelines/_alerts.py @@ -237,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. @@ -314,6 +340,18 @@ def _sync_worktree_reconciling_divergence( # 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 " diff --git a/orchestrator/routes/pipelines/_contract_bridge.py b/orchestrator/routes/pipelines/_contract_bridge.py new file mode 100644 index 000000000..8471dc184 --- /dev/null +++ b/orchestrator/routes/pipelines/_contract_bridge.py @@ -0,0 +1,298 @@ +"""Contract-decision bridge for routes/pipelines (#1889). + +Split out of ``_ledger.py`` to keep it under the 1,500-line file-size cap +(#3312). Barrel-resident and test-patched globals are reached via ``_pkg`` so +``patch("routes.pipelines.")`` keeps intercepting. +""" + +from __future__ import annotations + +import routes.pipelines as _pkg # noqa: E402,F401 + + +def _queue_and_await_contract_decisions( + dq: _pkg.Any, + worktree_repo_path: _pkg.Path, + 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 at two points, and + the two close different interleavings: + + - **Before pass 1**, because the cancel route's pending-decision sweep is + a one-time snapshot. A cancel that lands entirely *before* this batch is + queued has nothing to sweep, so every decision below is minted with + nobody to cancel it and the very first ``wait_for_decision`` — an + unbounded poll with no timeout — blocks for the process lifetime, + leaking the driver thread and skipping ``_run_pipeline``'s ``finally`` + (#3633 review round 3). The post-wait check cannot help here: it is + never reached. + - **After each blocking wait**, which covers the cancel that lands once + some of the batch is already pending. Those entries *are* swept, so + their waits return immediately with no resolution; the check stops the + remaining waits rather than letting the loop walk a cancelled run's + questions. + + A cancel landing between the pre-pass-1 check and a later + ``queue_decision`` in pass 1 still mints unsweepable entries — but the + *first* wait is on an entry the sweep did reach (or, if the cancel beat + the whole batch, the pre-check fired), so the post-wait check returns + before the unsweepable ones are ever waited on. + + 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 + operator cancelled them) are **not** counted: the contract ``cq-N`` stays + open, and counting it would re-run the phase, re-surface the still-open + question (carry-forward only adopts *resolved* questions), and loop with no + termination now that the force-advance backstop is gone. A non-zero count + means the operator just answered something, so the caller re-runs the + phase to fold the resolutions into the documents; a zero count means the + round resolved nothing new and the caller may advance. + + + Agents register architectural questions via ``egg-contract add-decision`` + and ``add-feedback``. Those writes only touch ``.egg-state/contracts/ + {identifier}.json`` — the orchestrator's decision queue never sees them, + so approving the phase_gate silently drops the questions and the next + phase's agents have to guess (issue #1889). + + This helper bridges contract-scoped questions for the current phase into + the orchestrator queue after phase_gate approval, so HTTP/MCP callers + (e.g. the ``/sdlc`` skill's Phase 4 handler) surface them as individual + ``choice`` / ``feedback`` decisions. Resolutions are written back to + the contract so implement-phase agents see the human's answers. + + All pending decisions (plus the feedback entry, if any) are queued up + front before any ``wait_for_decision`` call, so ``get_status`` surfaces + them as a single batch. Callers can then prompt for up to 4 at a time + and submit answers in parallel, collapsing what was previously N prompts + and N polling cycles into ~⌈N/4⌉ prompts and one cycle (issue #1956). + + Once the batch is queued, a single ``decision.created`` event is + published to the EventBus so event-driven watchers (the ``wait-status`` + monitor long-polling ``/status/wait``) wake immediately. + ``DecisionQueue.queue_decision`` itself emits no event, so without this + the bridged decisions are created silently and the operator only + discovers them via a manual ``get_status`` (issue #2770). + """ + try: + from egg_contracts.loader import load_contract, save_contract + except ImportError: + _pkg.logger.warning( + "egg_contracts not available, skipping contract decision bridge", + pipeline_id=pipeline_id, + ) + return 0 + + try: + contract = load_contract(pipeline_identifier, worktree_repo_path) + except Exception as e: + _pkg.logger.debug( + "Contract not loadable, skipping contract decision bridge", + pipeline_id=pipeline_id, + error=str(e), + ) + return 0 + + phase_value = phase.value + pending_decisions = [ + d + for d in contract.decisions + if not d.resolved + and getattr(d.type, "value", d.type) == "hitl" + and (d.phase is None or getattr(d.phase, "value", d.phase) == phase_value) + ] + fb = contract.feedback + pending_feedback = None + if fb is not None and not fb.submitted: + fb_phase_val = getattr(fb.phase, "value", fb.phase) if fb.phase is not None else None + if fb_phase_val is None or fb_phase_val == phase_value: + pending_feedback = fb + + if not pending_decisions and pending_feedback is None: + return 0 + + _pkg.logger.info( + "Bridging contract decisions/feedback into orchestrator queue", + pipeline_id=pipeline_id, + phase=phase_value, + decision_count=len(pending_decisions), + has_feedback=pending_feedback is not None, + ) + + def _save_contract_update(mutator: _pkg.Callable[[_pkg.Any], bool]) -> None: + try: + latest = load_contract(pipeline_identifier, worktree_repo_path) + except Exception as e: + _pkg.logger.warning( + "Could not reload contract to persist bridged resolution", + pipeline_id=pipeline_id, + error=str(e), + ) + return + if not mutator(latest): + return + try: + save_contract(latest, worktree_repo_path) + except Exception as e: + _pkg.logger.warning( + "Failed to save contract after bridged resolution", + pipeline_id=pipeline_id, + error=str(e), + ) + + # A cancel that landed before pass 1 left nothing in the queue to sweep, + # so every decision minted below would be unsweepable and the first wait + # in pass 2 would never return. Check once here, before anything is + # queued — the post-wait checks in pass 2 cannot reach that interleaving + # (#3633 review round 3). + if cancelled is not None and cancelled(): + _pkg.logger.info( + "Contract decision bridge skipped: pipeline cancelled before queueing (#3633)", + pipeline_id=pipeline_id, + phase=phase_value, + ) + return 0 + + # Pass 1: queue every pending decision + feedback up front. + queued_decisions: list[tuple[str, _pkg.Any]] = [] + for contract_decision in pending_decisions: + options_labels = [opt.label for opt in contract_decision.options] + queued = dq.queue_decision( + question=contract_decision.question, + context=( + f"Open contract question {contract_decision.id}, " + f"registered by an agent during the {phase_value} phase." + ), + options=options_labels, + decision_type="choice", + phase=phase, + ) + queued_decisions.append((contract_decision.id, queued)) + + queued_feedback: _pkg.HITLDecision | None = None + if pending_feedback is not None: + questions_payload = [ + {"id": q.id, "question": q.question, "answer": ""} for q in pending_feedback.questions + ] + queued_feedback = dq.queue_decision( + question=f"Open feedback request {pending_feedback.id}", + context=( + f"Open contract feedback {pending_feedback.id}, " + f"registered by an agent during the {phase_value} phase." + ), + options=[], + decision_type="feedback", + questions=questions_payload, + phase=phase, + ) + + # Surface the freshly-queued batch to event-driven watchers before + # blocking on resolution. ``DecisionQueue.queue_decision`` emits no + # EventBus event, so without this the bridged decisions are created + # silently — the operator's ``wait-status`` monitor never wakes and + # only finds them via a manual ``get_status`` (#2770). The phase_gate + # decision emits ``decision.created`` the same way. + if _pkg._emit_event is not None: + _pkg._emit_event( + _pkg.EventType.DECISION_CREATED, + pipeline_id, + data={"phase": phase_value}, + ) + + # Pass 2: wait for each to resolve and persist back to the contract. + # Count only decisions whose queue resolution was RESOLVED — a + # CANCELLED / non-resolved outcome leaves the contract ``cq-N`` open and + # must NOT count toward the convergence signal, or the caller would re-run + # the phase, re-surface the still-open question (carry-forward only adopts + # *resolved* questions), and loop without the operator ever being able to + # break out (#3392 review). + 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 + resolution_str = (resolved.resolution or "").strip() + + def _apply(latest: _pkg.Any, _cd_id: str = contract_id, _res: str = resolution_str) -> bool: + for d in latest.decisions: + if d.id == _cd_id: + d.resolved = True + d.resolution = _res + d.resolved_by = "human" + d.resolved_at = _pkg.datetime.now(_pkg.UTC) + return True + return False + + _save_contract_update(_apply) + + 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] = {} + try: + payload = _pkg.json.loads(resolved.resolution or "") + if isinstance(payload, dict): + raw_answers = payload.get("answers") + if isinstance(raw_answers, dict): + answers = {str(k): str(v) for k, v in raw_answers.items()} + except _pkg.json.JSONDecodeError, TypeError: + pass + + fb_id = pending_feedback.id + + def _apply_fb( + latest: _pkg.Any, _fb_id: str = fb_id, _answers: dict[str, str] = answers + ) -> bool: + if latest.feedback is None or latest.feedback.id != _fb_id: + return False + for q in latest.feedback.questions: + if q.id in _answers: + q.answer = _answers[q.id] + # Always mark submitted after resolution — even if + # individual answers didn't parse, the human responded + # and shouldn't be asked again. + latest.feedback.submitted = True + latest.feedback.submitted_by = "human" + latest.feedback.submitted_at = _pkg.datetime.now(_pkg.UTC) + return True + + _save_contract_update(_apply_fb) + + # Convergence signal (#3392): the number of decisions + feedback this + # round the operator actually *resolved* (not merely surfaced). Non-zero ⇒ + # the operator answered something ⇒ caller re-runs the phase to fold the + # resolutions in. A surfaced-but-cancelled decision is deliberately + # excluded: counting it would re-run the phase, re-surface the still-open + # question, and loop indefinitely now that the force-advance backstop is + # gone. + return resolved_count + (1 if feedback_resolved else 0) diff --git a/orchestrator/routes/pipelines/_ledger.py b/orchestrator/routes/pipelines/_ledger.py index 5efecd1d2..b36c5051d 100644 --- a/orchestrator/routes/pipelines/_ledger.py +++ b/orchestrator/routes/pipelines/_ledger.py @@ -242,8 +242,10 @@ def _handle_explicit_none_attestation_gate( - ``rerun_requested`` — True when the operator rejected the attestation and the phase has already been re-run here; the caller must ``continue`` its poll loop. False when the attestation was confirmed (or fail-open on - a cancelled/non-RESOLVED terminal state); the caller proceeds to the - phase gate. + a cancelled/non-RESOLVED terminal state, or on an operator cancel + detected *before* the decision was queued or parked); the caller + proceeds to the phase gate — where its own ``_gate_wait_cancelled`` + check turns the cancelled case into a driver break. - ``ledger_note`` — the note to thread into the phase_gate question, annotated with the confirmation outcome. - ``pipeline`` — the (possibly reloaded) pipeline the caller must rebind, @@ -288,6 +290,20 @@ def _handle_explicit_none_attestation_gate( attest_decision = pending_attest newly_created = False else: + # Never mint a decision for a cancelled run: the cancel route's + # pending-decision sweep is a one-time snapshot, so a decision created + # after it has nobody to cancel it and ``wait_for_decision`` below — + # an unbounded poll — would block for the process lifetime (#3633 + # review round 3). Fail open to the caller, which re-checks the cancel + # immediately on return and breaks the driver loop. + if _pkg._pipeline_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled before the explicit-none attestation was " + "queued — skipping the gate (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return False, ledger_note, pipeline attest_decision = dq.queue_decision( question=attest_question, context=ledger_note, @@ -299,12 +315,21 @@ def _handle_explicit_none_attestation_gate( phase=current_phase, ) newly_created = True - with _pkg.get_pipeline_state_lock(pipeline_id): - pipeline = store.load_pipeline(pipeline_id) - pipeline.status = _pkg.PipelineStatus.AWAITING_HUMAN - phase_execution = pipeline.get_phase_execution(current_phase) - phase_execution.status = _pkg.PipelineStatus.AWAITING_HUMAN - store.save_pipeline(pipeline) + # Park under the state lock, re-reading the status from inside it — the + # cancel route persists CANCELLED under the same lock, so an unconditional + # write here would lose it and the caller's post-return check would read + # back this gate's own AWAITING_HUMAN (#3633 review round 3). + pipeline, _park_cancelled = _pkg._park_at_gate_unless_cancelled( + store, pipeline_id, current_phase + ) + if _park_cancelled: + _pkg.logger.info( + "Pipeline cancelled before parking at the explicit-none " + "attestation — skipping the gate (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return False, ledger_note, pipeline # Only announce a freshly-created decision. Reusing a pending decision # across polls must not re-emit ``decision.created`` — a duplicate event # for a decision the operator is already looking at (#3462 review). @@ -549,264 +574,6 @@ def _collect_decision_ledger_status( ) -def _queue_and_await_contract_decisions( - dq: _pkg.Any, - worktree_repo_path: _pkg.Path, - 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 - operator cancelled them) are **not** counted: the contract ``cq-N`` stays - open, and counting it would re-run the phase, re-surface the still-open - question (carry-forward only adopts *resolved* questions), and loop with no - termination now that the force-advance backstop is gone. A non-zero count - means the operator just answered something, so the caller re-runs the - phase to fold the resolutions into the documents; a zero count means the - round resolved nothing new and the caller may advance. - - - Agents register architectural questions via ``egg-contract add-decision`` - and ``add-feedback``. Those writes only touch ``.egg-state/contracts/ - {identifier}.json`` — the orchestrator's decision queue never sees them, - so approving the phase_gate silently drops the questions and the next - phase's agents have to guess (issue #1889). - - This helper bridges contract-scoped questions for the current phase into - the orchestrator queue after phase_gate approval, so HTTP/MCP callers - (e.g. the ``/sdlc`` skill's Phase 4 handler) surface them as individual - ``choice`` / ``feedback`` decisions. Resolutions are written back to - the contract so implement-phase agents see the human's answers. - - All pending decisions (plus the feedback entry, if any) are queued up - front before any ``wait_for_decision`` call, so ``get_status`` surfaces - them as a single batch. Callers can then prompt for up to 4 at a time - and submit answers in parallel, collapsing what was previously N prompts - and N polling cycles into ~⌈N/4⌉ prompts and one cycle (issue #1956). - - Once the batch is queued, a single ``decision.created`` event is - published to the EventBus so event-driven watchers (the ``wait-status`` - monitor long-polling ``/status/wait``) wake immediately. - ``DecisionQueue.queue_decision`` itself emits no event, so without this - the bridged decisions are created silently and the operator only - discovers them via a manual ``get_status`` (issue #2770). - """ - try: - from egg_contracts.loader import load_contract, save_contract - except ImportError: - _pkg.logger.warning( - "egg_contracts not available, skipping contract decision bridge", - pipeline_id=pipeline_id, - ) - return 0 - - try: - contract = load_contract(pipeline_identifier, worktree_repo_path) - except Exception as e: - _pkg.logger.debug( - "Contract not loadable, skipping contract decision bridge", - pipeline_id=pipeline_id, - error=str(e), - ) - return 0 - - phase_value = phase.value - pending_decisions = [ - d - for d in contract.decisions - if not d.resolved - and getattr(d.type, "value", d.type) == "hitl" - and (d.phase is None or getattr(d.phase, "value", d.phase) == phase_value) - ] - fb = contract.feedback - pending_feedback = None - if fb is not None and not fb.submitted: - fb_phase_val = getattr(fb.phase, "value", fb.phase) if fb.phase is not None else None - if fb_phase_val is None or fb_phase_val == phase_value: - pending_feedback = fb - - if not pending_decisions and pending_feedback is None: - return 0 - - _pkg.logger.info( - "Bridging contract decisions/feedback into orchestrator queue", - pipeline_id=pipeline_id, - phase=phase_value, - decision_count=len(pending_decisions), - has_feedback=pending_feedback is not None, - ) - - def _save_contract_update(mutator: _pkg.Callable[[_pkg.Any], bool]) -> None: - try: - latest = load_contract(pipeline_identifier, worktree_repo_path) - except Exception as e: - _pkg.logger.warning( - "Could not reload contract to persist bridged resolution", - pipeline_id=pipeline_id, - error=str(e), - ) - return - if not mutator(latest): - return - try: - save_contract(latest, worktree_repo_path) - except Exception as e: - _pkg.logger.warning( - "Failed to save contract after bridged resolution", - pipeline_id=pipeline_id, - error=str(e), - ) - - # Pass 1: queue every pending decision + feedback up front. - queued_decisions: list[tuple[str, _pkg.Any]] = [] - for contract_decision in pending_decisions: - options_labels = [opt.label for opt in contract_decision.options] - queued = dq.queue_decision( - question=contract_decision.question, - context=( - f"Open contract question {contract_decision.id}, " - f"registered by an agent during the {phase_value} phase." - ), - options=options_labels, - decision_type="choice", - phase=phase, - ) - queued_decisions.append((contract_decision.id, queued)) - - queued_feedback: _pkg.HITLDecision | None = None - if pending_feedback is not None: - questions_payload = [ - {"id": q.id, "question": q.question, "answer": ""} for q in pending_feedback.questions - ] - queued_feedback = dq.queue_decision( - question=f"Open feedback request {pending_feedback.id}", - context=( - f"Open contract feedback {pending_feedback.id}, " - f"registered by an agent during the {phase_value} phase." - ), - options=[], - decision_type="feedback", - questions=questions_payload, - phase=phase, - ) - - # Surface the freshly-queued batch to event-driven watchers before - # blocking on resolution. ``DecisionQueue.queue_decision`` emits no - # EventBus event, so without this the bridged decisions are created - # silently — the operator's ``wait-status`` monitor never wakes and - # only finds them via a manual ``get_status`` (#2770). The phase_gate - # decision emits ``decision.created`` the same way. - if _pkg._emit_event is not None: - _pkg._emit_event( - _pkg.EventType.DECISION_CREATED, - pipeline_id, - data={"phase": phase_value}, - ) - - # Pass 2: wait for each to resolve and persist back to the contract. - # Count only decisions whose queue resolution was RESOLVED — a - # CANCELLED / non-resolved outcome leaves the contract ``cq-N`` open and - # must NOT count toward the convergence signal, or the caller would re-run - # the phase, re-surface the still-open question (carry-forward only adopts - # *resolved* questions), and loop without the operator ever being able to - # break out (#3392 review). - 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 - resolution_str = (resolved.resolution or "").strip() - - def _apply(latest: _pkg.Any, _cd_id: str = contract_id, _res: str = resolution_str) -> bool: - for d in latest.decisions: - if d.id == _cd_id: - d.resolved = True - d.resolution = _res - d.resolved_by = "human" - d.resolved_at = _pkg.datetime.now(_pkg.UTC) - return True - return False - - _save_contract_update(_apply) - - 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] = {} - try: - payload = _pkg.json.loads(resolved.resolution or "") - if isinstance(payload, dict): - raw_answers = payload.get("answers") - if isinstance(raw_answers, dict): - answers = {str(k): str(v) for k, v in raw_answers.items()} - except _pkg.json.JSONDecodeError, TypeError: - pass - - fb_id = pending_feedback.id - - def _apply_fb( - latest: _pkg.Any, _fb_id: str = fb_id, _answers: dict[str, str] = answers - ) -> bool: - if latest.feedback is None or latest.feedback.id != _fb_id: - return False - for q in latest.feedback.questions: - if q.id in _answers: - q.answer = _answers[q.id] - # Always mark submitted after resolution — even if - # individual answers didn't parse, the human responded - # and shouldn't be asked again. - latest.feedback.submitted = True - latest.feedback.submitted_by = "human" - latest.feedback.submitted_at = _pkg.datetime.now(_pkg.UTC) - return True - - _save_contract_update(_apply_fb) - - # Convergence signal (#3392): the number of decisions + feedback this - # round the operator actually *resolved* (not merely surfaced). Non-zero ⇒ - # the operator answered something ⇒ caller re-runs the phase to fold the - # resolutions in. A surfaced-but-cancelled decision is deliberately - # excluded: counting it would re-run the phase, re-surface the still-open - # question, and loop indefinitely now that the force-advance backstop is - # gone. - return resolved_count + (1 if feedback_resolved else 0) - - def _await_unresolved_gap_gate( store: _pkg.Any, pipeline_id: str, @@ -897,6 +664,12 @@ def _set_status(status: _pkg.PipelineStatus) -> _pkg.Pipeline: # Mirror the phase_gate block: drive both pipeline and the phase # box so the DAG visualization renders the gate on the right # phase, and the operator's wait-status monitor wakes. + # + # Only the RUNNING restore goes through here, and only after the + # post-wait cancel check has cleared. The AWAITING_HUMAN park uses + # ``_park_at_gate_unless_cancelled`` instead, which re-reads the + # status from inside this same lock so it cannot overwrite an + # operator cancel (#3633 review round 3). with _pkg.get_pipeline_state_lock(pipeline_id): pipeline = store.load_pipeline(pipeline_id) pipeline.status = status @@ -928,6 +701,22 @@ def _set_status(status: _pkg.PipelineStatus) -> _pkg.Pipeline: "and fails CI (test_models_gaps.py) red on the PR.\n\n" f"{gap_lines}" ) + # Never mint a decision for a cancelled run. The cancel route sweeps + # the pending queue once; a decision created after that sweep has + # nobody to cancel it, and ``wait_for_decision`` below is an unbounded + # poll — the driver thread would block for the process lifetime and + # ``_run_pipeline``'s ``finally`` would never run (#3633 review round + # 3). ``gated`` is already True here, which is the same value an + # ordinary gating returns, so ``_run_implement_advance``'s own cancel + # re-check is what stops the driver. + if _pkg._pipeline_cancelled(store, pipeline_id): + _pkg.logger.info( + "Unresolved-gap gate: pipeline cancelled before the gate was " + "queued — leaving the persisted CANCELLED intact (#3633)", + pipeline_id=pipeline_id, + phase=phase.value, + ) + return gated decision = dq.queue_decision( question=question, context=context, @@ -937,8 +726,20 @@ def _set_status(status: _pkg.PipelineStatus) -> _pkg.Pipeline: ) # Mark AWAITING_HUMAN + surface to event watchers, mirroring the - # phase_gate block so the operator's wait-status monitor wakes. - pipeline = _set_status(_pkg.PipelineStatus.AWAITING_HUMAN) + # phase_gate block so the operator's wait-status monitor wakes — but + # only if the operator has not cancelled since the check above. The + # re-read happens inside the same state lock as the write (the cancel + # route persists CANCELLED under it), so the park cannot clobber the + # cancel that the post-wait check below is trying to read. + pipeline, _park_cancelled = _pkg._park_at_gate_unless_cancelled(store, pipeline_id, phase) + if _park_cancelled: + _pkg.logger.info( + "Unresolved-gap gate: pipeline cancelled before parking at " + "the gate — leaving the persisted CANCELLED intact (#3633)", + pipeline_id=pipeline_id, + phase=phase.value, + ) + return gated _pkg.report_pipeline_status( pipeline, event_type="phase.gap_gate", diff --git a/orchestrator/routes/pipelines/_run_hitl_gate.py b/orchestrator/routes/pipelines/_run_hitl_gate.py index f0bb0d9d3..4b07cc95e 100644 --- a/orchestrator/routes/pipelines/_run_hitl_gate.py +++ b/orchestrator/routes/pipelines/_run_hitl_gate.py @@ -57,10 +57,31 @@ def _gate_wait_cancelled(store, pipeline_id: str) -> bool: 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 + *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. + That qualifier is load-bearing: the sweep is a one-time snapshot of the + pending queue, so it only reaches decisions that already existed when the + cancel ran. A wait that returns for any *other* reason reads whatever the + store holds — including, before the in-lock park checks landed, the gate's + own ``AWAITING_HUMAN``. This function is therefore the *last* of three + checks each gate needs, not the only one: the other two are the pre-queue + check (so a post-sweep decision is never minted, since nothing would ever + unblock its wait) and the in-lock park check + (:func:`_park_at_gate_unless_cancelled`, so the park write cannot clobber + the cancel this one is trying to read). See ``_pipeline_cancelled``. + + **The phase box is deliberately left at ``AWAITING_HUMAN``.** When this + returns True the park write has already landed, and every caller breaks + without restoring the phase execution's status. That is a choice, not an + oversight: a second write to a pipeline the operator has already stopped + buys nothing, and leaving the box records *where* the run stopped — parked + at this gate — which is what an operator reading a cancelled pipeline's + phase timeline wants. All gates behave the same way (the gap, attestation, + and divergence-reconcile gates included), so a cancelled pipeline renders + consistently. + Delegates to ``_pipeline_cancelled``, inheriting its FAILED carve-out (#1273) and its best-effort store-hiccup tolerance. """ @@ -189,6 +210,20 @@ def _run_hitl_gate_converge( if _ledger_missing: dq = _pkg.get_decision_queue(pipeline_id, repo_path) + # Never mint a decision for a cancelled run. The cancel route + # sweeps the pending queue *once*; a decision created after that + # sweep has nobody to cancel it, and ``wait_for_decision`` is an + # unbounded poll — the driver thread would block for the process + # lifetime, skipping the ``finally`` that cleans up containers and + # preserves worktrees (#3633 review round 3). + if _pkg._pipeline_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled before the decision-ledger backstop " + "was queued — exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" _backstop = dq.queue_decision( question=( f"The {current_phase.value} phase reached its gate " @@ -206,12 +241,21 @@ def _run_hitl_gate_converge( decision_type="choice", phase=current_phase, ) - with _pkg.get_pipeline_state_lock(pipeline_id): - pipeline = store.load_pipeline(pipeline_id) - pipeline.status = _pkg.PipelineStatus.AWAITING_HUMAN - phase_execution = pipeline.get_phase_execution(current_phase) - phase_execution.status = _pkg.PipelineStatus.AWAITING_HUMAN - store.save_pipeline(pipeline) + # Park under the state lock, re-reading the status from inside it: + # the cancel route persists CANCELLED under the same lock, so an + # unconditional write here would silently lose it and the + # post-wait check below would read back our own AWAITING_HUMAN. + pipeline, _park_cancelled = _pkg._park_at_gate_unless_cancelled( + store, pipeline_id, current_phase + ) + if _park_cancelled: + _pkg.logger.info( + "Pipeline cancelled before parking at the decision-ledger " + "backstop — exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" _pkg.report_pipeline_status( pipeline, event_type="decision.created", @@ -481,6 +525,22 @@ def _run_hitl_gate_converge( _content_changed = gate_context != _prev_gate.context dq = _pkg.get_decision_queue(pipeline_id, repo_path) + # Everything above this line — the phase-draft reads, the + # human-companion read, the previous-gate scan — is seconds of IO + # the cancel route can land in. Minting the gate for a cancelled + # run leaves an orphan decision the one-time pending sweep already + # passed, so nothing will ever cancel it and the unbounded + # ``wait_for_decision`` below never returns: the driver thread + # leaks for the process lifetime and ``_run_pipeline``'s + # ``finally`` never runs (#3633 review round 3). + if _pkg._pipeline_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled before the phase gate was queued — " + "exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" decision = dq.queue_decision( question=question, context=gate_context, @@ -491,15 +551,24 @@ def _run_hitl_gate_converge( ) # Reload pipeline to pick up the decision persisted by queue_decision(), - # otherwise the stale local object overwrites it with an empty decisions list. - with _pkg.get_pipeline_state_lock(pipeline_id): - pipeline = store.load_pipeline(pipeline_id) - pipeline.status = _pkg.PipelineStatus.AWAITING_HUMAN - # Also mark the phase as awaiting human so the DAG visualization - # shows the HITL gate on the correct phase box. - phase_execution = pipeline.get_phase_execution(current_phase) - phase_execution.status = _pkg.PipelineStatus.AWAITING_HUMAN - store.save_pipeline(pipeline) + # otherwise the stale local object overwrites it with an empty decisions + # list — and park at AWAITING_HUMAN, but only if the operator has not + # cancelled in the meantime. The check lives inside the same state lock + # as the write (see ``_park_at_gate_unless_cancelled``): this write is + # on the reuse arm as well as the create arm, and an unconditional one + # is what let a cancel arriving before the gate be read back as the + # gate's own AWAITING_HUMAN and then advanced (#3633 review round 3). + pipeline, _park_cancelled = _pkg._park_at_gate_unless_cancelled( + store, pipeline_id, current_phase + ) + if _park_cancelled: + _pkg.logger.info( + "Pipeline cancelled before parking at the phase gate — " + "exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" # Report HITL gate to collaborator _pkg.report_pipeline_status( @@ -612,6 +681,20 @@ def _run_hitl_gate_converge( ) except _pkg.json.JSONDecodeError, TypeError, AttributeError: display_resolution = resolution + # Same one-time-sweep hazard as the gate above: a cancel landing + # between the gate's resolution and this follow-up would mint a + # decision nothing can cancel, and the wait below would never + # return (#3633 review round 3). This site never parks at + # AWAITING_HUMAN — the gate's park is still in force — so the + # pre-queue check is the only one it needs. + if _pkg._pipeline_cancelled(store, pipeline_id): + _pkg.logger.info( + "Pipeline cancelled before the gate follow-up was queued " + "— exiting the gate without advancing (#3633)", + pipeline_id=pipeline_id, + phase=current_phase.value, + ) + return pipeline, "break" followup = dq.queue_decision( question=( f'You selected "{display_resolution}" but didn\'t provide specific feedback. ' diff --git a/orchestrator/routes/pipelines/_run_support.py b/orchestrator/routes/pipelines/_run_support.py index f4adecb66..fd9ad63e9 100644 --- a/orchestrator/routes/pipelines/_run_support.py +++ b/orchestrator/routes/pipelines/_run_support.py @@ -103,17 +103,35 @@ def _pipeline_cancelled(store, pipeline_id: str) -> bool: 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. + cancel. That is why each of them re-checks from inside, at **both** writes: + + - before the park write, from inside the same state lock that performs it + (:func:`_park_at_gate_unless_cancelled`, and the bespoke in-lock checks + in ``_alerts.py`` and the gap gate's ``_set_status``). Re-checking only + after the wait is not enough: the park write itself clobbers the + operator's ``CANCELLED``, and the post-wait check then reads back the + gate's own ``AWAITING_HUMAN`` and concludes nothing happened; + - before the resume write, after the wait returns + (``_gate_wait_cancelled``, at all five of its blocking waits; + ``_await_unresolved_gap_gate``; the divergence-reconcile pause). + + The park check has to be in-lock rather than merely "just before", because + ``StateStore.update_pipeline`` — which is how the cancel route persists + ``CANCELLED`` — takes the same per-pipeline lock. Reading outside it races + the cancel and loses the update. + + A third check belongs *before* the decision is queued, at every site that + mints one: the cancel route sweeps pending decisions once, so a decision + minted after that sweep is never cancelled and ``wait_for_decision`` — an + unbounded poll with no timeout — blocks for the process lifetime, leaking + the driver thread and skipping ``_run_pipeline``'s ``finally`` entirely + (#3633 review round 3). Any new park-and-resume block needs all three. 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 + block therefore has to propagate a stop the driver acts on: the five ``_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 @@ -140,6 +158,47 @@ def _pipeline_cancelled(store, pipeline_id: str) -> bool: return False +def _park_at_gate_unless_cancelled(store, pipeline_id: str, phase) -> tuple[_pkg.Any, bool]: + """Persist ``AWAITING_HUMAN`` for pipeline+phase, unless the run is cancelled (#3633). + + Every operator-blocking gate parks the pipeline (and its phase box, so the + DAG visualization renders the gate on the right phase) at + ``AWAITING_HUMAN`` before blocking in ``wait_for_decision``. That write is + unconditional in every pre-#3633 gate, and it is the write that made the + persisted-status layers blind: it lands *on top of* the operator's + ``CANCELLED``, so the post-wait re-check reads back the gate's own + ``AWAITING_HUMAN``, concludes the run is live, and advances the phase. + + The check is performed **inside** the same per-pipeline state lock that + performs the write, not just before taking it. ``StateStore.update_pipeline`` + — the cancel route's persistence path — holds that lock, so an out-of-lock + read can observe RUNNING and then have the write land after the cancel's, + silently losing it. In-lock, the two interleavings are the only ones + possible: the cancel wins the lock (we see ``CANCELLED`` and skip), or we + do (the cancel's own sweep then unblocks the wait and the post-wait check + fires). + + Returns ``(pipeline, cancelled)``. On ``cancelled=True`` nothing was + written and the caller must **not** wait — it has to propagate a stop the + driver acts on (``"break"`` from the gate blocks, ``aborted=True`` from + the divergence pause); skipping the write alone leaves the driver walking + the DAG until the next unguarded write clobbers the cancel anyway. + + The returned ``pipeline`` is the freshly-loaded object in both cases, so + callers can rebind and keep reporting against current state. + """ + with _pkg.get_pipeline_state_lock(pipeline_id): + pipeline = store.load_pipeline(pipeline_id) + if pipeline.status == _pkg.PipelineStatus.CANCELLED: + return pipeline, True + pipeline.status = _pkg.PipelineStatus.AWAITING_HUMAN + phase_execution = pipeline.get_phase_execution(phase) + if phase_execution is not None: + phase_execution.status = _pkg.PipelineStatus.AWAITING_HUMAN + store.save_pipeline(pipeline) + return pipeline, False + + def _spawn_and_wait( spawner, pipeline_id: str, diff --git a/orchestrator/tests/test_cancel_stops_driver.py b/orchestrator/tests/test_cancel_stops_driver.py index a2eb36396..eefe36323 100644 --- a/orchestrator/tests/test_cancel_stops_driver.py +++ b/orchestrator/tests/test_cancel_stops_driver.py @@ -710,6 +710,8 @@ def _run_gate( attestation=None, bridge=None, on_wait=None, + on_draft=None, + dq=None, ): """Drive ``_run_hitl_gate_converge`` to its phase-gate wait and back. @@ -718,18 +720,38 @@ def _run_gate( would let its own ``AWAITING_HUMAN`` write mask the persisted status the bail re-reads. + ``save_pipeline`` writes the status back into the cell, which is what makes + the lost-update mode testable at all: in the real store a park write + *becomes* the persisted status, so a gate that parks AWAITING_HUMAN over a + cancel and then re-reads would read back its own write and sail on. A fake + that only records saves without applying them can never reproduce that. + 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. + + ``on_draft`` fires from ``_read_phase_draft``, which is the *pre*-wait + window: it sits after the phase's own cancel bail and before both the + create arm's ``queue_decision`` and the park write, so it models a cancel + landing during the seconds-to-minutes of git IO the gate does before it + ever blocks — the window the post-wait checks cannot see. + + Pass ``dq`` to supply your own decision-queue mock when a test needs to + assert on what the gate did — or did not — queue. """ 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) + + def _save(p, *_args, **_kwargs): + saved.append(p.status) + cell.status = p.status + + store.save_pipeline.side_effect = _save waits: list[str] = [] @@ -739,7 +761,7 @@ def _wait(decision_id): on_wait(len(waits), cell) return resolved_decision - dq = MagicMock() + dq = dq if dq is not None else MagicMock() dq.wait_for_decision.side_effect = _wait dq.get_decision.return_value = resolved_decision @@ -753,6 +775,11 @@ def _bridge(*args, **kwargs): return bridge(cell) return 0 + def _draft(*_args, **_kwargs): + if on_draft is not None: + on_draft(cell) + return "draft body" + with ( patch.object( pipelines_pkg, @@ -769,7 +796,7 @@ def _bridge(*args, **kwargs): 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_phase_draft", side_effect=_draft), 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"), @@ -801,6 +828,12 @@ def test_gate_bails_when_the_cancel_route_cancels_its_decision(): assert PipelineStatus.RUNNING not in saved, ( "the gate rewrote the operator's CANCELLED back to RUNNING" ) + # Stronger than "no RUNNING": the gate must not persist *any* non-terminal + # status over CANCELLED. The park write is a status write too, and because + # the fake now applies saves back onto the cell, an AWAITING_HUMAN park here + # would be read back by the post-wait check as "not cancelled" and the gate + # would advance — the lost-update mode the in-lock park check closes. + assert saved == [], f"the gate persisted {saved} over the operator's CANCELLED" def test_gate_bails_on_a_cancelled_pipeline_even_if_the_decision_resolved(): @@ -911,6 +944,123 @@ def test_gate_bails_when_cancelled_bridging_contract_decisions(): ) +def _gate_pipeline_without_a_pending_gate() -> Pipeline: + """A pipeline at the plan gate with no pending decision — the *create* arm. + + The reuse arm waits on a decision the cancel route's sweep already reached; + the create arm mints a new one, which is the arm that can leave an orphan. + """ + pipeline = _gate_pipeline() + pipeline.decisions = [] + return pipeline + + +def test_gate_never_mints_a_decision_for_a_cancelled_pipeline(): + """The create arm must not queue a gate for a run the operator stopped. + + The cancel route's sweep of pending decisions is a one-time snapshot, so a + decision minted after it runs is never cancelled — and + ``DecisionQueue.wait_for_decision`` is a ``while True`` poll with no + timeout. Queueing here parks the driver thread for the lifetime of the + process: ``_run_pipeline``'s ``finally`` never runs, so there is no + container cleanup and no ``skip_cleanup`` worktree preservation, and the + operator's PATCH already returned 200. That is strictly worse than the + pre-#3633 behaviour on the same input, where the gate at least fell + through — which is why the check goes *before* ``queue_decision``. + """ + dq = MagicMock() + action, saved = _run_gate( + _gate_decision(DecisionStatus.CANCELLED), + persisted_status=PipelineStatus.AWAITING_HUMAN, + pipeline=_gate_pipeline_without_a_pending_gate(), + # The cancel lands while the gate is reading the draft — after the + # phase's own bail, before the queue. + on_draft=lambda cell: cell.cancel(), + dq=dq, + ) + + assert action == "break" + assert dq.queue_decision.call_args_list == [], ( + "the gate minted a decision nothing will ever cancel" + ) + dq.wait_for_decision.assert_not_called() + assert saved == [] + + +def test_gate_park_does_not_overwrite_a_cancel_that_lands_before_it(): + """The reuse arm's park write must not clobber the operator's CANCELLED. + + This is #3633 verbatim, through the half of the window the post-wait checks + cannot see. The park write is unconditional on both arms, so a cancel + landing before it is overwritten with ``AWAITING_HUMAN``; the decision was + pending at sweep time so the wait returns at once with no resolution; the + post-wait check re-reads the store and sees the gate's *own* write, not the + cancel; and ``"" in _APPROVE_KEYWORDS`` sends the run down "Approved — + resume and advance". Hence the in-lock check in + ``_park_at_gate_unless_cancelled``: ``StateStore.update_pipeline`` takes + the same per-pipeline lock, so checking inside it is what makes the read + and the write atomic against the cancel route. + """ + dq = MagicMock() + action, saved = _run_gate( + # A resolution that *would* advance the phase if the bail were missed. + _gate_decision(DecisionStatus.RESOLVED, "approve"), + persisted_status=PipelineStatus.AWAITING_HUMAN, + on_draft=lambda cell: cell.cancel(), + dq=dq, + ) + + assert action == "break" + assert saved == [], f"the park wrote {saved} over the operator's CANCELLED" + assert dq.wait_for_decision.call_args_list == [], ( + "the gate blocked on a decision belonging to a cancelled run" + ) + + +def test_park_at_gate_unless_cancelled_checks_inside_the_lock(): + """Unit coverage for the shared park helper. + + The ordering assertion is the point: a check that merely runs *just before* + the write still races ``update_pipeline``, which holds the same + per-pipeline lock. Only a check taken after the lock is acquired and before + the save is atomic against the cancel route. + """ + events: list[str] = [] + + lock = MagicMock() + lock.__enter__.side_effect = lambda: events.append("enter") + lock.__exit__.side_effect = lambda *_a: events.append("exit") + + def _store_for(pipeline): + store = MagicMock() + store.load_pipeline.side_effect = lambda _pid: (events.append("load"), pipeline)[1] + store.save_pipeline.side_effect = lambda *_a, **_k: events.append("save") + return store + + live = _store_for(_cancellable_pipeline()) + with patch.object(pipelines_pkg, "get_pipeline_state_lock", return_value=lock): + pipeline, cancelled = pipelines_pkg._park_at_gate_unless_cancelled( + live, PIPELINE_ID, PipelinePhase.PLAN + ) + + assert cancelled is False + assert pipeline.status == PipelineStatus.AWAITING_HUMAN + assert events == ["enter", "load", "save", "exit"], ( + "the status read must sit inside the lock that performs the write" + ) + + events.clear() + dead = _store_for(_cancellable_pipeline(status=PipelineStatus.CANCELLED)) + with patch.object(pipelines_pkg, "get_pipeline_state_lock", return_value=lock): + pipeline, cancelled = pipelines_pkg._park_at_gate_unless_cancelled( + dead, PIPELINE_ID, PipelinePhase.PLAN + ) + + assert cancelled is True + assert pipeline.status == PipelineStatus.CANCELLED + assert events == ["enter", "load", "exit"], "a cancelled run must not be written to" + + 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 diff --git a/orchestrator/tests/test_contract_decision_bridge.py b/orchestrator/tests/test_contract_decision_bridge.py index 2f55d062d..f95d8b485 100644 --- a/orchestrator/tests/test_contract_decision_bridge.py +++ b/orchestrator/tests/test_contract_decision_bridge.py @@ -278,12 +278,19 @@ def test_bridge_abandons_remaining_waits_when_the_pipeline_is_cancelled( ) -> 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. + This pins the interleaving the ``cancelled`` predicate actually closes: the + cancel lands *after* pass 1 queued the batch, so the route's one-time sweep + reached every entry. Each swept wait returns immediately with no resolution + — and an unset resolution reads as an approval — so without the predicate + the bridge would walk the whole remaining batch answering the operator's + cancel as consent. It stops after the wait that observed the cancel. + + The predicate is deliberately *not* what saves the other interleaving — a + cancel landing entirely before pass 1, where nothing was queued to sweep + and the first wait would never return. Nothing after that first wait runs, + so no post-wait check can help; that case is closed by the pre-pass-1 check + covered in ``test_bridge_never_queues_for_an_already_cancelled_pipeline`` + (#3633 review round 4). """ from routes.pipelines import _queue_and_await_contract_decisions @@ -339,6 +346,58 @@ def wait_for_decision(self, decision_id: str) -> HITLDecision: assert resolved_ids == ["decision-1"] +def test_bridge_never_queues_for_an_already_cancelled_pipeline( + tmp_path: Path, +) -> None: + """A cancel that lands before pass 1 must stop the bridge before it queues. + + This is the interleaving no post-wait check can reach (#3633 review round + 4). The cancel route's sweep of pending decisions is a one-time snapshot, + so a batch minted after it runs is never cancelled — and + ``DecisionQueue.wait_for_decision`` is an untimed poll. Queue here and the + first wait blocks for the process lifetime, taking the driver thread and + ``_run_pipeline``'s ``finally`` (container cleanup, worktree preservation) + with it. The pre-pass-1 check is what keeps the batch unminted. + """ + from routes.pipelines import _queue_and_await_contract_decisions + + identifier = "issue-3633-precancel" + _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) + ], + ) + + dq = _FakeQueue(resolutions=["a", "b"]) + + resolved_count = _queue_and_await_contract_decisions( + dq, + tmp_path, + "pipeline-id", + identifier, + PipelinePhase.REFINE, + # Already cancelled when the bridge is entered. + cancelled=lambda: True, + ) + + assert resolved_count == 0 + assert dq.queued == [], "the bridge minted decisions nothing will ever cancel" + + def test_bridge_without_a_cancelled_predicate_answers_every_decision( tmp_path: Path, ) -> None: diff --git a/orchestrator/tests/test_hard_reset_recovery.py b/orchestrator/tests/test_hard_reset_recovery.py index b14753c8e..37638111f 100644 --- a/orchestrator/tests/test_hard_reset_recovery.py +++ b/orchestrator/tests/test_hard_reset_recovery.py @@ -516,6 +516,56 @@ def _wait(_decision_id): # Bailed before the resolution read, so no second sync attempt. mock_sync.assert_called_once() + def test_cancel_before_the_pause_is_not_overwritten_by_the_park(self): + """The park write is the other half of the window (#3633 review round 4). + + The pause reloads inside the state lock and writes AWAITING_HUMAN to + the pipeline and its phase box. A cancel landing before that — during + the sync itself, which is git IO over the gateway — used to be + overwritten, and the post-wait check would then read back the pause's + own AWAITING_HUMAN rather than the operator's CANCELLED. Because + ``StateStore.update_pipeline`` takes the same per-pipeline lock, doing + the check inside it makes the read and the write atomic against the + cancel route: nothing is persisted, and no decision is minted for a run + that is already stopped. + """ + saved: list[PipelineStatus] = [] + store = MagicMock() + + def _load(_pipeline_id): + loaded = MagicMock() + loaded.status = PipelineStatus.CANCELLED + return loaded + + store.load_pipeline.side_effect = _load + store.save_pipeline.side_effect = lambda p, *a, **k: saved.append(p.status) + dq = MagicMock() + lock_p, persist_p, report_p, emit_p = self._patch_ctx() + with ( + patch( + "routes.pipelines._sync_worktree_with_remote", + return_value=_diverged_outcome(), + ), + patch("routes.pipelines.get_decision_queue", return_value=dq), + lock_p, + persist_p as mock_persist, + 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 saved == [], f"the pause persisted {saved} over the operator's CANCELLED" + mock_persist.assert_not_called() + dq.wait_for_decision.assert_not_called() + def test_reconcile_budget_exhausted_aborts(self): """If every resume re-diverges, the bounded budget eventually aborts rather than pausing forever."""