devflow 2.5.0

//! Pipeline seam C (D-06): stage transitions, gate firing and resolution,
//! loop-backs, workflow completion, and abort. Extracted mechanically
//! (19-08, D-09 pure move) out of `main.rs` — every function below is
//! byte-identical to its pre-move body modulo an added `pub(crate)` and
//! adjusted `use` paths.
//!
//! **This module closes the pipeline's three-way module cycle
//! (19-RESEARCH.md Pattern 1):** [`transition`] and [`loop_back_to_code`]
//! both call [`crate::pipeline_launch::launch_stage`] at their final step —
//! that call is what closes the cycle `pipeline_launch (advance) →
//! pipeline_outcomes (handle_*_outcome) → pipeline_gate (transition/
//! run_gate/finish_workflow) → pipeline_launch (launch_stage)` back to
//! where it started. This cycle is the state machine's real control flow
//! (Code → Validate → Ship, with loop-backs), and Rust permits cyclic
//! module references — only the crate dependency graph must be acyclic —
//! so this compiles cleanly. **A future change to pipeline logic is likely
//! to touch two or three of these files together:** the split buys
//! `pub(crate)` boundaries, reviewability, and wave independence for the
//! *other* clusters, not pipeline-internal parallelism (19-RESEARCH.md
//! Pitfall 1).

use crate::CliError;
use crate::config_parse::{foreground_gate_timeout_secs, gate_timeout_secs};
use crate::pipeline_launch::launch_stage;
use crate::pipeline_outcomes::{run_checkout_hooks, truncate_reason};
use devflow_core::gates::{self, GateAction, GateError, GateResponse, Gates};
use devflow_core::hooks;
use devflow_core::mode;
use devflow_core::phase_id::PhaseId;
use devflow_core::prompt::{self, FixType};
use devflow_core::stage::Stage;
use devflow_core::state::State;
use devflow_core::{events, lock, registry, workflow};
use std::path::Path;
use tracing::info;

/// Fire the hooks for `from → to`, persist the new stage, and launch its agent.
///
/// `infra_failures` resets unconditionally on every successful transition
/// (CR-01, 17-06 gap closure). Without this, an infra-fault ceiling meant to
/// bound a *stuck loop* (D-08, [`mode::MAX_INFRA_FAILURES`]) instead
/// accumulates across a phase's entire lifetime — several well-spaced,
/// cleanly-resolved infra faults would falsely reach the ceiling and
/// hard-abort a long-running but otherwise healthy phase.
///
/// `consecutive_failures` clears on every transition EXCEPT Code→Validate
/// (18d, [`mode::transition_resets_consecutive_failures`]): that hop is
/// crossed on every single Code↔Validate retry cycle, so unconditionally
/// clearing it there made [`mode::MAX_CONSECUTIVE_FAILURES`] unreachable for
/// the exact loop it bounds. The two counters deliberately no longer share a
/// single reset condition.
pub(crate) fn transition(
    project_root: &Path,
    state: &mut State,
    to: Stage,
) -> Result<(), CliError> {
    let from = state.stage;

    // 20c: `devflow start --until <stage>` halts cleanly once the requested
    // stage has completed — checked here, at the TOP, before anything else
    // runs. `stop_until == Some(from)` means the JUST-COMPLETED stage (the
    // one whose outcome triggered this call) is the requested stop point;
    // checking `to` instead would halt BEFORE the target stage ever ran
    // (review: Codex HIGH off-by-one). This bypasses the from→to checkout
    // hooks, `state.stage = to`, the normal `"transition"` event, and
    // `launch_stage` — no new monitor is spawned, and `loop_back_to_code`
    // (a retry, not an advance) is untouched by this check.
    if state.stop_until == Some(from) {
        state.stopped = true;
        state.stop_reason = Some(format!("stopped after {from} completed (--until {from})"));
        // T-20-03b / doctor gap (D-09): clear monitor_pid/gate_pending so
        // neither check_dead_agent nor check_dead_monitor misreports this
        // intentional stop as a crashed agent or dead monitor.
        state.monitor_pid = None;
        state.gate_pending = false;
        workflow::save_state(state)?;
        events::emit(
            project_root,
            state.phase,
            "workflow_finished",
            serde_json::json!({
                "reason": "stopped_at",
                "stage": from.to_string(),
            }),
        );
        return Ok(());
    }

    let _ = run_checkout_hooks(
        project_root,
        state,
        &hooks::hooks_for_transition(from, to),
        to,
    );
    state.stage = to;
    if mode::transition_resets_consecutive_failures(from, to) {
        state.consecutive_failures = 0;
    }
    state.infra_failures = 0;
    state.gate_pending = false;
    workflow::save_state(state)?;
    events::emit(
        project_root,
        state.phase,
        "transition",
        serde_json::json!({
            "from": from.to_string(),
            "to": to.to_string(),
        }),
    );
    launch_stage(state, None, Some(from))
}

/// Why the pipeline is looping back to Code, recorded on the `loop_back`
/// event so `events.jsonl` distinguishes cases that are indistinguishable
/// from the counters alone (IN-02, 999.78).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum LoopBackReason {
    /// A Validate failure recorded against an existing commit-count baseline
    /// — the ordinary case.
    ValidateFailure,
    /// A Validate failure recorded while NO commit-count baseline existed for
    /// this phase (IN-02). `last_validate_failure_commit_count`'s `None` means
    /// both "genuine first failure of this phase" and "state written by a
    /// binary predating that field", and nothing in `events.jsonl` told the
    /// two apart — so an operator who upgraded a binary mid-phase got no
    /// signal that the failure budget had widened back to its full width.
    /// This reason is that signal.
    ValidateFailureNoBaseline,
    /// A human answered a gate with a loop-back — a Ship `review:` rejection,
    /// an ambiguous Validate adjudication, or a finalization retry. Makes no
    /// claim about the commit-count baseline, because none was consulted.
    GateResponse,
}

impl LoopBackReason {
    /// The stable string recorded on the `loop_back` event. Distinct per
    /// variant by construction: an operator greps these, so two variants
    /// sharing a string would silently re-merge the cases IN-02 separated.
    pub(crate) fn as_str(self) -> &'static str {
        match self {
            LoopBackReason::ValidateFailure => "validate_failure",
            LoopBackReason::ValidateFailureNoBaseline => "validate_failure_no_commit_baseline",
            LoopBackReason::GateResponse => "gate_response",
        }
    }
}

/// Loop the pipeline back to Code with the given fix prompt (`GapsOnly` for a
/// Validate rejection, `AuditFix` for a Ship `review:` rejection).
pub(crate) fn loop_back_to_code(
    project_root: &Path,
    state: &mut State,
    fix: FixType,
    reason: LoopBackReason,
) -> Result<(), CliError> {
    let from = state.stage;
    let prompt = prepare_loop_back_to_code(project_root, state, fix, reason)?;
    launch_stage(state, Some(prompt), Some(from))
}

/// The state-mutating half of `loop_back_to_code`, split out so it's
/// unit-testable without spawning a real agent process (`launch_stage`
/// invokes the actual configured agent CLI). Cleans up the stale gate for
/// the stage the gate fired on (CR-01), moves `state` to Code, persists it,
/// and returns the fix prompt the caller should launch with.
pub(crate) fn prepare_loop_back_to_code(
    project_root: &Path,
    state: &mut State,
    fix: FixType,
    reason: LoopBackReason,
) -> Result<String, CliError> {
    // Capture the stage the gate actually fired on before it's mutated below,
    // so cleanup targets the right stage's gate files (see CR-01: a stale
    // response/ack left on disk after a loop-back is silently reused by a
    // later gate for the same phase+stage).
    let gate_stage = state.stage;
    let _ = Gates::cleanup(project_root, state.phase, gate_stage);
    state.stage = Stage::Code;
    state.gate_pending = false;
    workflow::save_state(state)?;
    events::emit(
        project_root,
        state.phase,
        "loop_back",
        serde_json::json!({
            "from": gate_stage.to_string(),
            "consecutive_failures": state.consecutive_failures,
            "phase_validate_failures": state.phase_validate_failures,
            "reason": reason.as_str(),
            "fix": format!("{fix:?}"),
        }),
    );
    println!(
        "looping back to Code ({} validate failure(s) this phase, {} in the current streak)",
        state.phase_validate_failures, state.consecutive_failures
    );
    Ok(prompt::fix_prompt(fix, state.phase))
}

/// Run the terminal hooks (version bump + branch cleanup) and clear state.
///
/// Uses [`gate_timeout_secs`]'s multi-day production default for the
/// retry-gate wait below — safe for every caller EXCEPT the foreground
/// `ship_override` path (WR-02), which calls
/// [`finish_workflow_with_gate_timeout`] directly with a bounded timeout
/// instead.
pub(crate) fn finish_workflow(project_root: &Path, state: &mut State) -> Result<(), CliError> {
    finish_workflow_with_gate_timeout(project_root, state, gate_timeout_secs())
}

/// `finish_workflow`'s body, parameterized on how long the retry-gate poll
/// (below) waits for a response before failing fast (WR-02, phase 20
/// review). Every caller reached through a detached monitor process should
/// keep using `finish_workflow`'s multi-day default — invisible to an
/// operator's terminal by construction. `ship_override` is the one caller
/// invoked directly from the foreground CLI, so it passes
/// [`crate::config_parse::foreground_gate_timeout_secs`] here instead,
/// bounding how long a terminal-hook failure can block the operator's shell
/// without weakening the fail-closed terminal-Ship invariant: an unanswered
/// gate still fails the operation entirely (via `run_gate_with_timeout`'s
/// existing timeout error), just after seconds instead of days.
pub(crate) fn finish_workflow_with_gate_timeout(
    project_root: &Path,
    state: &mut State,
    gate_timeout_secs: u64,
) -> Result<(), CliError> {
    loop {
        if run_checkout_hooks(project_root, state, &hooks::hooks_after_ship(), Stage::Ship) {
            break;
        }
        // The original Ship approval has already been consumed. Reopen an
        // actionable gate and keep this monitor waiting so a terminal-hook
        // failure cannot turn into an invisible stalled Ship state.
        let _ = Gates::cleanup(project_root, state.phase, Stage::Ship);
        let context = format!(
            "[finalization failed] phase {} terminal hooks did not complete. Resolve the git/version error, then approve to retry; reject to loop back or abort.",
            state.phase
        );
        // This is the reopened finalization-retry gate, NEVER the routine
        // Ship approval — it must always pass `None` here. Auto-approving
        // it would mean "the merge could not be completed" gets silently
        // retried forever with no human ever seeing it (T-23-91). The one
        // and only site permitted to pass a non-`None` auto-response is
        // `handle_ship_outcome` in `pipeline_outcomes.rs`.
        match run_gate_with_timeout(
            project_root,
            state,
            Stage::Ship,
            &context,
            gate_timeout_secs,
            None,
        )? {
            GateAction::Advance => {
                let _ = Gates::cleanup(project_root, state.phase, Stage::Ship);
            }
            GateAction::LoopBack(_) => {
                return loop_back_to_code(
                    project_root,
                    state,
                    FixType::AuditFix,
                    LoopBackReason::GateResponse,
                );
            }
            GateAction::Abort(reason) => return abort(project_root, state, &reason),
        }
    }
    let _ = Gates::cleanup(project_root, state.phase, Stage::Validate);
    let _ = Gates::cleanup(project_root, state.phase, Stage::Ship);
    workflow::clear_state(project_root, state.phase)?;
    // 23b: the workflow is genuinely over — deregister this (project_root,
    // phase) from the machine-global registry so `devflow gate list
    // --all-roots` stops naming a phase that no longer exists.
    registry::deregister(project_root, state.phase);
    // 23-06 / T-23-67: the ONLY site in the workspace permitted to emit
    // `workflow_shipped`. This is the strict shipped predicate `ship_evidence`
    // reads — deliberately NOT `workflow_finished` below, which is also
    // emitted by `transition`'s `--until` clean-stop branch (`pipeline_gate.rs`,
    // `state.stop_until == Some(from)` arm) with a `"stopped_at"` reason,
    // before any hook, stage assignment, or launch runs. A phase halted after
    // one stage must never read as shipped. Ordering here is load-bearing in
    // both directions: after the hook-success loop `break` above, so it can
    // only fire once the entire `hooks_after_ship` batch has succeeded; and
    // before `workflow_finished` below, because existing tests assert a
    // phase's event stream ENDS in `workflow_finished` — do not move this
    // emission after it.
    events::emit(
        project_root,
        state.phase,
        "workflow_shipped",
        serde_json::json!({
            "stage": Stage::Ship.to_string(),
        }),
    );
    events::emit(
        project_root,
        state.phase,
        "workflow_finished",
        serde_json::Value::Null,
    );
    println!("phase {} shipped — workflow complete", state.phase);
    Ok(())
}

/// Write a gate file and block (in the detached monitor) until a response or
/// the long poll timeout. Acks the response so the Hermes poller can clean up.
pub(crate) fn run_gate(
    project_root: &Path,
    state: &mut State,
    stage: Stage,
    context: &str,
) -> Result<GateAction, CliError> {
    run_gate_with_timeout(
        project_root,
        state,
        stage,
        context,
        gate_timeout_secs(),
        None,
    )
}

/// `run_gate`'s body, parameterized on the poll timeout (WR-02, phase 20
/// review) so [`finish_workflow_with_gate_timeout`]'s foreground retry-gate
/// wait can pass a bounded timeout instead of [`gate_timeout_secs`]'s
/// multi-day production default.
pub(crate) fn run_gate_with_timeout(
    project_root: &Path,
    state: &mut State,
    stage: Stage,
    context: &str,
    timeout_secs: u64,
    // 23-09 (D-04/D-05/D-06): a pre-authorized response to write for this
    // gate, or `None` for a real human gate. This function must NEVER
    // derive this value from `state` (e.g. `state.yes_ship`) — the decision
    // belongs to the caller precisely so the reopened finalization-retry
    // gate in `finish_workflow_with_gate_timeout` cannot inherit an
    // authorization meant only for the routine Ship approval. The regression
    // test `finalization_retry_gate_never_auto_approves_even_with_yes_ship_set`
    // (pipeline_outcomes.rs) fails loudly if a future refactor reads the
    // flag from `state` inside this function instead of receiving it here.
    auto_response: Option<&GateResponse>,
) -> Result<GateAction, CliError> {
    state.gate_pending = true;
    workflow::save_state(state)?;
    Gates::write_gate(project_root, state.phase, stage, context)?;
    println!(
        "gate written: .devflow/gates/{}-{stage}.json — awaiting response",
        state.phase.padded()
    );
    // A gate is "unexpected" when the active mode would not normally fire
    // one for this stage (e.g. a Define/Plan/Code failure in Auto mode) —
    // WR-11's never-silent path gates unconditionally, independent of mode.
    let unexpected = !state.mode.should_gate(
        stage,
        state.consecutive_failures,
        state.phase_validate_failures,
    );
    if unexpected {
        info!(
            "never-silent gate: {stage} failed in {:?} mode — surfacing an unattended gate this mode would not normally fire",
            state.mode
        );
    }
    events::emit(
        project_root,
        state.phase,
        "gate_fired",
        serde_json::json!({
            "stage": stage.to_string(),
            "unexpected": unexpected,
            "context": context,
        }),
    );
    gates::fire_gate_notify(state.phase, stage, context, unexpected);
    events::emit(
        project_root,
        state.phase,
        "notify_fired",
        serde_json::json!({ "stage": stage.to_string(), "unexpected": unexpected }),
    );
    // The auto-response, if any, is written here — after the gate request
    // and both `gate_fired`/`notify_fired` events, so the event stream still
    // reads as a real gate that was really answered, and before
    // `Gates::poll_response` below so the poll's very first read finds it
    // (`Gates::respond` refuses with `NoOpenGate` if written before
    // `Gates::write_gate` above, and `poll_response` would never observe a
    // response written after it returns).
    if let Some(response) = auto_response {
        match Gates::respond(project_root, state.phase, stage, response) {
            Ok(_) => {}
            // A human or 23b's stale-gate sweep may have answered first;
            // first-writer-wins is the correct resolution, not an error.
            Err(GateError::AlreadyResponded { .. }) => {}
            Err(err) => return Err(err.into()),
        }
    }
    match Gates::poll_response(project_root, state.phase, stage, timeout_secs) {
        Some(response) => {
            state.gate_pending = false;
            workflow::save_state(state)?;
            Gates::ack(project_root, state.phase, stage)?;
            let action = GateAction::from_response(&response);
            events::emit(
                project_root,
                state.phase,
                "gate_resolved",
                serde_json::json!({
                    "stage": stage.to_string(),
                    "approved": response.approved,
                    "action": match &action {
                        GateAction::Advance => "advance",
                        GateAction::LoopBack(_) => "loop_back",
                        GateAction::Abort(_) => "abort",
                    },
                    "responded_by": response.responded_by,
                }),
            );
            Ok(action)
        }
        None => {
            events::emit(
                project_root,
                state.phase,
                "gate_timeout",
                serde_json::json!({ "stage": stage.to_string() }),
            );
            Err(CliError::Message(format!(
                "gate for stage {stage} timed out awaiting a response"
            )))
        }
    }
}

/// Abort the workflow with a reason, clearing state.
pub(crate) fn abort(project_root: &Path, state: &State, reason: &str) -> Result<(), CliError> {
    println!("workflow aborted for phase {}: {reason}", state.phase);
    // See CR-01: without this, a stale response/ack for this phase+stage
    // survives on disk and is silently reused if the gate fires again later.
    let _ = Gates::cleanup(project_root, state.phase, state.stage);
    let _ = workflow::clear_state(project_root, state.phase);
    // 23b: an abort also ends the workflow — deregister the same as the
    // success path so an aborted phase does not linger in the registry.
    registry::deregister(project_root, state.phase);
    events::emit(
        project_root,
        state.phase,
        "workflow_aborted",
        serde_json::json!({ "reason": truncate_reason(reason) }),
    );
    Ok(())
}

/// Manual ship override (20e, D-01): a second, out-of-process consumer of
/// the SAME on-disk Ship gate response `run_gate`'s live blocking poll
/// consumes. `devflow gate approve` only WRITES a response file
/// (`Gates::respond`) — a live monitor polling `Gates::poll_response` is
/// what actually advances the workflow. If that monitor died before
/// consuming the response, the approval sits unconsumed forever; this
/// function reads the already-written response directly and, on
/// `GateAction::Advance`, drives the SAME `finish_workflow` the live poll
/// loop would have called (D-01) — not a reimplementation of the after-ship
/// hook batch.
///
/// Guard order (D-02, review: Codex HIGH + MEDIUM, Hermes ack-race):
/// 1. Acquire the per-phase lock ([`lock::acquire`]) BEFORE touching state,
///    so this can never race a still-live monitor's `poll_response` — the
///    exact idiom [`crate::pipeline_launch::resume`] uses.
/// 2. `state.stage` must be EXACTLY `Stage::Ship`; any earlier stage is
///    refused, naming the stage to resolve first.
/// 3. Both the Ship gate REQUEST (`Gates::gate_path`) and RESPONSE
///    (`Gates::response_path`) must exist on disk.
/// 4. The ACK (`Gates::ack_path`) must be ABSENT — its presence means a
///    (now-dead) monitor already consumed this response and may have died
///    mid-`finish_workflow`; re-running the terminal hooks in that state
///    would risk a double-run, so this refuses and directs the operator to
///    `devflow doctor` instead.
///
/// `force` is accepted and echoed in the CLI output for explicit operator
/// auditability (Hermes LOW: make `--force` semantics explicit), but is
/// deliberately NOT consulted by any guard above — D-02 scopes `--force` to
/// "skip Ship-gate re-verification," never to bypassing the stage, lock,
/// gate-existence, or ack checks themselves. This design has no separate
/// Ship-gate re-verification step beyond those four guards, so `--force`
/// currently changes no observable behavior; it exists so the flag is never
/// silently ignored and cannot later be wired to widen scope without an
/// explicit, reviewed change to this function.
pub(crate) fn ship_override(
    project_root: &Path,
    phase: PhaseId,
    force: bool,
) -> Result<(), CliError> {
    let _lock = match lock::acquire(project_root, phase) {
        Ok(guard) => guard,
        Err(lock::LockError::Contended { pid, .. }) => {
            return Err(CliError::Message(format!(
                "phase {phase}: another devflow process (pid {pid}) holds the per-phase lock — \
                 refusing to race its poll of the Ship gate response"
            )));
        }
        Err(err) => return Err(CliError::Message(format!("lock error: {err}"))),
    };

    let mut state = workflow::load_state(project_root, phase)?;

    if state.stage != Stage::Ship {
        return Err(CliError::Message(format!(
            "phase {phase} is at stage {} — `devflow ship` requires state.stage == Stage::Ship; \
             resolve stage {} first (--force does not skip stages)",
            state.stage, state.stage
        )));
    }

    if !Gates::gate_path(project_root, phase, Stage::Ship).exists()
        || !Gates::response_path(project_root, phase, Stage::Ship).exists()
    {
        return Err(CliError::Message(format!(
            "phase {phase}: no Ship gate response written yet — nothing to ship (a dead monitor \
             never wrote or received one; wait for `devflow gate approve` or resolve the pipeline first)"
        )));
    }

    if Gates::ack_path(project_root, phase, Stage::Ship).exists() {
        return Err(CliError::Message(format!(
            "phase {phase}: the Ship gate response was already consumed (an ack file is present) \
             — the phase may be mid-finalization from a monitor that died partway through; run \
             `devflow doctor` to inspect it rather than re-running terminal hooks"
        )));
    }

    let response_path = Gates::response_path(project_root, phase, Stage::Ship);
    let contents = std::fs::read_to_string(&response_path).map_err(|err| {
        CliError::Message(format!("could not read the Ship gate response: {err}"))
    })?;
    let response: GateResponse = serde_json::from_str(&contents).map_err(|err| {
        CliError::Message(format!("could not parse the Ship gate response: {err}"))
    })?;

    println!(
        "phase {phase}: manual ship override (--force={force}) — driving the already-written \
         Ship response through the same terminal path the live monitor would have used"
    );

    match GateAction::from_response(&response) {
        GateAction::Advance => {
            // WR-02 (phase 20 review): finish_workflow's retry-gate wait
            // (on a terminal-hook failure) normally uses gate_timeout_secs'
            // multi-day production default, invisible to an operator
            // because every OTHER caller runs inside a detached monitor.
            // ship_override runs in the FOREGROUND CLI — bound the wait so
            // a hook failure fails fast with an actionable message instead
            // of blocking this shell for days.
            let timeout = foreground_gate_timeout_secs();
            println!(
                "phase {phase}: if terminal-hook finalization fails, this foreground command \
                 will wait up to {timeout}s for the reopened Ship gate before failing (vs. the \
                 multi-day background default) — set DEVFLOW_FOREGROUND_GATE_TIMEOUT_SECS to \
                 change this"
            );
            finish_workflow_with_gate_timeout(project_root, &mut state, timeout)
        }
        GateAction::LoopBack(_) => {
            // Antigravity LOW: `loop_back_to_code` → `launch_stage` forks a
            // NEW detached monitor daemon — say so explicitly, so `devflow
            // ship` is not a silently long-running process the operator
            // can't account for.
            println!(
                "phase {phase}: Ship response loops back to Code — launching a new, detached \
                 monitor agent to drive the retry"
            );
            loop_back_to_code(
                project_root,
                &mut state,
                FixType::AuditFix,
                LoopBackReason::GateResponse,
            )
        }
        GateAction::Abort(reason) => abort(project_root, &state, &reason),
    }
}

/// Print the full pipeline that a `start` would run, without launching anything.
pub(crate) fn print_dry_run(state: &State) {
    println!(
        "dry run — phase {} | agent {} | mode {}",
        state.phase, state.agent, state.mode
    );
    println!("\nstage pipeline:");
    let mut stage = Some(Stage::Define);
    while let Some(s) = stage {
        let command = s.gsd_command().replace("{N}", &state.phase.to_string());
        // F-7: these are PREDICTION probes, not live reads. Every one passes
        // literals rather than `state.consecutive_failures` /
        // `state.phase_validate_failures` on purpose — the preview answers
        // "what will this pipeline do", not "where is this run right now",
        // and a dry run is printed before a run has any position to report.
        // Widening `should_gate` forces these open; closing them with a
        // placeholder would leave `--dry-run` silently no longer predicting
        // the per-phase ceiling gate, which is the operator's only advance
        // account of where a run will stop.
        let unconditional = state.mode.should_gate(s, 0, 0);
        let gate = if unconditional {
            " [GATE]".to_string()
        } else if state.mode.should_gate(s, mode::MAX_CONSECUTIVE_FAILURES, 0) {
            format!(
                " [GATE after {} consecutive failures]",
                mode::MAX_CONSECUTIVE_FAILURES
            )
        } else {
            String::new()
        };
        // The per-phase ceiling probe is a SEPARATE clause, not a third `else
        // if`. As an `else if` after the streak probe it is unreachable for
        // Validate in both modes — Auto's streak probe already returns true
        // and Supervise's unconditional probe already returns true — so the
        // preview would silently never name the new gate, which is precisely
        // the T-35-20c harm F-7 exists to prevent. Measured before the fix:
        // `devflow start --phase 7 --mode auto --dry-run` printed only
        // `[GATE after 3 consecutive failures]` on the Validate line.
        //
        // `!unconditional` suppresses it where it would be noise rather than
        // information: Ship and Supervise-mode Validate gate regardless of any
        // failure count, so naming a failure ceiling there tells the operator
        // nothing about where this run will stop.
        let phase_gate = if !unconditional
            && state
                .mode
                .should_gate(s, 0, mode::MAX_PHASE_VALIDATE_FAILURES)
        {
            format!(
                " [GATE at {} validate failures for this phase]",
                mode::MAX_PHASE_VALIDATE_FAILURES
            )
        } else {
            String::new()
        };
        // WR-01 (phase 20 review): annotate the stage matching `--until` so
        // the dry-run preview reflects that a real invocation would halt
        // here instead of always showing the full Define→Ship pipeline.
        let stop_marker = if state.stop_until == Some(s) {
            " [STOPS HERE — --until]"
        } else {
            ""
        };
        println!("  {s:<9} {command}{gate}{phase_gate}{stop_marker}");
        if let Some(next) = s.next() {
            let transition_hooks = hooks::hooks_for_transition(s, next);
            if !transition_hooks.is_empty() {
                println!("            ↳ hooks: {transition_hooks:?}");
            }
        }
        stage = s.next();
    }
    if let Some(until) = state.stop_until {
        println!("\nnote: --until {until} — this run will halt after {until} completes");
    }
    // D-12 (`28-CONTEXT.md`): the resolved combine of `--yes-ship` and a
    // `devflow.toml` `yes_ship` key, reported here so a dry run gives the
    // operator a stable, observable preview of whether the Ship gate is
    // pre-authorized — this line reports `state.yes_ship` only, it does not
    // itself distinguish flag- from config-sourced authorization (that
    // distinction is the separate never-silent notice `commands::start`
    // prints for the config-sourced case).
    println!(
        "\nship gate: {}",
        if state.yes_ship {
            "pre-authorized"
        } else {
            "not pre-authorized"
        }
    );
    println!("\nafter ship: {:?}", hooks::hooks_after_ship());
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pipeline_launch::advance;
    use crate::pipeline_outcomes::{
        ValidateOutcome, handle_infra_outcome, handle_validate_outcome,
    };
    use crate::test_support::*;
    use devflow_core::agent_result;
    use devflow_core::gates::GateResponse;
    use devflow_core::mode::Mode;
    use devflow_core::state::AgentKind;

    /// `advance()` over a Ship-stage success with an approved Ship gate must run
    /// the terminal `finish_workflow` path (after-ship hooks + gate cleanup +
    /// state cleared) — the only non-spawning branch of `advance`'s orchestration
    /// (11-VALIDATION.md 12f). The gate response is pre-seeded on disk so
    /// `run_gate`'s poll returns immediately instead of blocking.
    #[test]
    fn advance_ship_success_runs_finish_workflow() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);

        let phase = PhaseId::new(21);
        let branch = format!("feature/phase-{padded}", padded = phase.padded());
        let branch_created = devflow_core::test_support::git_command(root)
            .args(["branch", &branch, "develop"])
            .status()
            .unwrap()
            .success();
        assert!(branch_created);
        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Ship;
        workflow::save_state(&state).unwrap();

        // Seed a DEVFLOW_RESULT success marker so `evaluate_agent_result` resolves
        // at Layer 1 without needing the exit-code/commit-count fallback.
        std::fs::write(
            agent_result::stdout_path(root, phase),
            "DEVFLOW_RESULT: {\"status\":\"success\"}\n",
        )
        .unwrap();

        // Pre-write an approved Ship gate response so `run_gate` returns
        // `GateAction::Advance` immediately instead of polling.
        let response_path = Gates::response_path(root, phase, Stage::Ship);
        std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
        std::fs::write(
            &response_path,
            r#"{"approved":true,"note":null,"responded_by":"test"}"#,
        )
        .unwrap();

        advance(root, Some(phase)).unwrap();

        let err = workflow::load_state(root, phase).unwrap_err();
        assert!(matches!(err, workflow::WorkflowError::MissingState(_)));
        assert!(!Gates::gate_path(root, phase, Stage::Ship).exists());
        assert!(!Gates::response_path(root, phase, Stage::Ship).exists());
        assert!(!Gates::ack_path(root, phase, Stage::Ship).exists());
        assert!(!Gates::gate_path(root, phase, Stage::Validate).exists());
    }

    /// 23-06 Task 1 acceptance: a real Ship finalization emits the
    /// terminal-only `workflow_shipped` event, and
    /// `ship_evidence::collect` reads it as `shipped: true` — proving the
    /// event this module emits is exactly the one the oracle consumes.
    #[test]
    fn advance_ship_success_emits_workflow_shipped_and_ship_evidence_reports_shipped() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);

        let phase = PhaseId::new(23);
        let branch = format!("feature/phase-{padded}", padded = phase.padded());
        let branch_created = devflow_core::test_support::git_command(root)
            .args(["branch", &branch, "develop"])
            .status()
            .unwrap()
            .success();
        assert!(branch_created);
        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Ship;
        workflow::save_state(&state).unwrap();

        std::fs::write(
            agent_result::stdout_path(root, phase),
            "DEVFLOW_RESULT: {\"status\":\"success\"}\n",
        )
        .unwrap();

        let response_path = Gates::response_path(root, phase, Stage::Ship);
        std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
        std::fs::write(
            &response_path,
            r#"{"approved":true,"note":null,"responded_by":"test"}"#,
        )
        .unwrap();

        advance(root, Some(phase)).unwrap();

        assert!(
            devflow_core::events::has_event_for_phase(root, phase, "workflow_shipped"),
            "a real Ship finalization must emit the terminal-only workflow_shipped event"
        );
        let evidence = devflow_core::ship_evidence::collect(root, phase);
        assert!(
            evidence.shipped,
            "ship_evidence must read the just-emitted workflow_shipped event as shipped"
        );
        // The pre-existing invariant several other tests in this module
        // depend on: the phase's event stream still ends in
        // workflow_finished, unmodified by this task's additive emission.
        let last = devflow_core::events::last_event_for_phase(root, phase)
            .expect("events recorded for phase");
        assert_eq!(last["event"], "workflow_finished");
    }

    /// 23-06 Task 1's named blocker regression guard (BLOCKER 1,
    /// cross-AI-review-caught): a phase halted cleanly by `devflow start
    /// --until <stage>` runs `transition`'s `stop_until` clean-stop branch,
    /// which emits `workflow_finished` with a `"stopped_at"` reason and
    /// returns before any hook, stage assignment, or `launch_stage` call —
    /// `workflow_shipped` must never be emitted on this path, and
    /// `ship_evidence::collect(..).shipped` must read false even though
    /// `workflow_finished_seen` is true.
    #[test]
    fn until_stop_never_emits_workflow_shipped_and_ship_evidence_reports_not_shipped() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);

        let phase = PhaseId::new(24);
        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Plan;
        state.stop_until = Some(Stage::Plan);
        workflow::save_state(&state).unwrap();

        transition(root, &mut state, Stage::Code).unwrap();

        assert!(
            !devflow_core::events::has_event_for_phase(root, phase, "workflow_shipped"),
            "the --until clean-stop branch must never emit workflow_shipped"
        );
        let evidence = devflow_core::ship_evidence::collect(root, phase);
        assert!(
            !evidence.shipped,
            "a phase that only stopped after one stage must not read as shipped"
        );
        assert!(evidence.workflow_finished_seen);
        assert_eq!(evidence.finished_reason.as_deref(), Some("stopped_at"));
        assert!(devflow_core::ship_evidence::is_stopped_at(&evidence));
    }

    #[test]
    fn terminal_merge_failure_reopens_actionable_gate_and_never_reports_finished() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);
        let git = |args: &[&str]| {
            let output = devflow_core::test_support::git_command(root)
                .args(args)
                .output()
                .unwrap();
            assert!(output.status.success(), "git {args:?} failed");
        };
        git(&["checkout", "-q", "-b", "feature/phase-22"]);
        std::fs::write(root.join("conflict.txt"), "feature\n").unwrap();
        git(&["add", "conflict.txt"]);
        git(&["commit", "-q", "-m", "feature change"]);
        git(&["checkout", "-q", "develop"]);
        std::fs::write(root.join("conflict.txt"), "develop\n").unwrap();
        git(&["add", "conflict.txt"]);
        git(&["commit", "-q", "-m", "develop change"]);

        let mut state = State::new(
            PhaseId::new(22),
            AgentKind::Claude,
            Mode::Auto,
            root.to_path_buf(),
        );
        state.stage = Stage::Ship;
        workflow::save_state(&state).unwrap();

        let root_owned = root.to_path_buf();
        let handle = std::thread::spawn(move || {
            let mut state = workflow::load_state(&root_owned, PhaseId::new(22)).unwrap();
            finish_workflow(&root_owned, &mut state)
        });
        let gate_path = Gates::gate_path(root, PhaseId::new(22), Stage::Ship);
        for _ in 0..100 {
            if gate_path.exists() {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }

        assert!(
            gate_path.exists(),
            "finalization failure must reopen Ship gate"
        );
        assert!(
            workflow::load_state(root, PhaseId::new(22))
                .unwrap()
                .gate_pending
        );
        Gates::respond(
            root,
            PhaseId::new(22),
            Stage::Ship,
            &GateResponse {
                approved: false,
                note: Some("abort after merge conflict".into()),
                responded_by: Some("test".into()),
            },
        )
        .unwrap();
        handle.join().unwrap().unwrap();

        assert_ne!(
            events::last_event_for_phase(root, PhaseId::new(22))
                .and_then(|event| event["event"].as_str().map(str::to_owned))
                .as_deref(),
            Some("workflow_finished")
        );
        let tags = devflow_core::test_support::git_command(root)
            .arg("tag")
            .output()
            .unwrap();
        assert!(tags.stdout.is_empty());
    }

    /// 23-09 Task 2 (T-23-91's regression guard, cross-AI-review HIGH): with
    /// `state.yes_ship` set, a terminal-hook failure that reopens the
    /// finalization gate must NOT be auto-approved — the reopened gate is a
    /// different call site (passing `None`) than the routine Ship approval
    /// in `handle_ship_outcome` (passing `Some`), and this is the concrete
    /// proof. If a future refactor folds `state.yes_ship` into
    /// `run_gate_with_timeout`'s own body instead of threading it as a
    /// caller argument (see that parameter's doc comment), this test fails
    /// loudly: the reopened gate would then find an unexpected response and
    /// never block for a human.
    ///
    /// `VersionBump` is failed deterministically via `version::compute_version`
    /// itself refusing (D-10's `UnreachableBaseline`), rather than by
    /// pre-creating a tag name for `git.tag(&tag)` to collide with
    /// (`hooks.rs:286-287`).
    ///
    /// **Why not a tag-name collision (25-01 rewrite, D-07/D-08):** the prior
    /// (pre-25-01) algorithm derived MINOR from a raw, reachability-blind git
    /// tag *count*, so pre-creating one extra tag deterministically
    /// incremented that count by exactly one, and the fixture could predict
    /// and pre-create the exact resulting tag name. The 25-01 algorithm is
    /// no longer count-based: `compute_version` derives its baseline from
    /// the highest semver tag *reachable from HEAD*, and always bumps
    /// strictly past that baseline (D-10's no-bump-collapses-to-patch floor
    /// guarantees this even when nothing in the range warrants a bump — see
    /// `apply_bump`'s doc comment in `version.rs`). Consequently, pre-creating
    /// ANY tag reachable from HEAD makes it become the new baseline the next
    /// time `compute_version` runs, and the recomputed result is
    /// *unconditionally different* from (one patch past) whatever we just
    /// created — there is no reachable tag this fixture could create that
    /// `compute_version` would ever predict again, so a tag-name collision
    /// can no longer be constructed this way. (Confirmed empirically during
    /// this rewrite: pre-creating the tag `compute_version` returned before
    /// any tag existed did not collide — `VersionBump`'s own later call saw
    /// that tag as the new reachable baseline and computed the next patch
    /// past it instead, so `finish_workflow` shipped successfully instead of
    /// re-opening the gate this test exists to exercise.)
    ///
    /// Instead, this fixture tags an **unreachable** orphan commit
    /// (`git tag` sees it via [`devflow_core::version::highest_semver_tag`]'s
    /// reachability-blind repo-wide scan; `develop`'s own reachable baseline
    /// stays `None` because the tag is not an ancestor of `develop`) — D-10's
    /// refusal fires unconditionally in that state, independent of any
    /// version arithmetic, so `VersionBump` fails deterministically before it
    /// ever reaches its `git.tag(&tag)` call. `Merge` still runs first in
    /// `hooks_after_ship` and succeeds — the feature branch is created
    /// identical to `develop`, so `is_merged_into_develop` is immediately
    /// true and `merge_feature` takes its already-merged no-op success
    /// path — exactly the "Merge has already succeeded" state the reopened
    /// gate exists for (no rollback policy, `hooks.rs` doc comment on
    /// `merge_feature`).
    #[test]
    fn finalization_retry_gate_never_auto_approves_even_with_yes_ship_set() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);

        // Tag an orphan commit unreachable from `develop` — this makes
        // `version::compute_version` refuse unconditionally (D-10's
        // `UnreachableBaseline`) once `VersionBump` calls it, regardless of
        // any subsequent state. `develop` is where `init_repo` leaves HEAD
        // checked out, and is restored as the checkout before continuing.
        let git_cmd = |args: &[&str]| {
            assert!(
                devflow_core::test_support::git_command(root)
                    .args(args)
                    .output()
                    .unwrap()
                    .status
                    .success(),
                "git {args:?} failed"
            );
        };
        git_cmd(&["checkout", "--orphan", "unreachable-release"]);
        git_cmd(&["commit", "--allow-empty", "-q", "-m", "chore: orphan"]);
        git_cmd(&["tag", "v9.9.9"]);
        git_cmd(&["checkout", "develop"]);
        assert!(
            devflow_core::version::compute_version(root).is_err(),
            "the orphan tag must make compute_version refuse (D-10) before \
             VersionBump ever gets to its own git.tag(&tag) call"
        );

        let phase = PhaseId::new(60);
        let branch = format!("feature/phase-{padded}", padded = phase.padded());
        let branch_created = devflow_core::test_support::git_command(root)
            .args(["branch", &branch, "develop"])
            .status()
            .unwrap()
            .success();
        assert!(branch_created);

        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Ship;
        state.yes_ship = true;
        workflow::save_state(&state).unwrap();

        let root_owned = root.to_path_buf();
        let handle = std::thread::spawn(move || {
            let mut state = workflow::load_state(&root_owned, phase).unwrap();
            finish_workflow(&root_owned, &mut state)
        });

        let gate_path = Gates::gate_path(root, phase, Stage::Ship);
        for _ in 0..150 {
            if gate_path.exists() {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        assert!(
            gate_path.exists(),
            "a terminal-hook failure must reopen the Ship gate even with yes_ship set"
        );

        // Give any (incorrect) auto-write a moment to have happened if it
        // were ever going to.
        std::thread::sleep(std::time::Duration::from_millis(50));
        let response_path = Gates::response_path(root, phase, Stage::Ship);
        assert!(
            !response_path.exists(),
            "yes_ship must NEVER auto-approve the reopened finalization-retry gate \
             (T-23-91) — a failing finalization must wait for a human"
        );
        assert!(
            workflow::load_state(root, phase).unwrap().gate_pending,
            "the reopened gate must be recorded as pending, awaiting a human"
        );
        assert!(
            devflow_core::events::has_event_for_phase(root, phase, "merge_result"),
            "Merge must have succeeded before VersionBump failed — the reopened gate must be \
             exercised in the state it actually occurs in"
        );

        // Unblock the poll (reject, not approve — this test's subject is
        // that nothing auto-approves it) so the spawned thread can finish.
        Gates::respond(
            root,
            phase,
            Stage::Ship,
            &GateResponse {
                approved: false,
                note: Some("abort: test cleanup".into()),
                responded_by: Some("test".into()),
            },
        )
        .unwrap();
        handle.join().unwrap().unwrap();
    }

    /// 13-DEFERRED-CR-03 acceptance: two phases advancing their Ship stages
    /// CONCURRENTLY must each finish their own stage machine — per-phase
    /// state files prevent cross-phase clobbering, and the coarse checkout
    /// lock serializes both `finish_workflow`s' git operations on the shared
    /// primary checkout. Gate responses are pre-seeded so neither advance
    /// blocks polling on its *first* Ship gate.
    ///
    /// 17-09 gap closure (GAP-2): both phases compute their next version from
    /// the same starting git state, and on some runs genuinely race to
    /// create the same version tag — confirmed directly during this plan's
    /// RED phase via temporary debug instrumentation, which caught both
    /// threads inside `version_bump` with the identical computed version
    /// (`2.0.1`) within ~1.8ms of each other, and the loser's `git tag`
    /// failing with git's own "reference already exists". That failure
    /// reopens the loser's Ship gate for human review (`finish_workflow`'s
    /// retry loop) — but only ONE gate response was ever pre-written per
    /// phase (consumed by its first gate open), so the reopened gate has
    /// nothing to consume. Unbounded, `Gates::poll_response` then polls the
    /// 7-day production default (`DEVFLOW_GATE_TIMEOUT_SECS`) with no
    /// response ever arriving — that is the wedge this plan closes.
    ///
    /// The binding constraint is "never hangs," not "always both succeed."
    /// This test does not try to make the race loser also succeed (that
    /// would require re-answering a gate reactively and still not rule out
    /// a second, equally rare collision) — instead it bounds the reopened
    /// gate's poll to a few seconds via `DEVFLOW_GATE_TIMEOUT_SECS`
    /// (overridden ONLY for this test's poll, under the established
    /// `ENV_MUTEX` guard — the 7-day production default is never touched)
    /// and asserts the loser's documented behavior: a bounded timeout error,
    /// state left intact (not cleared), and an actionable Ship gate still on
    /// disk awaiting a human. The common case (no collision) still asserts
    /// both phases finish independently, exactly as before.
    #[test]
    fn concurrent_ship_advances_finish_both_phases_independently() {
        let _guard = env_lock();
        let original_gate_timeout = std::env::var_os("DEVFLOW_GATE_TIMEOUT_SECS");
        // SAFETY: serialized under ENV_MUTEX. Bounds a reopened Ship gate's
        // poll to a few seconds instead of the 7-day production default.
        // Every OTHER test that reaches `run_gate` pre-writes its response
        // before calling in, so `poll_response` finds it on the very first
        // read regardless of this value — only a *reopened*, unanswered
        // gate (this test's race-loser path) ever actually waits it out.
        unsafe {
            std::env::set_var("DEVFLOW_GATE_TIMEOUT_SECS", "2");
        }

        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);

        let phases = [PhaseId::new(31), PhaseId::new(32)];
        for &phase in &phases {
            let branch = format!("feature/phase-{padded}", padded = phase.padded());
            let branch_created = devflow_core::test_support::git_command(root)
                .args(["branch", &branch, "develop"])
                .status()
                .unwrap()
                .success();
            assert!(branch_created);
            let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
            state.stage = Stage::Ship;
            workflow::save_state(&state).unwrap();
            std::fs::write(
                agent_result::stdout_path(root, phase),
                "DEVFLOW_RESULT: {\"status\":\"success\"}\n",
            )
            .unwrap();
            let response_path = Gates::response_path(root, phase, Stage::Ship);
            std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
            std::fs::write(
                &response_path,
                r#"{"approved":true,"note":null,"responded_by":"test"}"#,
            )
            .unwrap();
        }

        let results: Vec<(PhaseId, Result<(), CliError>)> = std::thread::scope(|scope| {
            let handles: Vec<_> = phases
                .iter()
                .map(|&phase| (phase, scope.spawn(move || advance(root, Some(phase)))))
                .collect();
            handles
                .into_iter()
                .map(|(phase, handle)| (phase, handle.join().expect("advance thread")))
                .collect()
        });

        // SAFETY: still serialized under ENV_MUTEX from above.
        unsafe {
            match &original_gate_timeout {
                Some(value) => std::env::set_var("DEVFLOW_GATE_TIMEOUT_SECS", value),
                None => std::env::remove_var("DEVFLOW_GATE_TIMEOUT_SECS"),
            }
        }

        let succeeded = results.iter().filter(|(_, r)| r.is_ok()).count();
        assert!(
            succeeded == 1 || succeeded == 2,
            "at least one phase must finish independently of the other; got {succeeded}/2 successes"
        );

        for (phase, result) in &results {
            match result {
                Ok(()) => {
                    assert!(
                        matches!(
                            workflow::load_state(root, *phase),
                            Err(workflow::WorkflowError::MissingState(_))
                        ),
                        "phase {phase} must be finished (state cleared)"
                    );
                    assert!(!Gates::gate_path(root, *phase, Stage::Ship).exists());
                    let last = devflow_core::events::last_event_for_phase(root, *phase)
                        .expect("events recorded for phase");
                    assert_eq!(
                        last["event"], "workflow_finished",
                        "phase {phase}'s own event stream must end in workflow_finished"
                    );
                }
                Err(err) => {
                    // The documented loser behavior (GAP-2): a version-tag
                    // race lost by VersionBump reopens the Ship gate for a
                    // human; with no second response pre-written, the
                    // bounded poll above times out rather than hanging.
                    assert!(
                        err.to_string().contains("timed out"),
                        "phase {phase}'s only non-success outcome must be a bounded gate \
                         timeout, not some other failure: {err}"
                    );
                    let state = workflow::load_state(root, *phase)
                        .expect("a timed-out gate leaves state intact, not cleared");
                    assert!(
                        state.gate_pending,
                        "phase {phase} must leave an actionable, still-open gate for a human"
                    );
                    assert!(
                        Gates::gate_path(root, *phase, Stage::Ship).exists(),
                        "phase {phase}'s reopened Ship gate file must remain on disk"
                    );
                }
            }
        }
    }

    /// Regression test for CR-01: `abort()` must clean up the gate's
    /// response/ack files for the stage the gate actually fired on. Without
    /// that cleanup, a later gate for the same phase+stage would find the
    /// old, already-consumed response still on disk and `poll_response`
    /// would resolve from it instantly instead of waiting for a fresh human
    /// decision.
    ///
    /// 33-04: the seeded 999.66 forward-progress baseline, the
    /// neutralized PATH and the two branch-pinning assertions below exist
    /// because this test once silently left its intended path — 999.66's
    /// reset-vs-accumulate change reset the directly-seeded streak to 1, so no
    /// gate fired and the test fell through to a real agent launch while every
    /// gate-file assertion it owned still passed (`prepare_loop_back_to_code`
    /// calls the same `Gates::cleanup` that `abort()` does).
    #[test]
    fn abort_cleans_up_gate_files_so_a_later_gate_does_not_reuse_stale_response() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();

        let phase = PhaseId::new(23);
        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Validate;
        state.consecutive_failures = mode::MAX_CONSECUTIVE_FAILURES - 1;
        // 999.66 (33-03) made the counter reset-vs-accumulate against a
        // persisted baseline. This test seeds the streak directly rather than
        // driving it through repeated `handle_validate_outcome` calls, so it
        // must also seed the baseline — a `None` baseline reads as
        // "first-ever failure" and resets the streak to 1, which drops this
        // test out of the gated path it asserts against and into
        // `loop_back_to_code` -> `launch_stage` (a real agent spawn).
        state.last_validate_failure_commit_count = Some(0);
        workflow::save_state(&state).unwrap();

        // Pre-write a rejected response whose note says "abort" so
        // `GateAction::from_response` resolves to `Abort`.
        let response_path = Gates::response_path(root, phase, Stage::Validate);
        std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
        std::fs::write(
            &response_path,
            r#"{"approved":false,"note":"abort: requirements changed","responded_by":"test"}"#,
        )
        .unwrap();

        {
            let _guard = env_lock();

            let neutral_path_dir = agent_free_git_only_path_dir();
            let original_path = std::env::var_os("PATH");
            // SAFETY: serialized under ENV_MUTEX.
            unsafe {
                std::env::set_var("PATH", neutral_path_dir.path());
            }

            handle_validate_outcome(root, &mut state, ValidateOutcome::Failed).unwrap();

            // SAFETY: still serialized under ENV_MUTEX from above.
            unsafe {
                match &original_path {
                    Some(path) => std::env::set_var("PATH", path),
                    None => std::env::remove_var("PATH"),
                }
            }
        }

        assert!(
            state.consecutive_failures >= mode::MAX_CONSECUTIVE_FAILURES,
            "gate threshold must have been reached (got {}) — a reset means the gate never fired",
            state.consecutive_failures
        );
        assert_eq!(
            state.stage,
            Stage::Validate,
            "gate must have fired and aborted — Stage::Code means it silently looped back and tried to launch an agent"
        );

        // The gate, response, and ack files for the stage the gate fired on
        // (Validate) must all be gone after the Abort path runs.
        assert!(!Gates::gate_path(root, phase, Stage::Validate).exists());
        assert!(
            !Gates::response_path(root, phase, Stage::Validate).exists(),
            "stale response file must not survive an aborted gate"
        );
        assert!(!Gates::ack_path(root, phase, Stage::Validate).exists());

        // Simulate the phase reaching the same gate again later (e.g. after
        // a restart) — write a fresh request but no new response. If cleanup
        // had not happened, `poll_response` would instantly return the old,
        // already-consumed response instead of blocking for a fresh human
        // decision.
        Gates::write_gate(root, phase, Stage::Validate, "re-fired gate").unwrap();
        let started = std::time::Instant::now();
        let got = Gates::poll_response(root, phase, Stage::Validate, 1);
        assert!(
            got.is_none(),
            "poll_response must not instantly resolve from a stale response after cleanup"
        );
        assert!(started.elapsed() >= std::time::Duration::from_secs(1));
    }

    /// CR-01 regression (17-06 gap closure): `transition()` resets
    /// `infra_failures` to 0 alongside `consecutive_failures` — both in the
    /// in-memory `State` and the persisted `state.json` — and a subsequent
    /// infra fault after a clean transition starts counting from 1, not the
    /// pre-transition count. PATH is neutralized under `ENV_MUTEX` (pointed
    /// at a directory containing ONLY a `git` symlink, so
    /// `agent_binary_available`'s PATH scan has zero possible matches) before
    /// calling `transition()`, because this host genuinely has
    /// `claude`/`codex`/`opencode` on PATH — without neutralizing it,
    /// `transition()`'s downstream `launch_stage` would try to actually spawn
    /// a real agent CLI subprocess, which this test must never do. The
    /// resulting `Err` from `ensure_agent_binary` is expected and ignored:
    /// the counter reset happens earlier in `transition()` and is unaffected
    /// by that downstream failure.
    ///
    /// 19i: PATH must NOT be pointed at an empty directory. `set_var`
    /// mutates the whole process's environment, and Rust's default test
    /// runner executes tests in parallel threads within that one process —
    /// an empty PATH here previously made every OTHER concurrently running,
    /// unguarded git-spawning test fail with `Os { NotFound }` (confirmed
    /// live: both duplicate CI runs for the same commit hit this race).
    /// `agent_free_git_only_path_dir` keeps `git` resolvable for every other
    /// thread while still hiding agent CLIs from this one.
    #[test]
    fn transition_resets_infra_failures() {
        let _guard = env_lock();

        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let phase = PhaseId::new(80);
        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Code;
        state.infra_failures = mode::MAX_INFRA_FAILURES - 1;
        workflow::save_state(&state).unwrap();

        let neutral_path_dir = agent_free_git_only_path_dir();
        let original_path = std::env::var_os("PATH");
        // SAFETY: serialized under ENV_MUTEX.
        unsafe {
            std::env::set_var("PATH", neutral_path_dir.path());
        }

        let _ = transition(root, &mut state, Stage::Validate);

        // SAFETY: still serialized under ENV_MUTEX from above.
        unsafe {
            match &original_path {
                Some(path) => std::env::set_var("PATH", path),
                None => std::env::remove_var("PATH"),
            }
        }

        assert_eq!(
            state.infra_failures, 0,
            "transition() must reset infra_failures in-memory, not just consecutive_failures"
        );
        let reloaded = workflow::load_state(root, phase).unwrap();
        assert_eq!(
            reloaded.infra_failures, 0,
            "transition() must persist the infra_failures reset to state.json"
        );

        // A fresh infra fault after the clean transition starts counting
        // from 1, not resuming the pre-transition MAX_INFRA_FAILURES - 1
        // count toward a false premature abort.
        let response_path = Gates::response_path(root, phase, Stage::Validate);
        std::fs::create_dir_all(response_path.parent().unwrap()).unwrap();
        std::fs::write(
            &response_path,
            r#"{"approved":false,"note":"abort: test cleanup","responded_by":"test"}"#,
        )
        .unwrap();

        handle_infra_outcome(root, &mut state, Stage::Validate, Some("killed".into())).unwrap();

        assert_eq!(state.infra_failures, 1);
    }

    /// 18d idempotency edge: a repeated Code→Validate transition leaves
    /// `consecutive_failures` unchanged rather than zeroing it. `state.stage`
    /// is reset to `Code` before each call so both calls exercise the exact
    /// hop under test.
    #[test]
    fn repeated_code_to_validate_transition_is_idempotent_on_the_counter() {
        let _guard = env_lock();

        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let phase = PhaseId::new(83);
        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Code;
        state.consecutive_failures = 2;
        workflow::save_state(&state).unwrap();

        let neutral_path_dir = agent_free_git_only_path_dir();
        let original_path = std::env::var_os("PATH");
        // SAFETY: serialized under ENV_MUTEX.
        unsafe {
            std::env::set_var("PATH", neutral_path_dir.path());
        }

        let _ = transition(root, &mut state, Stage::Validate);
        state.stage = Stage::Code;
        let _ = transition(root, &mut state, Stage::Validate);

        // SAFETY: still serialized under ENV_MUTEX from above.
        unsafe {
            match &original_path {
                Some(path) => std::env::set_var("PATH", path),
                None => std::env::remove_var("PATH"),
            }
        }

        assert_eq!(state.consecutive_failures, 2);
    }

    /// 20e Task 1: the tracer end-to-end case — a phase parked at
    /// `Stage::Ship` with a REQUEST + RESPONSE (approved) written via
    /// `Gates::respond`, no ack, and no live process polling. `ship_override`
    /// must reach the SAME terminal state `finish_workflow` produces: state
    /// cleared, gate/response/ack files gone, `workflow_finished` emitted.
    #[test]
    fn ship_override_advances_via_written_response() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);

        let phase = PhaseId::new(90);
        let branch = format!("feature/phase-{padded}", padded = phase.padded());
        let branch_created = devflow_core::test_support::git_command(root)
            .args(["branch", &branch, "develop"])
            .status()
            .unwrap()
            .success();
        assert!(branch_created);

        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Ship;
        workflow::save_state(&state).unwrap();

        Gates::write_gate(root, phase, Stage::Ship, "Ship complete — approve merge?").unwrap();
        Gates::respond(
            root,
            phase,
            Stage::Ship,
            &GateResponse {
                approved: true,
                note: None,
                responded_by: Some("test".into()),
            },
        )
        .unwrap();

        ship_override(root, phase, false).unwrap();

        let err = workflow::load_state(root, phase).unwrap_err();
        assert!(matches!(err, workflow::WorkflowError::MissingState(_)));
        assert!(!Gates::gate_path(root, phase, Stage::Ship).exists());
        assert!(!Gates::response_path(root, phase, Stage::Ship).exists());
        assert!(!Gates::ack_path(root, phase, Stage::Ship).exists());
        let last = devflow_core::events::last_event_for_phase(root, phase)
            .expect("events recorded for phase");
        assert_eq!(last["event"], "workflow_finished");
    }

    /// WR-02 (phase 20 review): `ship_override` runs in the FOREGROUND CLI,
    /// unlike every other `finish_workflow` caller (which runs inside a
    /// detached monitor). A terminal-hook failure (merge conflict) reopens
    /// the Ship gate; with no response ever written for the REOPENED gate,
    /// this must fail fast within `DEVFLOW_FOREGROUND_GATE_TIMEOUT_SECS`
    /// (bounded here to a couple seconds) rather than block the caller's
    /// shell for `DEVFLOW_GATE_TIMEOUT_SECS`' multi-day production default —
    /// which is left untouched, proving the two timeouts are genuinely
    /// independent knobs.
    #[test]
    fn ship_override_bounds_foreground_wait_on_terminal_hook_failure() {
        let _guard = env_lock();
        let original_foreground_timeout = std::env::var_os("DEVFLOW_FOREGROUND_GATE_TIMEOUT_SECS");
        // SAFETY: serialized under ENV_MUTEX. Bounds ONLY the foreground
        // knob — DEVFLOW_GATE_TIMEOUT_SECS (the background default) is
        // never touched by this test, so a regression that made
        // `ship_override` fall back to the multi-day default would hang
        // this test instead of silently passing.
        unsafe {
            std::env::set_var("DEVFLOW_FOREGROUND_GATE_TIMEOUT_SECS", "2");
        }

        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        init_repo(root);
        let git = |args: &[&str]| {
            let output = devflow_core::test_support::git_command(root)
                .args(args)
                .output()
                .unwrap();
            assert!(output.status.success(), "git {args:?} failed");
        };
        let phase = PhaseId::new(96);
        let branch = format!("feature/phase-{padded}", padded = phase.padded());
        git(&["checkout", "-q", "-b", &branch]);
        std::fs::write(root.join("conflict.txt"), "feature\n").unwrap();
        git(&["add", "conflict.txt"]);
        git(&["commit", "-q", "-m", "feature change"]);
        git(&["checkout", "-q", "develop"]);
        std::fs::write(root.join("conflict.txt"), "develop\n").unwrap();
        git(&["add", "conflict.txt"]);
        git(&["commit", "-q", "-m", "develop change"]);

        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Ship;
        workflow::save_state(&state).unwrap();

        Gates::write_gate(root, phase, Stage::Ship, "Ship complete — approve merge?").unwrap();
        Gates::respond(
            root,
            phase,
            Stage::Ship,
            &GateResponse {
                approved: true,
                note: None,
                responded_by: Some("test".into()),
            },
        )
        .unwrap();

        let started = std::time::Instant::now();
        let result = ship_override(root, phase, false);
        let elapsed = started.elapsed();

        // SAFETY: still serialized under ENV_MUTEX from above.
        unsafe {
            match &original_foreground_timeout {
                Some(value) => {
                    std::env::set_var("DEVFLOW_FOREGROUND_GATE_TIMEOUT_SECS", value);
                }
                None => std::env::remove_var("DEVFLOW_FOREGROUND_GATE_TIMEOUT_SECS"),
            }
        }

        assert!(
            result.is_err(),
            "an unresolved reopened Ship gate must fail closed, not silently advance"
        );
        assert!(
            elapsed < std::time::Duration::from_secs(30),
            "the foreground wait must be bounded by DEVFLOW_FOREGROUND_GATE_TIMEOUT_SECS, \
             not gate_timeout_secs' multi-day default — took {elapsed:?}"
        );
        assert!(
            Gates::gate_path(root, phase, Stage::Ship).exists(),
            "the merge failure must reopen an actionable Ship gate for a human, not silently \
             drop the phase"
        );
        assert!(
            workflow::load_state(root, phase).is_ok(),
            "state must NOT be cleared — finish_workflow_with_gate_timeout must fail before \
             reaching workflow::clear_state"
        );
    }

    /// 20e Task 3: an `Abort`-routing Ship response (a rejection whose note
    /// says "abort") must reach the SAME shared `abort` helper the live
    /// `handle_ship_outcome` path uses — no special-cased branch inside
    /// `ship_override`. Asserts the shared path's own effects: state
    /// cleared, gate files cleaned up, `workflow_aborted` emitted.
    #[test]
    fn ship_override_abort_routes_through_abort() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let phase = PhaseId::new(95);

        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Ship;
        workflow::save_state(&state).unwrap();

        Gates::write_gate(root, phase, Stage::Ship, "Ship complete — approve merge?").unwrap();
        Gates::respond(
            root,
            phase,
            Stage::Ship,
            &GateResponse {
                approved: false,
                note: Some("abort: found a blocking regression during manual review".into()),
                responded_by: Some("test".into()),
            },
        )
        .unwrap();

        ship_override(root, phase, false).unwrap();

        let err = workflow::load_state(root, phase).unwrap_err();
        assert!(matches!(err, workflow::WorkflowError::MissingState(_)));
        assert!(!Gates::gate_path(root, phase, Stage::Ship).exists());
        assert!(!Gates::response_path(root, phase, Stage::Ship).exists());
        assert!(!Gates::ack_path(root, phase, Stage::Ship).exists());
        let last = devflow_core::events::last_event_for_phase(root, phase)
            .expect("events recorded for phase");
        assert_eq!(last["event"], "workflow_aborted");
    }

    /// 20e Task 2 (D-02, EoP regression): `--force` must never let
    /// `ship_override` reach `finish_workflow` from a non-Ship stage. Checked
    /// for every earlier `Stage` with `--force` both true and false.
    #[test]
    fn ship_override_refuses_when_not_at_ship_stage() {
        for stage in [Stage::Define, Stage::Plan, Stage::Code, Stage::Validate] {
            for force in [true, false] {
                let dir = tempfile::tempdir().unwrap();
                let root = dir.path();
                let phase = PhaseId::new(91);

                let mut state =
                    State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
                state.stage = stage;
                workflow::save_state(&state).unwrap();

                let err = ship_override(root, phase, force).unwrap_err();
                let msg = err.to_string();
                assert!(
                    msg.contains(&stage.to_string()),
                    "error for stage {stage} (force={force}) must name the stage: {msg}"
                );

                // finish_workflow was never reached: state is untouched, not cleared.
                let reloaded = workflow::load_state(root, phase)
                    .expect("state must survive a stage-mismatch refusal, not be cleared");
                assert_eq!(reloaded.stage, stage);
            }
        }
    }

    /// 20e Task 2 (edge-probe 20e/empty): `state.stage == Stage::Ship` but no
    /// Ship gate request/response exists on disk — a dead monitor never
    /// wrote or received one. Must fail-closed with a clear error, both with
    /// `--force` true and false.
    #[test]
    fn ship_override_refuses_when_no_response_written() {
        for force in [true, false] {
            let dir = tempfile::tempdir().unwrap();
            let root = dir.path();
            let phase = PhaseId::new(92);

            let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
            state.stage = Stage::Ship;
            workflow::save_state(&state).unwrap();

            // Neither request nor response written yet.
            let err = ship_override(root, phase, force).unwrap_err();
            assert!(
                err.to_string().contains("no Ship gate response"),
                "force={force}: {err}"
            );

            // A request written but no response yet is the same fail-closed case.
            Gates::write_gate(root, phase, Stage::Ship, "ctx").unwrap();
            let err = ship_override(root, phase, force).unwrap_err();
            assert!(
                err.to_string().contains("no Ship gate response"),
                "force={force}: {err}"
            );
            let _ = Gates::cleanup(root, phase, Stage::Ship);
        }
    }

    /// 20e Task 2 (review: Codex HIGH — lock race): a contended per-phase
    /// lock refuses fail-closed, naming the holder pid, with `--force` both
    /// true and false — `ship_override` must never race a live holder's
    /// `poll_response`.
    #[test]
    fn ship_override_refuses_when_lock_contended() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        let phase = PhaseId::new(93);

        let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
        state.stage = Stage::Ship;
        workflow::save_state(&state).unwrap();

        let _held = lock::acquire(root, phase).expect("hold the per-phase lock");
        let this_pid = std::process::id().to_string();

        for force in [true, false] {
            let err = ship_override(root, phase, force).unwrap_err();
            let msg = err.to_string();
            assert!(msg.contains(&this_pid), "force={force}: {msg}");
        }
    }

    /// 20e Task 2 (review: Hermes ack-race): a Ship response that already
    /// has an ack file alongside it was already consumed by a (now-dead)
    /// monitor — `ship_override` refuses and directs to `devflow doctor`
    /// rather than re-running terminal hooks, with `--force` both true and
    /// false. `finish_workflow` must never be reached: state stays intact.
    #[test]
    fn ship_override_refuses_when_response_already_acked() {
        for force in [true, false] {
            let dir = tempfile::tempdir().unwrap();
            let root = dir.path();
            let phase = PhaseId::new(94);

            let mut state = State::new(phase, AgentKind::Claude, Mode::Auto, root.to_path_buf());
            state.stage = Stage::Ship;
            workflow::save_state(&state).unwrap();

            Gates::write_gate(root, phase, Stage::Ship, "ctx").unwrap();
            Gates::respond(
                root,
                phase,
                Stage::Ship,
                &GateResponse {
                    approved: true,
                    note: None,
                    responded_by: Some("test".into()),
                },
            )
            .unwrap();
            Gates::ack(root, phase, Stage::Ship).unwrap();

            let err = ship_override(root, phase, force).unwrap_err();
            assert!(
                err.to_string().contains("devflow doctor"),
                "force={force}: {err}"
            );

            let reloaded = workflow::load_state(root, phase)
                .expect("an already-acked refusal must never clear state");
            assert_eq!(reloaded.stage, Stage::Ship);
        }
    }

    /// 18d concurrency edge: two concurrently-active phases' `consecutive_failures`
    /// counters are independent — a Code→Validate hop on one phase must not
    /// reset a sibling phase's counter.
    #[test]
    fn consecutive_failures_are_independent_across_phases() {
        let _guard = env_lock();

        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();

        let mut state_a = State::new(
            PhaseId::new(84),
            AgentKind::Claude,
            Mode::Auto,
            root.to_path_buf(),
        );
        state_a.stage = Stage::Code;
        state_a.consecutive_failures = 1;
        workflow::save_state(&state_a).unwrap();

        let mut state_b = State::new(
            PhaseId::new(85),
            AgentKind::Claude,
            Mode::Auto,
            root.to_path_buf(),
        );
        state_b.stage = Stage::Code;
        state_b.consecutive_failures = 2;
        workflow::save_state(&state_b).unwrap();

        let neutral_path_dir = agent_free_git_only_path_dir();
        let original_path = std::env::var_os("PATH");
        // SAFETY: serialized under ENV_MUTEX.
        unsafe {
            std::env::set_var("PATH", neutral_path_dir.path());
        }

        let _ = transition(root, &mut state_a, Stage::Validate);

        // SAFETY: still serialized under ENV_MUTEX from above.
        unsafe {
            match &original_path {
                Some(path) => std::env::set_var("PATH", path),
                None => std::env::remove_var("PATH"),
            }
        }

        let reloaded_a = workflow::load_state(root, PhaseId::new(84)).unwrap();
        let reloaded_b = workflow::load_state(root, PhaseId::new(85)).unwrap();

        assert_eq!(
            reloaded_a.consecutive_failures, 1,
            "the Code->Validate hop must not reset consecutive_failures"
        );
        assert_eq!(
            reloaded_b.consecutive_failures, 2,
            "an untouched sibling phase's counter must be unaffected"
        );
    }
}