cflx 0.6.322

//! Shared state management for orchestration operations.
//!
//! Provides a unified state structure that tracks orchestration progress
//! across every frontend (CLI `run`, TUI, remote control).
//!
//! ## Integration Status
//!
//! - **CLI Orchestrator** (`src/orchestrator.rs`): Fully integrated. The orchestrator
//!   maintains a `shared_state: Arc<RwLock<OrchestratorState>>` instance and updates it
//!   via `apply_execution_event` when processing changes (ProcessingStarted, ApplyStarted,
//!   ApplyCompleted, ChangeArchived). The shared state is wrapped in Arc<RwLock<>> to enable
//!   sharing with TUI and Web monitoring.
//!
//! - **TUI** (`src/tui/state/mod.rs`): Integrated via optional reference. TUI AppState has
//!   a `shared_orchestrator_state` field that can be set via `set_shared_state()`. TUI can
//!   query this for pending/archived status, apply counts, and current change tracking while
//!   maintaining its own UI-specific state for rendering and interaction.
//!
//! - **Web** (`src/web/state.rs`): Integrated via optional reference. WebState has a
//!   `shared_orchestrator_state` field set via `set_shared_state()` (called automatically
//!   by `Orchestrator::set_web_state()`). When generating `OrchestratorStateSnapshot` via
//!   `from_changes_with_shared_state()`, WebState queries shared state to enrich change
//!   metadata with apply counts, pending/archived status, and iteration numbers.
//!
//! ## Usage
//!
//! The shared state provides a single source of truth for tracking:
//! - Pending, completed, and archived changes
//! - Apply counts per change
//! - Current change being processed
//! - Iteration counters and limits
//!
//! ### Integration Pattern
//!
//! 1. **Orchestrator creates and owns shared state:**
//!    ```rust,ignore
//!    let shared_state = Arc::new(RwLock::new(OrchestratorState::new(changes, max_iters)));
//!    ```
//!
//! 2. **Orchestrator updates state via events:**
//!    ```rust,ignore
//!    shared_state.write().await.apply_execution_event(&event);
//!    ```
//!
//! 3. **TUI/Web receive shared state reference:**
//!    ```rust,ignore
//!    app_state.set_shared_state(shared_state.clone());
//!    web_state.set_shared_state(shared_state.clone()).await;
//!    ```
//!
//! 4. **TUI/Web query shared state when needed:**
//!    ```rust,ignore
//!    if let Some(shared) = &app_state.shared_orchestrator_state {
//!        let guard = shared.read().await;
//!        let apply_count = guard.apply_count(change_id);
//!        let is_pending = guard.is_pending(change_id);
//!    }
//!    ```
use std::collections::{HashMap, HashSet};

use crate::error_history::{CircuitBreakerConfig, ErrorHistory};

// ============================================================================
// ChangeRuntimeState types (Phase 1 – reducer-owned state)
// ============================================================================

/// Intent to include or exclude a change from the execution queue.
#[derive(Debug, Clone, PartialEq, Default)]
pub enum QueueIntent {
    /// Not requested to be queued.
    #[default]
    NotQueued,
    /// Requested to be queued for execution.
    Queued,
}

/// Active execution stage for a change.
#[derive(Debug, Clone, PartialEq, Default)]
pub enum ActivityState {
    /// No active execution.
    #[default]
    Idle,
    /// Admitted to an execution slot and preparing its managed workspace.
    ///
    /// Covers force-recreate cleanup, worktree creation/recreation, `.wt/setup`,
    /// and workspace inspection — everything between slot admission and the
    /// first operation agent. It never claims that an agent has started, and it
    /// is discarded on restart: the next action is recomputed from workspace
    /// files, Git state, and base-tree comparison.
    Preparing,
    /// Currently applying.
    Applying,
    /// Currently running acceptance checks.
    Accepting,
    /// Currently running dedicated rejection review.
    Rejecting,
    /// Currently archiving.
    Archiving,
    /// Currently executing a merge resolve.
    Resolving,
}

/// Reason a change is blocked waiting for an external condition.
#[derive(Debug, Clone, PartialEq, Default)]
pub enum WaitState {
    /// Not waiting.
    #[default]
    None,
    /// Waiting for a merge to be attempted (parallel only).
    MergeWait,
    /// Waiting for a resolve sub-task to start (queued resolve intent).
    ResolveWait,
    /// Waiting for rejection review to start once the base-mutating lane is free.
    RejectWait,
    /// Waiting because a dependency has not yet completed.
    DependencyBlocked,
    /// Waiting because the orchestrator validated a non-repository prerequisite
    /// reported by Apply or Acceptance.
    ///
    /// Displays as `blocked` alongside [`WaitState::DependencyBlocked`], but
    /// keeps its own blocker kind so the two waits stay distinguishable and the
    /// external wait never becomes a synthetic proposal dependency edge.
    ExternalBlocked,
    /// Waiting because apply/rejecting reported a resumable hold,
    /// including acceptance gate observations before follow-up routing.
    Stalled,
}

/// Machine-readable reason a change is `blocked`.
///
/// Mirrors [`crate::runtime::proposal::BlockerKind`] on the reducer that owns
/// operator-facing display, so surfaces never re-derive the kind from prose,
/// filenames, or task text.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum BlockerKind {
    /// The change is not blocked.
    #[default]
    None,
    /// Waiting on an unarchived proposal dependency.
    Dependency,
    /// Waiting on a validated non-repository prerequisite.
    External,
}

impl BlockerKind {
    /// Stable token exposed to TUI, WebSocket/API, and the dashboard.
    pub fn as_str(self) -> Option<&'static str> {
        match self {
            Self::None => None,
            Self::Dependency => Some("dependency"),
            Self::External => Some("external"),
        }
    }
}

/// Operator-facing one-line explanation of a dependency wait.
fn dependency_blocker_detail(dependency_ids: &[String]) -> String {
    if dependency_ids.is_empty() {
        return "waiting for unresolved dependencies".to_string();
    }
    format!("waiting for dependencies: {}", dependency_ids.join(", "))
}

/// Verifiable condition that clears a dependency wait.
fn dependency_unblock_condition(dependency_ids: &[String]) -> String {
    if dependency_ids.is_empty() {
        return "every declared dependency is integrated into the effective dependency base"
            .to_string();
    }
    format!(
        "dependencies {} are integrated into the effective dependency base",
        dependency_ids.join(", ")
    )
}

/// Additional metadata preserved while a change is blocked.
#[derive(Debug, Clone, PartialEq, Default)]
pub struct BlockedMetadata {
    /// Machine-readable blocker category.
    pub blocker_reason: Option<String>,
    /// Optional operator-facing unblock guidance.
    pub unblock_metadata: Option<String>,
    /// Optional snapshot of worktree context captured when blocker was recorded.
    pub worktree_snapshot: Option<String>,
    /// True when the hold originated in the Acceptance phase.
    ///
    /// This is the whole Acceptance stall hold. It is reducer-owned in-memory
    /// state for one process lifetime — nothing is written outside the managed
    /// worktree — and it may only suppress ordinary dispatch, drive the
    /// `stalled` presentation, and make an explicit retry eligible. It never
    /// proves PASS, archive readiness, or merge eligibility.
    pub acceptance_stall: bool,
    /// Whether the blocker declared that work can resume once the external
    /// prerequisite is satisfied.
    pub resumable: bool,
    /// Machine-readable blocker kind for a `blocked` change.
    ///
    /// `None` for stalled holds: an execution stop is not a wait on a named
    /// prerequisite, and calling it one would hide why automation gave up.
    pub blocker_kind: BlockerKind,
    /// Verifiable condition that clears a validated external prerequisite wait.
    pub unblock_condition: Option<String>,
    /// Owning team/role or named prerequisite for an external wait.
    pub prerequisite_owner: Option<String>,
    /// Phase that observed a validated external prerequisite.
    pub blocker_origin: Option<String>,
    /// Current unresolved dependency IDs for a dependency wait.
    ///
    /// Structured rather than folded into the detail string: a consumer that
    /// has to decide *which* dependency it is waiting on must not have to parse
    /// operator prose. Empty for every non-dependency hold.
    pub dependency_ids: Vec<String>,
}

/// Reducer-derived operator-facing view of one blocked or stalled change.
///
/// Every surface consumes this instead of inferring lifecycle state itself, so
/// TUI, WebSocket/API, and the dashboard cannot disagree about whether a row is
/// dependency-blocked, externally blocked, or stalled.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BlockerView {
    /// Reducer display status: `blocked` or `stalled`.
    pub status: &'static str,
    /// Machine-readable blocker kind; `None` for a stalled execution hold.
    pub kind: BlockerKind,
    /// Machine-readable blocker reason/category.
    pub category: Option<String>,
    /// One-line operator-facing explanation.
    pub detail: Option<String>,
    /// Verifiable condition that clears an external wait.
    pub unblock_condition: Option<String>,
    /// Owning team/role for an external wait.
    pub prerequisite_owner: Option<String>,
    /// Phase that observed an external prerequisite.
    pub origin: Option<String>,
    /// Whether work can resume once the prerequisite is satisfied.
    pub resumable: bool,
    /// Current unresolved dependency IDs for a dependency wait; empty otherwise.
    pub dependencies: Vec<String>,
}

/// Terminal outcome for a change (once reached, no further transitions).
#[derive(Debug, Clone, PartialEq, Default)]
pub enum TerminalState {
    /// Not yet in a terminal state.
    #[default]
    None,
    /// Successfully merged to the base branch.
    Merged,
    /// Successfully pushed to a remote branch.
    Pushed,
    /// Rejected after acceptance blocker detection.
    Rejected(String),
    /// Encountered a non-recoverable error.
    Error(String),
    /// Stopped by user request.
    Stopped,
}

/// Repository evidence that decides how a stale deferred resolve retry settles.
///
/// Base-branch tree comparison is the authoritative completion evidence, so an
/// unreadable comparison is kept apart from a comparison that ran and found
/// nothing: only the first is a proof of anything.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StaleResolveEvidence {
    /// Base-branch tree comparison proved the change is already integrated.
    Proven,
    /// Base-branch tree comparison ran and the change is not integrated.
    Absent,
    /// Integration evidence could not be read safely.
    Unknown,
}

/// Reducer state a stale deferred resolve retry settled into.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StaleResolveSettlement {
    /// Terminal `merged` recorded from repository-proven base integration.
    Merged,
    /// Retryable manual `merge wait` retained for an explicit operator retry.
    MergeWaitRetained,
    /// The change already carried an immutable outcome or an explicit dequeue,
    /// so only the consumed retry reservation was released.
    AlreadySettled,
}

/// Observation derived from a workspace refresh scan.
/// Used by `apply_observation()` to reconcile MergeWait without overwriting
/// active activity.
#[derive(Debug, Clone, PartialEq, Default)]
pub enum WorkspaceObservation {
    /// No relevant observation.
    #[default]
    None,
    /// Workspace is in `Archived` state: change may need manual `MergeWait` reconciliation.
    WorkspaceArchived,
    /// Worktree is NOT ahead of base: `MergeWait` can be cleared.
    WorktreeNotAhead,
}

/// Full runtime state for a single change, owned by the reducer.
#[derive(Debug, Clone, Default)]
pub struct ChangeRuntimeState {
    /// Queue intent: whether the change has been requested to run.
    pub queue_intent: QueueIntent,
    /// Active execution stage.
    pub activity: ActivityState,
    /// Wait condition (may co-exist with `Queued` intent).
    pub wait_state: WaitState,
    /// Preserved metadata for blocked states.
    pub blocked_metadata: BlockedMetadata,
    /// Terminal outcome once reached.
    pub terminal: TerminalState,
    /// Latest workspace observation (used for reconcile only).
    pub observation: WorkspaceObservation,
    /// True when a change was explicitly force-stopped and dequeued.
    /// While true, stale in-flight execution events must not re-activate this change.
    pub dequeued: bool,
    /// Ephemeral final-commit presentation: the finalization attempt currently
    /// running, if any.
    ///
    /// Presentation only, and process-local by construction. It never changes
    /// [`Self::display_status`], is never persisted, and must never be read by
    /// scheduler eligibility, resume routing, acceptance, archive, merge, or any
    /// other next-action decision — those are re-derived from the workspace
    /// after a restart, which is exactly when this field is gone.
    pub commit_phase_attempt: Option<u32>,
    /// Ephemeral, process-local permission for one manual resolve retry to
    /// continue this change's own unfinished target merge.
    ///
    /// Set only by an explicit operator resolve intent, and cleared the moment a
    /// dispatch that owns the base lane consumes it. It is never persisted and
    /// never routes work on its own: it only decides whether the single admitted
    /// dispatch uses the scoped merge-continuation evidence check instead of the
    /// generic base-dirty preflight. A restart drops it, and the next action is
    /// re-derived from the workspace exactly as before.
    pub manual_resolve_retry: bool,
}

impl ChangeRuntimeState {
    fn clear_blocked_metadata(&mut self) {
        self.blocked_metadata = BlockedMetadata::default();
    }

    fn set_blocked_metadata(
        &mut self,
        blocker_reason: impl Into<String>,
        unblock_metadata: impl Into<String>,
        worktree_snapshot: impl Into<String>,
    ) {
        self.blocked_metadata = BlockedMetadata {
            blocker_reason: Some(blocker_reason.into()),
            unblock_metadata: Some(unblock_metadata.into()),
            worktree_snapshot: Some(worktree_snapshot.into()),
            acceptance_stall: false,
            resumable: false,
            blocker_kind: BlockerKind::None,
            unblock_condition: None,
            prerequisite_owner: None,
            blocker_origin: None,
            dependency_ids: Vec::new(),
        };
    }

    /// Record — or reconcile — the current unresolved dependency wait.
    ///
    /// This is a *state* write, deliberately separate from the edge-triggered
    /// `DependencyBlocked` diagnostic: the scheduler calls it on every coherent
    /// classification, so a reducer that was replaced, refreshed, or never saw
    /// the original transition still carries the current blocker. It is
    /// idempotent, and it never touches queue intent — a dependency wait is a
    /// dispatch exclusion, not a revocation of admitted work.
    ///
    /// Returns `true` when the projected wait or its dependency set changed.
    fn reconcile_dependency_blocker(&mut self, dependency_ids: &[String]) -> bool {
        let unchanged = matches!(self.wait_state, WaitState::DependencyBlocked)
            && self.blocked_metadata.dependency_ids == dependency_ids;
        if unchanged {
            return false;
        }
        self.wait_state = WaitState::DependencyBlocked;
        self.blocked_metadata = BlockedMetadata {
            blocker_reason: Some("dependency_blocked".to_string()),
            unblock_metadata: Some(dependency_blocker_detail(dependency_ids)),
            worktree_snapshot: Some(
                "worktree snapshot not required for dependency blocker".to_string(),
            ),
            acceptance_stall: false,
            resumable: true,
            blocker_kind: BlockerKind::Dependency,
            unblock_condition: Some(dependency_unblock_condition(dependency_ids)),
            prerequisite_owner: None,
            blocker_origin: Some("scheduler".to_string()),
            dependency_ids: dependency_ids.to_vec(),
        };
        true
    }

    /// Reset transient execution state for non-terminal idle/wait transitions.
    fn clear_activity_wait_and_blocker(&mut self) {
        self.activity = ActivityState::Idle;
        self.wait_state = WaitState::None;
        // The permission only means anything while this row is waiting on the
        // base-mutating lane; losing the wait loses the retry it authorized.
        self.manual_resolve_retry = false;
        self.clear_blocked_metadata();
    }

    /// Set a final terminal outcome and clear transient activity/wait metadata.
    fn transition_to_terminal(&mut self, terminal: TerminalState) {
        self.terminal = terminal;
        self.activity = ActivityState::Idle;
        self.wait_state = WaitState::None;
        self.manual_resolve_retry = false;
        self.clear_blocked_metadata();
    }

    /// Record a recoverable stalled/blocker state with structured operator guidance.
    fn transition_to_stalled(
        &mut self,
        blocker_reason: impl Into<String>,
        unblock_metadata: impl Into<String>,
        worktree_snapshot: impl Into<String>,
    ) {
        self.activity = ActivityState::Idle;
        self.wait_state = WaitState::Stalled;
        self.terminal = TerminalState::None;
        self.set_blocked_metadata(blocker_reason, unblock_metadata, worktree_snapshot);
    }

    /// Record a recoverable stalled state that Acceptance owns.
    ///
    /// Identical to [`Self::transition_to_stalled`] except that it also marks
    /// the hold as Acceptance-owned and carries the blocker's resumability,
    /// which together keep the change out of ordinary dispatch until an
    /// explicit retry of a resumable hold — or a restart — clears it.
    fn transition_to_acceptance_stalled(
        &mut self,
        blocker_reason: impl Into<String>,
        unblock_metadata: impl Into<String>,
        worktree_snapshot: impl Into<String>,
        resumable: bool,
    ) {
        self.transition_to_stalled(blocker_reason, unblock_metadata, worktree_snapshot);
        self.blocked_metadata.acceptance_stall = true;
        self.blocked_metadata.resumable = resumable;
    }

    /// Record a validated external prerequisite wait.
    ///
    /// This is the only way to reach [`WaitState::ExternalBlocked`], and it is
    /// reachable only from the orchestrator's classifier — an agent cannot set
    /// it by choosing a verdict word. The hold is in-memory reducer state for
    /// one process lifetime: it may suppress ordinary dispatch and drive the
    /// `blocked` presentation, and it never establishes completion, Acceptance
    /// PASS, archive readiness, or merge eligibility.
    fn transition_to_external_blocked(
        &mut self,
        blocker: &crate::runtime::proposal::ExternalBlockerInfo,
        worktree_snapshot: impl Into<String>,
    ) {
        self.activity = ActivityState::Idle;
        self.wait_state = WaitState::ExternalBlocked;
        self.terminal = TerminalState::None;
        self.blocked_metadata = BlockedMetadata {
            blocker_reason: Some(format!("external-blocked:{}", blocker.category)),
            unblock_metadata: Some(blocker.summary()),
            worktree_snapshot: Some(worktree_snapshot.into()),
            acceptance_stall: matches!(
                blocker.origin,
                crate::runtime::proposal::BlockerOrigin::Acceptance
            ),
            resumable: blocker.resumable,
            blocker_kind: BlockerKind::External,
            unblock_condition: Some(blocker.unblock_condition.clone()),
            prerequisite_owner: blocker.prerequisite_owner.clone(),
            blocker_origin: Some(blocker.origin.as_str().to_string()),
            dependency_ids: Vec::new(),
        };
    }

    /// Whether an Acceptance-owned in-memory hold currently keeps this change
    /// out of ordinary dispatch.
    ///
    /// Both the execution `stalled` hold and a validated external `blocked` wait
    /// observed by Acceptance qualify: they suppress dispatch identically and
    /// differ only in how they are explained to an operator.
    pub fn is_acceptance_stalled(&self) -> bool {
        matches!(
            self.wait_state,
            WaitState::Stalled | WaitState::ExternalBlocked
        ) && self.blocked_metadata.acceptance_stall
    }

    /// Whether the current Acceptance hold allows an acceptance-only retry.
    ///
    /// A non-resumable hold keeps its blocker evidence instead: retrying past it
    /// would dispatch ambiguous work the operator has not unblocked.
    pub fn is_resumable_acceptance_stall(&self) -> bool {
        self.is_acceptance_stalled() && self.blocked_metadata.resumable
    }

    /// Whether a validated external prerequisite currently blocks this change.
    #[allow(dead_code)] // Consumed by external-blocked lifecycle regression coverage.
    pub fn is_external_blocked(&self) -> bool {
        matches!(self.wait_state, WaitState::ExternalBlocked)
    }

    /// Whether a structured blocker classification currently owns this wait.
    ///
    /// `AcceptanceGated` and `ExecutionBlocked` carry validated facts — category,
    /// origin, prerequisite owner, unblock condition, next action, resumability —
    /// that a generic `WorkspaceStatus::Blocked` observation does not have. A
    /// hold established from those facts therefore outranks the coarse
    /// observation for the same non-terminal change: the observation may confirm
    /// that the change is still held, but it must not rebuild the metadata from
    /// generic strings.
    ///
    /// Only the two holds that own routing qualify. A validated external wait
    /// drives dispatch suppression and blocked-phase retry, and an
    /// Acceptance-owned stall drives the same suppression plus resumability.
    /// An Apply-origin non-external stall carries neither, so the existing
    /// generic fallback keeps owning it.
    pub fn has_structured_blocker_hold(&self) -> bool {
        match self.wait_state {
            WaitState::ExternalBlocked => true,
            WaitState::Stalled => self.blocked_metadata.acceptance_stall,
            _ => false,
        }
    }

    /// Machine-readable blocker kind for a `blocked` change.
    ///
    /// Derived from the wait state, never from the recorded metadata alone, so a
    /// stale metadata field can never make a non-blocked change look blocked.
    pub fn blocker_kind(&self) -> BlockerKind {
        match self.wait_state {
            WaitState::DependencyBlocked => BlockerKind::Dependency,
            WaitState::ExternalBlocked => BlockerKind::External,
            _ => BlockerKind::None,
        }
    }

    /// Operator-facing blocker detail for a blocked or stalled row.
    pub fn blocker_detail(&self) -> Option<&str> {
        self.blocked_metadata.unblock_metadata.as_deref()
    }

    /// Reducer-derived blocker view for a blocked or stalled row.
    ///
    /// Returns `None` for every other status so a surface can never render a
    /// blocker badge on a change that is not actually held.
    pub fn blocker_view(&self) -> Option<BlockerView> {
        let status = self.display_status();
        if !matches!(status, "blocked" | "stalled") {
            return None;
        }
        Some(BlockerView {
            status,
            kind: self.blocker_kind(),
            category: self.blocked_metadata.blocker_reason.clone(),
            detail: self.blocker_detail().map(str::to_string),
            unblock_condition: self.blocked_metadata.unblock_condition.clone(),
            prerequisite_owner: self.blocked_metadata.prerequisite_owner.clone(),
            origin: self.blocked_metadata.blocker_origin.clone(),
            resumable: self.blocked_metadata.resumable,
            // Only a dependency wait publishes IDs; deriving them from the kind
            // rather than from the retained field keeps a stale metadata list
            // from decorating an external or stalled hold.
            dependencies: match self.blocker_kind() {
                BlockerKind::Dependency => self.blocked_metadata.dependency_ids.clone(),
                _ => Vec::new(),
            },
        })
    }

    /// Check whether this runtime state represents active execution
    /// (applying, accepting, archiving, or resolving).
    pub fn is_active(&self) -> bool {
        !matches!(self.activity, ActivityState::Idle)
    }

    /// Check whether the change is in a terminal state.
    pub fn is_terminal(&self) -> bool {
        !matches!(self.terminal, TerminalState::None)
    }

    /// Check whether the entry carries no reducer-owned state at all.
    ///
    /// This is the only shape a workspace refresh observation may fill in.
    /// Any activity, wait, queue intent, terminal outcome, or explicit dequeue
    /// is stronger reducer-owned evidence than a refresh scan, so a startup
    /// reconciliation must leave those entries exactly as they are.
    fn is_fresh_idle(&self) -> bool {
        matches!(self.activity, ActivityState::Idle)
            && matches!(self.wait_state, WaitState::None)
            && matches!(self.queue_intent, QueueIntent::NotQueued)
            && matches!(self.terminal, TerminalState::None)
            && !self.dequeued
    }

    /// Check whether a repository-visible success event may replace the current terminal state.
    ///
    /// Recoverable execution errors are terminal for retry/slot accounting, but later
    /// same-change archive/merge/resolve success events are more authoritative for current
    /// display state. Final outcomes such as rejected, already merged, archived, or stopped
    /// remain immutable.
    fn can_success_supersede_terminal(&self) -> bool {
        matches!(self.terminal, TerminalState::None | TerminalState::Error(_))
    }

    /// Verify invariants. Returns `false` if an invalid combination is detected.
    ///
    /// Forbidden combinations:
    /// - `Merged` terminal + any non-Idle activity
    /// - `ResolveWait` + `Resolving` activity simultaneously
    /// - Any terminal state + active activity
    #[allow(dead_code)]
    pub fn invariants_hold(&self) -> bool {
        // Terminal changes must not have active activity.
        if self.is_terminal() && self.is_active() {
            return false;
        }
        // Base-lane wait states and matching active lane occupancy cannot coexist.
        if matches!(self.wait_state, WaitState::ResolveWait)
            && matches!(self.activity, ActivityState::Resolving)
        {
            return false;
        }
        if matches!(self.wait_state, WaitState::RejectWait)
            && matches!(self.activity, ActivityState::Rejecting)
        {
            return false;
        }
        true
    }

    /// Operation label the Apply lane renders for this change.
    ///
    /// `"commit"` while the ephemeral finalization subphase is active, `"apply"`
    /// otherwise. This is a *rendering* projection layered on top of the
    /// canonical `applying` status, which [`Self::display_status`] keeps
    /// returning unchanged throughout finalization.
    pub fn apply_operation_label(&self) -> &'static str {
        match self.commit_phase_attempt {
            Some(_) => "commit",
            None => "apply",
        }
    }

    /// Derive the display status string used by TUI and Web.
    ///
    /// Returns one of: "not queued", "queued", "blocked", "stalled", "preparing",
    /// "applying", "accepting", "rejecting", "archiving", "resolving", "merge wait",
    /// "resolve pending", "reject pending", "merged", "pushed", "error", "stopped".
    ///
    /// The ephemeral commit subphase deliberately does not appear here: it is
    /// presentation detail, and the canonical lifecycle stays `applying` for the
    /// whole finalization sequence.
    pub fn display_status(&self) -> &'static str {
        // Terminal states take precedence.
        match &self.terminal {
            TerminalState::Merged => return "merged",
            TerminalState::Pushed => return "pushed",
            TerminalState::Rejected(_) => return "rejected",
            TerminalState::Error(_) => return "error",
            TerminalState::Stopped => return "stopped",
            TerminalState::None => {}
        }
        // Active execution stages next.
        match self.activity {
            ActivityState::Preparing => return "preparing",
            ActivityState::Applying => return "applying",
            ActivityState::Accepting => return "accepting",
            ActivityState::Rejecting => return "rejecting",
            ActivityState::Archiving => return "archiving",
            ActivityState::Resolving => return "resolving",
            ActivityState::Idle => {}
        }
        // Wait conditions.
        match self.wait_state {
            WaitState::MergeWait => return "merge wait",
            WaitState::ResolveWait => return "resolve pending",
            WaitState::RejectWait => return "reject pending",
            // Dependency waits and validated external prerequisite waits share
            // the `blocked` word; `blocker_kind()` is what keeps them apart.
            WaitState::DependencyBlocked | WaitState::ExternalBlocked => return "blocked",
            WaitState::Stalled => return "stalled",
            WaitState::None => {}
        }
        // Queue intent.
        match self.queue_intent {
            QueueIntent::Queued => "queued",
            QueueIntent::NotQueued => "not queued",
        }
    }

    /// Derive the display color used by TUI status rendering.
    #[allow(dead_code)]
    pub fn display_color(&self) -> ratatui::style::Color {
        match self.display_status() {
            "not queued" => ratatui::style::Color::DarkGray,
            "queued" => ratatui::style::Color::Yellow,
            "blocked" => ratatui::style::Color::Gray,
            "stalled" => ratatui::style::Color::LightYellow,
            "preparing" => ratatui::style::Color::Green,
            "applying" => ratatui::style::Color::Cyan,
            "accepting" => ratatui::style::Color::LightGreen,
            "rejecting" => ratatui::style::Color::LightYellow,
            "archiving" => ratatui::style::Color::Magenta,
            "merged" => ratatui::style::Color::LightBlue,
            "rejected" => ratatui::style::Color::LightRed,
            "merge wait" => ratatui::style::Color::LightMagenta,
            "resolving" => ratatui::style::Color::LightCyan,
            "resolve pending" => ratatui::style::Color::Magenta,
            "reject pending" => ratatui::style::Color::LightMagenta,
            "error" => ratatui::style::Color::Red,
            "stopped" => ratatui::style::Color::DarkGray,
            _ => ratatui::style::Color::DarkGray,
        }
    }

    /// Returns the terminal error message when in error state.
    #[allow(dead_code)]
    pub fn error_message(&self) -> Option<&str> {
        match &self.terminal {
            TerminalState::Error(message) => Some(message.as_str()),
            _ => None,
        }
    }
}

// ============================================================================
// Reducer API types (Phase 2)
// ============================================================================

/// Commands that express user intent and drive state transitions via the reducer.
#[derive(Debug, Clone)]
pub enum ReducerCommand {
    /// Request a change to be added to the execution queue.
    AddToQueue(String),
    /// Explicitly retry a recoverable terminal-error change.
    RetryError(String),
    /// Request a change to be removed from the execution queue.
    RemoveFromQueue(String),
    /// Request merge resolution for a change in MergeWait or ResolveWait.
    ResolveMerge(String),
    /// Cancel a running or queued change and dequeue it back to not-queued.
    DequeueChange(String),
    /// Settle a change into the terminal `stopped` outcome.
    ///
    /// The difference from [`Self::DequeueChange`] is the row an observer is
    /// left with: a dequeued change is idle work the owner may admit again on
    /// its own, while a stopped change is a settled outcome only a new operator
    /// action can move. Targeted force-stop is the one path that produces it —
    /// killing a proposal outright is an operator verdict on that proposal, not
    /// a return to the queue.
    StopChange(String),
    /// Resume a change whose only terminal evidence is an operator stop.
    ///
    /// The counterpart of [`Self::StopChange`], and deliberately narrow: it is
    /// the *one* transition that may clear `TerminalState::Stopped`, and it
    /// clears nothing else. Only explicit Start intent in process mode
    /// `Stopped` submits it — a mark, a bulk mark, a re-mark, a refresh, or a
    /// mark-settlement deadline never does, because "this change is selected"
    /// is not "run this change again".
    ///
    /// [`Self::AddToQueue`] cannot express it: every terminal row is a no-op
    /// there, and widening that guard would let ordinary queue intent erase
    /// merged, pushed, rejected, or error evidence too.
    ResumeStopped(String),
}

/// Outcome of applying a reducer command.
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub enum ReduceOutcome {
    /// The command produced a state change described by the effect.
    Changed(ReducerEffect),
    /// The command was a no-op (idempotent duplicate or invalid in current state).
    NoOp,
}

/// Side-effects produced by a successful reducer command.
#[derive(Debug, Clone)]
#[allow(clippy::enum_variant_names, dead_code)]
pub enum ReducerEffect {
    /// Queue intent was updated.
    QueueIntentSet {
        change_id: String,
        intent: QueueIntent,
    },
    /// Wait state was updated.
    WaitStateSet { change_id: String, wait: WaitState },
    /// Terminal state was set.
    TerminalStateSet {
        change_id: String,
        terminal: TerminalState,
    },
}

/// Shared state for orchestration operations.
///
/// This structure tracks:
/// - Which changes are pending, completed, or archived
/// - Apply counts per change
/// - Iteration counters and progress
#[derive(Debug, Clone)]
pub struct OrchestratorState {
    /// Change IDs captured at run start (snapshot).
    /// Only changes present in this snapshot will be processed.
    initial_change_ids: HashSet<String>,

    /// Changes that are still pending (not yet archived).
    pending_changes: HashSet<String>,

    /// Changes that have been archived.
    archived_changes: HashSet<String>,

    /// Apply counts per change (how many times each change has been applied).
    apply_counts: HashMap<String, u32>,

    /// Changes marked as stalled (failed due to no progress or blockers).
    stalled_change_ids: HashSet<String>,

    /// Changes skipped because one of their dependencies stalled.
    skipped_change_ids: HashSet<String>,

    /// Task progress per change (completed_tasks, total_tasks).
    task_progress: HashMap<String, (u32, u32)>,

    /// Error history per change for repeated failure detection.
    error_histories: HashMap<String, ErrorHistory>,

    /// Number of changes processed (archived).
    changes_processed: usize,

    /// Total number of changes at run start.
    total_changes: usize,

    /// Maximum iterations limit (0 = no limit).
    max_iterations: u32,

    /// Current iteration number.
    iteration: u32,

    /// Current change ID being processed.
    current_change_id: Option<String>,

    /// Reducer-owned runtime state per change.
    change_runtime: HashMap<String, ChangeRuntimeState>,

    /// Reducer-owned resolve-wait queue (FIFO list of change_ids awaiting resolve).
    resolve_wait_queue: Vec<String>,

    /// Reducer-owned reject-wait queue (FIFO list of change_ids awaiting rejection review).
    reject_wait_queue: Vec<String>,

    /// Changes stopped by the per-change Apply-dispatch ceiling, in observation
    /// order.
    ///
    /// Parallel execution has no return channel for a typed Apply outcome, so
    /// the budget owner's refusal is recorded here with its exact cumulative
    /// count. That keeps `iteration_limit` a typed run-level outcome instead of
    /// something a caller would have to re-derive by parsing an error string.
    /// Active-run memory like the rest of this reducer.
    apply_iteration_limits: Vec<ApplyIterationLimit>,
}

/// One change stopped by the configured Apply-dispatch ceiling.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApplyIterationLimit {
    /// The change whose Apply budget was spent.
    pub change_id: String,
    /// Exact cumulative configured Apply dispatches reserved for that change.
    pub attempts: u32,
    /// The configured ceiling that refused the next dispatch.
    pub max: u32,
}

#[allow(dead_code)] // Public API for future use by TUI/Web states
impl OrchestratorState {
    /// Create a new orchestrator state with the given initial changes.
    pub fn new(change_ids: Vec<String>, max_iterations: u32) -> Self {
        let initial_set: HashSet<String> = change_ids.iter().cloned().collect();
        let pending_set = initial_set.clone();
        let total = change_ids.len();

        // Initialise each change with the "not queued + idle + no wait + no terminal" state.
        let change_runtime: HashMap<String, ChangeRuntimeState> = change_ids
            .iter()
            .map(|id| (id.clone(), ChangeRuntimeState::default()))
            .collect();

        Self {
            initial_change_ids: initial_set,
            pending_changes: pending_set,
            archived_changes: HashSet::new(),
            apply_counts: HashMap::new(),
            stalled_change_ids: HashSet::new(),
            skipped_change_ids: HashSet::new(),
            task_progress: HashMap::new(),
            error_histories: HashMap::new(),
            changes_processed: 0,
            total_changes: total,
            max_iterations,
            iteration: 0,
            current_change_id: None,
            change_runtime,
            resolve_wait_queue: Vec::new(),
            reject_wait_queue: Vec::new(),
            apply_iteration_limits: Vec::new(),
        }
    }

    /// Get the initial snapshot of change IDs.
    pub fn initial_change_ids(&self) -> &HashSet<String> {
        &self.initial_change_ids
    }

    /// Get the set of pending changes.
    pub fn pending_changes(&self) -> &HashSet<String> {
        &self.pending_changes
    }

    /// Get the set of archived changes.
    pub fn archived_changes(&self) -> &HashSet<String> {
        &self.archived_changes
    }

    /// Get stalled change IDs.
    pub fn stalled_change_ids(&self) -> &HashSet<String> {
        &self.stalled_change_ids
    }

    /// Get skipped change IDs.
    pub fn skipped_change_ids(&self) -> &HashSet<String> {
        &self.skipped_change_ids
    }

    /// Change IDs currently held by an in-memory Acceptance stall.
    ///
    /// This is the whole hold: nothing is persisted, so a restarted process
    /// returns an empty set and repository evidence alone decides the next
    /// action. Apply-origin and dependency blockers are deliberately excluded —
    /// only an Acceptance-phase blocker observation sets the flag.
    /// Changes held by a validated external prerequisite, whichever phase
    /// observed it.
    ///
    /// Ordinary dispatch must skip these. They are *not* dependency edges: a
    /// dependent proposal keeps its own dependency blocker kind, and unrelated
    /// ready changes stay eligible.
    pub fn externally_blocked_change_ids(&self) -> HashSet<String> {
        self.change_runtime
            .iter()
            .filter(|(_, rt)| rt.is_external_blocked())
            .map(|(id, _)| id.clone())
            .collect()
    }

    pub fn acceptance_stalled_change_ids(&self) -> HashSet<String> {
        self.change_runtime
            .iter()
            .filter(|(_, rt)| rt.is_acceptance_stalled())
            .map(|(change_id, _)| change_id.clone())
            .collect()
    }

    /// Mark a change as stalled.
    pub fn mark_stalled(&mut self, change_id: String) {
        self.stalled_change_ids.insert(change_id);
    }

    /// Record that the Apply-dispatch ceiling refused another dispatch.
    ///
    /// Repeated observations for the same change keep the first record: the
    /// exact count that first hit the ceiling is the one the finish hook
    /// reports.
    pub fn record_apply_iteration_limit(&mut self, change_id: &str, attempts: u32, max: u32) {
        if self
            .apply_iteration_limits
            .iter()
            .any(|record| record.change_id == change_id)
        {
            return;
        }
        self.apply_iteration_limits.push(ApplyIterationLimit {
            change_id: change_id.to_string(),
            attempts,
            max,
        });
    }

    /// Changes stopped by the Apply-dispatch ceiling, in observation order.
    pub fn apply_iteration_limits(&self) -> &[ApplyIterationLimit] {
        &self.apply_iteration_limits
    }

    /// Drop the Apply-dispatch ceiling record for one change.
    ///
    /// Returns true when a record was actually removed. Called from
    /// [`Self::retry_terminal_error`], so the diagnostic is consumed by the same
    /// explicit intent that consumes the terminal error it explains — and by
    /// nothing else.
    pub fn clear_apply_iteration_limit(&mut self, change_id: &str) -> bool {
        let before = self.apply_iteration_limits.len();
        self.apply_iteration_limits
            .retain(|record| record.change_id != change_id);
        self.apply_iteration_limits.len() != before
    }

    /// The typed Apply-dispatch ceiling record for one change, if it has one.
    ///
    /// Diagnostic evidence, never an operator-action gate: it answers why one
    /// invocation stopped, not whether a later explicit command may open a new
    /// one. Retry admission reads the target's own terminal-error evidence
    /// instead, so a record retained under a still-live persistent scheduler
    /// cannot make an explicit retry impossible.
    pub fn apply_iteration_limit(&self, change_id: &str) -> Option<&ApplyIterationLimit> {
        self.apply_iteration_limits
            .iter()
            .find(|record| record.change_id == change_id)
    }

    /// Finish status and Apply count a parallel run reports to `on_finish`.
    ///
    /// Parallel execution has no `LoopControl` return path, so this reducer is
    /// where the typed budget outcome crosses the boundary. Every parallel run
    /// boundary — `cflx run` and the TUI alike — derives the hook's status from
    /// this one observation instead of parsing an error string, so both report
    /// `iteration_limit` with the same exact cumulative count.
    pub fn parallel_finish_report(&self) -> (&'static str, u32) {
        match self.apply_iteration_limits.first() {
            Some(record) => ("iteration_limit", record.attempts),
            None => ("completed", 0),
        }
    }

    /// Mark a change as skipped.
    ///
    /// Returns true when the value was newly inserted.
    pub fn mark_skipped(&mut self, change_id: String) -> bool {
        self.skipped_change_ids.insert(change_id)
    }

    /// Clear transient counters for a stalled change.
    pub fn clear_stalled_change(&mut self, change_id: &str) {
        self.stalled_change_ids.remove(change_id);
        self.skipped_change_ids.remove(change_id);
        self.apply_counts.remove(change_id);
        if self.current_change_id.as_deref() == Some(change_id) {
            self.current_change_id = None;
        }
    }

    /// Get the number of changes processed.
    pub fn changes_processed(&self) -> usize {
        self.changes_processed
    }

    /// Get the total number of changes.
    pub fn total_changes(&self) -> usize {
        self.total_changes
    }

    /// Get the number of remaining changes.
    pub fn remaining_changes(&self) -> usize {
        self.pending_changes.len()
    }

    /// Get the current iteration number.
    pub fn iteration(&self) -> u32 {
        self.iteration
    }

    /// Get the maximum iterations limit.
    pub fn max_iterations(&self) -> u32 {
        self.max_iterations
    }

    /// Get the current change ID being processed.
    pub fn current_change_id(&self) -> Option<&String> {
        self.current_change_id.as_ref()
    }

    /// Get the apply count for a specific change.
    pub fn apply_count(&self, change_id: &str) -> u32 {
        *self.apply_counts.get(change_id).unwrap_or(&0)
    }

    /// Get the task progress for a specific change (completed_tasks, total_tasks).
    pub fn task_progress(&self, change_id: &str) -> (u32, u32) {
        *self.task_progress.get(change_id).unwrap_or(&(0, 0))
    }

    /// Update task progress for a change.
    pub fn set_task_progress(&mut self, change_id: String, completed: u32, total: u32) {
        self.task_progress.insert(change_id, (completed, total));
    }

    /// Check if a change is in the initial snapshot.
    pub fn is_in_snapshot(&self, change_id: &str) -> bool {
        self.initial_change_ids.contains(change_id)
    }

    /// Check if a change is pending.
    pub fn is_pending(&self, change_id: &str) -> bool {
        self.pending_changes.contains(change_id)
    }

    /// Check if a change is archived.
    pub fn is_archived(&self, change_id: &str) -> bool {
        self.archived_changes.contains(change_id)
    }

    /// Check if all changes are done.
    pub fn is_complete(&self) -> bool {
        self.pending_changes.is_empty()
    }

    /// Check if max iterations has been reached.
    pub fn is_iteration_limit_reached(&self) -> bool {
        self.max_iterations > 0 && self.iteration >= self.max_iterations
    }

    /// Check if we're approaching the iteration limit (80%).
    pub fn is_approaching_iteration_limit(&self) -> bool {
        if self.max_iterations == 0 {
            return false;
        }
        let threshold = (self.max_iterations as f32 * 0.8) as u32;
        self.iteration == threshold
    }

    /// Increment the iteration counter.
    pub fn increment_iteration(&mut self) {
        self.iteration += 1;
    }

    /// Set the current change being processed.
    pub fn set_current_change(&mut self, change_id: Option<String>) {
        self.current_change_id = change_id;
    }

    /// Increment the apply count for a change and return the new count.
    pub fn increment_apply_count(&mut self, change_id: &str) -> u32 {
        let count = self.apply_counts.entry(change_id.to_string()).or_insert(0);
        *count += 1;
        *count
    }

    /// Mark a change as archived.
    ///
    /// This:
    /// - Moves the change from pending to archived
    /// - Increments the changes_processed counter
    /// - Clears the current change if it matches
    /// - Removes apply count tracking
    pub fn mark_archived(&mut self, change_id: &str) {
        if self.pending_changes.remove(change_id) {
            self.archived_changes.insert(change_id.to_string());
            self.changes_processed += 1;
            self.apply_counts.remove(change_id);

            if self.current_change_id.as_deref() == Some(change_id) {
                self.current_change_id = None;
            }
        }
    }

    /// Add a new change dynamically (during execution).
    ///
    /// This is used for dynamic queue support in TUI mode.
    pub fn add_dynamic_change(&mut self, change_id: String) {
        if !self.initial_change_ids.contains(&change_id)
            && !self.pending_changes.contains(&change_id)
            && !self.archived_changes.contains(&change_id)
        {
            self.initial_change_ids.insert(change_id.clone());
            self.pending_changes.insert(change_id.clone());
            self.total_changes += 1;
            // Initialise reducer runtime state for newly discovered change.
            self.change_runtime.entry(change_id).or_default();
        }
    }

    // -----------------------------------------------------------------------
    // Reducer-owned runtime state accessors (Phase 1)
    // -----------------------------------------------------------------------

    /// Get the runtime state for a change, creating a default entry if absent.
    fn runtime_entry(&mut self, change_id: &str) -> &mut ChangeRuntimeState {
        self.change_runtime
            .entry(change_id.to_string())
            .or_default()
    }

    /// Every change the reducer currently tracks, sorted for determinism.
    ///
    /// This is the authoritative candidate set for an operation that addresses
    /// "all changes": it is the same map `display_status` answers from, so a
    /// bulk mutation can never classify a row the reducer does not know about.
    pub fn tracked_change_ids(&self) -> Vec<String> {
        let mut ids: Vec<String> = self.change_runtime.keys().cloned().collect();
        ids.sort();
        ids
    }

    /// Read-only access to the runtime state of a change.
    pub fn change_runtime(&self, change_id: &str) -> Option<&ChangeRuntimeState> {
        self.change_runtime.get(change_id)
    }

    /// Whether the reducer tracks this change at all.
    ///
    /// Distinct from `display_status(change_id) == "not queued"`, which is also
    /// what an untracked ID answers: a command that must refuse an unknown
    /// target cannot tell the two apart from the status string alone.
    pub fn is_tracked_change(&self, change_id: &str) -> bool {
        self.change_runtime.contains_key(change_id)
    }

    /// Derive the UI display status string for a change (Phase 1.4).
    pub fn display_status(&self, change_id: &str) -> &'static str {
        match self.change_runtime.get(change_id) {
            Some(rt) => rt.display_status(),
            None => "not queued",
        }
    }

    /// Return true if the change is actively executing (applying/accepting/archiving/resolving).
    /// Used for parallel slot accounting (Phase 1.5).
    pub fn is_active_change(&self, change_id: &str) -> bool {
        self.change_runtime
            .get(change_id)
            .map(|rt| rt.is_active())
            .unwrap_or(false)
    }

    /// Return true when any change is running agent-driven workspace execution.
    ///
    /// Applying, accepting, and archiving are the activities backed by an agent
    /// command process. Resolving and rejecting occupy the base-mutating lane and
    /// are scheduler-owned shutdown work, so they are deliberately excluded here:
    /// they must never justify a force-stopped-process claim.
    pub fn is_agent_execution_active(&self) -> bool {
        self.change_runtime.values().any(|rt| {
            matches!(
                rt.activity,
                ActivityState::Applying | ActivityState::Accepting | ActivityState::Archiving
            ) && !rt.is_terminal()
        })
    }

    /// Return true when any change is currently in resolving activity.
    pub fn is_resolving_active(&self) -> bool {
        self.change_runtime
            .values()
            .any(|rt| matches!(rt.activity, ActivityState::Resolving) && !rt.is_terminal())
    }

    /// Return true when any change is currently in rejecting activity.
    pub fn is_rejecting_active(&self) -> bool {
        self.change_runtime
            .values()
            .any(|rt| matches!(rt.activity, ActivityState::Rejecting) && !rt.is_terminal())
    }

    /// Return true when another change occupies the post-archive merge/reject lane.
    pub fn has_other_post_archive_lane_blocker(&self, change_id: &str) -> bool {
        self.change_runtime.iter().any(|(id, rt)| {
            id != change_id
                && matches!(
                    rt.activity,
                    ActivityState::Resolving | ActivityState::Rejecting
                )
                && !rt.is_terminal()
        })
    }

    /// Return the non-terminal change that occupies the base-mutating lane, if any.
    pub fn base_mutating_lane_occupant(&self) -> Option<String> {
        self.change_runtime.iter().find_map(|(id, rt)| {
            if matches!(
                rt.activity,
                ActivityState::Resolving | ActivityState::Rejecting
            ) && !rt.is_terminal()
            {
                Some(id.clone())
            } else {
                None
            }
        })
    }

    /// Return true if the single base-mutating lane is currently occupied.
    pub fn is_base_mutating_lane_occupied(&self) -> bool {
        self.base_mutating_lane_occupant().is_some()
    }

    /// Verify global reducer invariants that span multiple changes.
    pub fn global_invariants_hold(&self) -> bool {
        let lane_occupants = self
            .change_runtime
            .values()
            .filter(|rt| {
                matches!(
                    rt.activity,
                    ActivityState::Resolving | ActivityState::Rejecting
                ) && !rt.is_terminal()
            })
            .count();
        lane_occupants <= 1
            && self
                .change_runtime
                .values()
                .all(ChangeRuntimeState::invariants_hold)
    }

    /// Return change IDs that are currently waiting for manual merge resolution.
    pub fn merge_wait_change_ids(&self) -> Vec<String> {
        self.change_runtime
            .iter()
            .filter_map(|(id, rt)| {
                if matches!(rt.wait_state, WaitState::MergeWait) && !rt.is_terminal() {
                    Some(id.clone())
                } else {
                    None
                }
            })
            .collect()
    }

    /// Return change IDs that are currently waiting for scheduler-owned resolve/merge retry.
    pub fn resolve_wait_change_ids(&self) -> Vec<String> {
        self.change_runtime
            .iter()
            .filter_map(|(id, rt)| {
                if matches!(rt.wait_state, WaitState::ResolveWait) && !rt.is_terminal() {
                    Some(id.clone())
                } else {
                    None
                }
            })
            .collect()
    }

    fn remove_from_resolve_wait_queue(&mut self, change_id: &str) {
        self.resolve_wait_queue.retain(|id| id != change_id);
    }

    fn remove_from_reject_wait_queue(&mut self, change_id: &str) {
        self.reject_wait_queue.retain(|id| id != change_id);
    }

    fn clear_base_mutating_wait_queues(&mut self, change_id: &str) {
        self.remove_from_resolve_wait_queue(change_id);
        self.remove_from_reject_wait_queue(change_id);
    }

    fn enqueue_unique_resolve_wait(&mut self, change_id: &str) {
        if !self.resolve_wait_queue.iter().any(|id| id == change_id) {
            self.resolve_wait_queue.push(change_id.to_string());
        }
    }

    fn enqueue_unique_reject_wait(&mut self, change_id: &str) {
        if !self.reject_wait_queue.iter().any(|id| id == change_id) {
            self.reject_wait_queue.push(change_id.to_string());
        }
    }

    /// Clear reducer-owned resolve retry intent for a change after repository-visible
    /// merge success or stale already-merged detection.
    ///
    /// Releasing a consumed reservation is bookkeeping, never an outcome.
    /// `ReducerCommand::ResolveMerge` leaves `QueueIntent::NotQueued` behind the
    /// `ResolveWait`, so dropping the wait on its own would expose an idle
    /// `not queued` row for a change that is still archived and unmerged. An
    /// entry with no terminal state, no activity, and no queue intent therefore
    /// falls back to the retryable manual `merge wait` it was promoted from;
    /// a later success event supersedes that safely.
    pub fn clear_resolve_wait_intent(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        if matches!(rt.wait_state, WaitState::ResolveWait) {
            rt.wait_state = if rt.is_terminal()
                || rt.dequeued
                || rt.is_active()
                || matches!(rt.queue_intent, QueueIntent::Queued)
            {
                WaitState::None
            } else {
                WaitState::MergeWait
            };
            rt.clear_blocked_metadata();
        }
        self.remove_from_resolve_wait_queue(change_id);
    }

    /// Settle a stale deferred resolve retry from base-branch tree evidence.
    ///
    /// This is the typed transition the scheduler applies at the stale-retry
    /// boundary. Terminal `merged` is recorded *before* the retry reservation is
    /// dropped, so no reader observes an idle `not queued` gap between accepted
    /// retry intent and the settled outcome. Evidence that is absent or
    /// unreadable fails closed to retryable manual `merge wait`; only proven
    /// base integration may report success.
    ///
    /// Purely a state transition: it reads no repository content and mutates no
    /// index, worktree, or file.
    pub fn settle_stale_resolve_retry(
        &mut self,
        change_id: &str,
        evidence: StaleResolveEvidence,
    ) -> StaleResolveSettlement {
        match evidence {
            StaleResolveEvidence::Proven => {
                if !self
                    .runtime_entry(change_id)
                    .can_success_supersede_terminal()
                {
                    self.clear_base_mutating_wait_queues(change_id);
                    return StaleResolveSettlement::AlreadySettled;
                }
                // Terminal first, membership second.
                self.transition_change_to_merged(change_id);
                StaleResolveSettlement::Merged
            }
            StaleResolveEvidence::Absent | StaleResolveEvidence::Unknown => {
                let rt = self.runtime_entry(change_id);
                if rt.is_terminal() || rt.dequeued {
                    self.clear_base_mutating_wait_queues(change_id);
                    return StaleResolveSettlement::AlreadySettled;
                }
                rt.activity = ActivityState::Idle;
                rt.wait_state = WaitState::MergeWait;
                rt.queue_intent = QueueIntent::NotQueued;
                rt.clear_blocked_metadata();
                self.clear_base_mutating_wait_queues(change_id);
                StaleResolveSettlement::MergeWaitRetained
            }
        }
    }

    /// Return change IDs that are currently waiting for scheduler-owned rejection review.
    pub fn reject_wait_change_ids(&self) -> Vec<String> {
        self.change_runtime
            .iter()
            .filter_map(|(id, rt)| {
                if matches!(rt.wait_state, WaitState::RejectWait) && !rt.is_terminal() {
                    Some(id.clone())
                } else {
                    None
                }
            })
            .collect()
    }

    /// Mark a change as waiting for rejection review after a rejection proposal handoff.
    pub fn mark_reject_wait(&mut self, change_id: &str) {
        let lane_blocked = self.has_other_post_archive_lane_blocker(change_id);
        let rt = self.runtime_entry(change_id);
        if rt.is_terminal() || rt.dequeued {
            return;
        }
        rt.activity = if lane_blocked {
            ActivityState::Idle
        } else {
            ActivityState::Rejecting
        };
        rt.wait_state = if lane_blocked {
            WaitState::RejectWait
        } else {
            WaitState::None
        };
        rt.clear_blocked_metadata();
        self.remove_from_resolve_wait_queue(change_id);
        if lane_blocked {
            self.enqueue_unique_reject_wait(change_id);
        } else {
            self.remove_from_reject_wait_queue(change_id);
        }
    }

    /// Clear reducer-owned rejection-review wait intent.
    pub fn clear_reject_wait_intent(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        if matches!(rt.wait_state, WaitState::RejectWait) {
            rt.wait_state = WaitState::None;
            rt.clear_blocked_metadata();
        }
        self.remove_from_reject_wait_queue(change_id);
    }

    /// Release a promoted base-mutating retry occupant after a non-terminal retry outcome.
    ///
    /// Spawned ResolveWait/RejectWait retries promote a queued waiter by clearing its
    /// wait state and marking it Resolving/Rejecting. If the detached retry then
    /// defers auto-resumably or fails before a terminal reducer event, the lane must
    /// be returned to its reducer-owned wait state so later completion triggers can
    /// promote it again. Terminal entries, non-occupants, and unsupported wait kinds
    /// are left unchanged.
    pub fn release_base_mutating_lane_after_retry(
        &mut self,
        change_id: &str,
        wait_state: WaitState,
    ) -> bool {
        if !matches!(wait_state, WaitState::ResolveWait | WaitState::RejectWait) {
            return false;
        }

        let rt = self.runtime_entry(change_id);
        if rt.is_terminal()
            || !matches!(
                rt.activity,
                ActivityState::Resolving | ActivityState::Rejecting
            )
        {
            return false;
        }

        rt.activity = ActivityState::Idle;
        rt.wait_state = wait_state.clone();
        rt.clear_blocked_metadata();
        self.clear_base_mutating_wait_queues(change_id);
        match wait_state {
            WaitState::ResolveWait => self.enqueue_unique_resolve_wait(change_id),
            WaitState::RejectWait => self.enqueue_unique_reject_wait(change_id),
            _ => unreachable!("wait_state was validated above"),
        }
        true
    }

    /// Abandon a promoted base-mutating retry occupant after a give-up outcome.
    ///
    /// Unlike `release_base_mutating_lane_after_retry`, this does not restore the
    /// original wait state or re-enqueue the change. Give-up paths have determined
    /// there is no live retry work left, so the lane is simply released.
    pub fn abandon_base_mutating_lane_occupant(&mut self, change_id: &str) -> bool {
        let rt = self.runtime_entry(change_id);
        if rt.is_terminal()
            || !matches!(
                rt.activity,
                ActivityState::Resolving | ActivityState::Rejecting
            )
        {
            return false;
        }

        rt.activity = ActivityState::Idle;
        rt.wait_state = WaitState::None;
        rt.clear_blocked_metadata();
        self.clear_base_mutating_wait_queues(change_id);
        true
    }

    /// Promote exactly one pending base-mutating operation when the lane is free.
    /// Resolve waits take priority over reject waits to preserve existing merge retry semantics.
    pub fn promote_next_base_mutating_lane_waiter(&mut self) -> Option<(String, WaitState)> {
        if self.is_base_mutating_lane_occupied() {
            return None;
        }

        while let Some(change_id) = self.resolve_wait_queue.first().cloned() {
            self.resolve_wait_queue.remove(0);
            let rt = self.runtime_entry(&change_id);
            if !rt.is_terminal() && matches!(rt.wait_state, WaitState::ResolveWait) {
                rt.wait_state = WaitState::None;
                rt.activity = ActivityState::Resolving;
                rt.clear_blocked_metadata();
                return Some((change_id, WaitState::ResolveWait));
            }
        }

        while let Some(change_id) = self.reject_wait_queue.first().cloned() {
            self.reject_wait_queue.remove(0);
            let rt = self.runtime_entry(&change_id);
            if !rt.is_terminal() && matches!(rt.wait_state, WaitState::RejectWait) {
                rt.wait_state = WaitState::None;
                rt.activity = ActivityState::Rejecting;
                rt.clear_blocked_metadata();
                return Some((change_id, WaitState::RejectWait));
            }
        }

        None
    }

    /// Whether `change_id` currently carries an unconsumed manual resolve retry
    /// permission.
    ///
    /// Read — never taken — at promotion, so a dispatch that is deferred by
    /// base-lane occupancy leaves the permission for the auto-resumed retry.
    pub fn has_manual_resolve_retry(&self, change_id: &str) -> bool {
        self.change_runtime
            .get(change_id)
            .is_some_and(|rt| rt.manual_resolve_retry)
    }

    /// Consume `change_id`'s manual resolve retry permission.
    ///
    /// Answers what the permission was, and clears it, so the next ordinary
    /// scheduled attempt observes the unchanged generic preflight.
    pub fn consume_manual_resolve_retry(&mut self, change_id: &str) -> bool {
        match self.change_runtime.get_mut(change_id) {
            Some(rt) => std::mem::take(&mut rt.manual_resolve_retry),
            None => false,
        }
    }

    /// Return change IDs that still carry queued intent and are not terminal.
    ///
    /// Scheduler reconciliation uses this as reducer-visible source of truth for
    /// queue intent, then intersects with loadable OpenSpec changes and runtime
    /// active-state checks before dispatch/analysis.
    pub fn queued_change_ids(&self) -> Vec<String> {
        self.change_runtime
            .iter()
            .filter_map(|(id, rt)| {
                if matches!(rt.queue_intent, QueueIntent::Queued) && !rt.is_terminal() {
                    Some(id.clone())
                } else {
                    None
                }
            })
            .collect()
    }

    /// Return change IDs currently considered active by reducer runtime state.
    pub fn active_change_ids(&self) -> Vec<String> {
        self.change_runtime
            .iter()
            .filter_map(|(id, rt)| {
                if rt.is_active() {
                    Some(id.clone())
                } else {
                    None
                }
            })
            .collect()
    }

    /// Return a snapshot of display status strings for all known changes.
    /// Used by the TUI to sync `ChangeState.queue_status` from the reducer.
    pub fn all_display_statuses(&self) -> HashMap<String, &'static str> {
        self.change_runtime
            .iter()
            .map(|(id, rt)| (id.clone(), rt.display_status()))
            .collect()
    }

    /// Ephemeral Apply-lane operation label for every known change.
    ///
    /// `"commit"` only while a finalization subphase is active. Frontends read
    /// this for rendering; nothing in the scheduler, resume router, or gating
    /// logic may consult it.
    pub fn all_apply_operation_labels(&self) -> HashMap<String, &'static str> {
        self.change_runtime
            .iter()
            .map(|(id, rt)| (id.clone(), rt.apply_operation_label()))
            .collect()
    }

    /// Reducer-retained final diagnostics for every change in
    /// [`TerminalState::Error`].
    ///
    /// Sanitized with the same helper the `/api/v2` `error_detail` projection
    /// uses, so the diagnostic a TUI row shows and the one a remote client reads
    /// are the same text rather than two independently derived strings. A change
    /// that is not in error contributes no entry at all — absence is what the
    /// frontends turn into their explicit "unavailable" fallback.
    pub fn all_error_details(&self) -> HashMap<String, String> {
        self.change_runtime
            .iter()
            .filter_map(|(id, rt)| {
                rt.error_message()
                    .map(|message| (id.clone(), crate::events::sanitize_detail(message)))
            })
            .collect()
    }

    /// Reducer-derived blocker views for every change currently blocked or
    /// stalled.
    ///
    /// This is the single source TUI, WebSocket/API, and the dashboard read, so
    /// no surface has to re-derive `blocked` versus `stalled` — or the blocker
    /// kind — from filenames, prose, or task text.
    pub fn all_blocker_views(&self) -> HashMap<String, BlockerView> {
        self.change_runtime
            .iter()
            .filter_map(|(id, rt)| rt.blocker_view().map(|view| (id.clone(), view)))
            .collect()
    }

    /// Reducer-derived blocker view for one change.
    pub fn blocker_view(&self, change_id: &str) -> Option<BlockerView> {
        self.change_runtime
            .get(change_id)
            .and_then(ChangeRuntimeState::blocker_view)
    }

    /// Reconcile the current unresolved dependency set for one change.
    ///
    /// The scheduler calls this on every coherent dependency classification,
    /// whether or not the edge-triggered `DependencyBlocked` diagnostic was
    /// suppressed as a duplicate. Deduplication decides whether an *operator
    /// diagnostic* is worth emitting again; it must never decide whether the
    /// current blocker exists, or a refreshed, replaced, or newly built reducer
    /// would report an unresolved dependency wait as plain `queued`.
    ///
    /// Queue intent is deliberately untouched: a dependency wait excludes a
    /// change from dispatch and never revokes admitted work.
    ///
    /// Returns `true` when the projection changed.
    pub fn reconcile_dependency_blocker(
        &mut self,
        change_id: &str,
        dependency_ids: &[String],
    ) -> bool {
        let rt = self.runtime_entry(change_id);
        // Exactly the guard the `DependencyBlocked` event applies, so the
        // reconciliation path and the transition path can never disagree about
        // which rows may carry a dependency wait.
        if rt.is_terminal() || rt.is_active() {
            return false;
        }
        rt.reconcile_dependency_blocker(dependency_ids)
    }

    /// Clear a dependency wait once classification proved every dependency resolved.
    ///
    /// Only a dependency wait is cleared: an external prerequisite, an
    /// acceptance stall, or a merge wait belongs to another owner, and dropping
    /// it here would let dependency evidence erase an unrelated hold.
    ///
    /// Returns `true` when a dependency wait was actually cleared.
    pub fn clear_dependency_blocker(&mut self, change_id: &str) -> bool {
        let Some(rt) = self.change_runtime.get_mut(change_id) else {
            return false;
        };
        if !matches!(rt.wait_state, WaitState::DependencyBlocked) {
            return false;
        }
        rt.wait_state = WaitState::None;
        rt.clear_blocked_metadata();
        true
    }

    /// Return true if the change has reached a terminal state.
    pub fn is_terminal_change(&self, change_id: &str) -> bool {
        self.change_runtime
            .get(change_id)
            .map(|rt| rt.is_terminal())
            .unwrap_or(false)
    }

    /// Remove a change from pending (e.g., due to failure).
    pub fn remove_from_pending(&mut self, change_id: &str) {
        self.pending_changes.remove(change_id);
        if self.current_change_id.as_deref() == Some(change_id) {
            self.current_change_id = None;
        }
    }

    /// Remove a change from pending and decrement total_changes.
    ///
    /// Returns true when the change was pending and removed.
    pub fn drop_pending_change(&mut self, change_id: &str) -> bool {
        let removed = self.pending_changes.remove(change_id);
        if removed {
            self.total_changes = self.total_changes.saturating_sub(1);
        }
        if self.current_change_id.as_deref() == Some(change_id) {
            self.current_change_id = None;
        }
        removed
    }

    /// Clear all pending changes.
    pub fn clear_pending_changes(&mut self) {
        self.pending_changes.clear();
    }

    /// Record an error and return whether circuit breaker should trip.
    pub fn record_error_and_check_circuit_breaker(
        &mut self,
        change_id: &str,
        error: &str,
        config: CircuitBreakerConfig,
    ) -> bool {
        let history = self
            .error_histories
            .entry(change_id.to_string())
            .or_insert_with(|| ErrorHistory::new(config.clone()));

        history.record_error(error);
        history.detect_same_error()
    }

    /// Return the most recent normalized error for a change.
    pub fn last_error(&self, change_id: &str) -> Option<&str> {
        self.error_histories
            .get(change_id)
            .and_then(ErrorHistory::last_error)
    }

    /// Clear error history for a change.
    pub fn clear_error_history(&mut self, change_id: &str) {
        self.error_histories.remove(change_id);
    }

    /// Return true when a final terminal state must stop ordinary apply/archive dispatch.
    ///
    /// Recoverable terminal errors intentionally remain separate: they are terminal for
    /// slot/accounting purposes, but can only become dispatchable through an explicit
    /// `ReducerCommand::RetryError` transition.
    pub fn is_final_terminal_dispatch_stop(&self, change_id: &str) -> bool {
        self.change_runtime
            .get(change_id)
            .map(|rt| {
                matches!(
                    rt.terminal,
                    TerminalState::Merged | TerminalState::Pushed | TerminalState::Rejected(_)
                )
            })
            .unwrap_or(false)
    }

    /// Return true when current reducer intent still admits `change_id` as ordinary queued work.
    ///
    /// This is the revocation check a scheduler wake-up hint must pass before it
    /// is treated as work. `RemoveFromQueue` and `DequeueChange` clear queue
    /// intent immediately, so a dynamic-queue entry left over from an earlier
    /// accepted addition cannot reacquire the change; an explicit `AddToQueue`
    /// or `RetryError` restores eligibility.
    ///
    /// A change the reducer has never seen carries no execution intent at all, so
    /// it is not eligible either. Every accepted path into ordinary work —
    /// start, `AddToQueue`, `RetryError`, and catalog registration through
    /// `add_dynamic_change` — records reducer runtime state before any scheduler
    /// wake-up hint is published, so "unknown to the reducer" cannot describe
    /// work an operator asked for. Treating unknown as admissible would let any
    /// ID that reaches the wake-up channel be resolved from the catalog and
    /// executed without intent, which is the bypass this predicate exists to
    /// close. Catalog membership alone still is not eligibility: a registered
    /// change stays `QueueIntent::NotQueued` until an explicit command.
    pub fn is_ordinary_queue_eligible(&self, change_id: &str) -> bool {
        match self.change_runtime.get(change_id) {
            None => false,
            Some(rt) => {
                !rt.is_terminal()
                    && !rt.dequeued
                    && (rt.is_active() || matches!(rt.queue_intent, QueueIntent::Queued))
            }
        }
    }

    /// Set form of [`Self::is_ordinary_queue_eligible`].
    ///
    /// A scheduler evaluation decides hint admission, queue reconciliation, and
    /// queue classification for many IDs at once. Capturing the eligible set from
    /// one read keeps every one of those decisions on the same reducer revision;
    /// an ID missing from the set is exactly the `false` the per-ID predicate
    /// returns, including for an ID the reducer has never seen.
    pub fn ordinary_queue_eligible_change_ids(&self) -> HashSet<String> {
        self.change_runtime
            .keys()
            .filter(|id| self.is_ordinary_queue_eligible(id))
            .cloned()
            .collect()
    }

    /// Set form of [`Self::is_final_terminal_dispatch_stop`].
    pub fn final_terminal_dispatch_stop_change_ids(&self) -> HashSet<String> {
        self.change_runtime
            .keys()
            .filter(|id| self.is_final_terminal_dispatch_stop(id))
            .cloned()
            .collect()
    }

    /// Change IDs whose admitted lifecycle has settled.
    ///
    /// The scheduler's concurrency accounting gives one admitted change one slot
    /// from workspace admission through merge and resolve settlement, so it needs
    /// a single vocabulary for "this lifecycle is over": merged, pushed,
    /// rejected, a terminal error awaiting explicit retry, and an explicit
    /// stop/dequeue are all settlements that release the slot.
    ///
    /// Deliberately positive evidence only. Absence of wait or queue intent is
    /// not settlement — a change between archive and background merge carries
    /// neither — so nothing here is derived from what a row is *not*.
    pub fn settled_lifecycle_change_ids(&self) -> HashSet<String> {
        self.change_runtime
            .iter()
            .filter(|(_, rt)| rt.is_terminal())
            .map(|(id, _)| id.clone())
            .collect()
    }

    /// Set form of [`Self::is_terminal_error_change`].
    pub fn terminal_error_change_ids(&self) -> HashSet<String> {
        self.change_runtime
            .iter()
            .filter(|(_, rt)| matches!(rt.terminal, TerminalState::Error(_)))
            .map(|(id, _)| id.clone())
            .collect()
    }

    /// Active change IDs whose display status is `resolving`.
    pub fn resolving_change_ids(&self) -> HashSet<String> {
        self.active_change_ids()
            .into_iter()
            .filter(|id| self.display_status(id) == "resolving")
            .collect()
    }

    /// Return true when a recoverable terminal error is currently gating ordinary apply dispatch.
    pub fn is_terminal_error_change(&self, change_id: &str) -> bool {
        self.change_runtime
            .get(change_id)
            .map(|rt| matches!(rt.terminal, TerminalState::Error(_)))
            .unwrap_or(false)
    }

    /// Return true when the *only* thing holding `change_id` back is an operator stop.
    ///
    /// Read-only, and the authority Start admission classifies a preserved mark
    /// against. Every clause is a separate refusal rather than a spelling of
    /// `display_status == "stopped"`:
    ///
    /// - the terminal outcome is `Stopped`, so Error, rejected, merged, and
    ///   pushed evidence keeps its own explicit route and is never resumed here;
    /// - queue intent is the `NotQueued` the stop itself produced, so a row an
    ///   unrelated path already re-admitted is not resumed a second time;
    /// - activity is idle and no wait, dependency hold, or stall is recorded
    ///   underneath the stop, so the stop is the whole reason the row is parked.
    ///
    /// A change the reducer has never seen carries no stop evidence either, and
    /// is not resumable.
    pub fn is_resumable_stopped(&self, change_id: &str) -> bool {
        self.change_runtime
            .get(change_id)
            .map(|rt| {
                matches!(rt.terminal, TerminalState::Stopped)
                    && matches!(rt.queue_intent, QueueIntent::NotQueued)
                    && matches!(rt.activity, ActivityState::Idle)
                    && matches!(rt.wait_state, WaitState::None)
            })
            .unwrap_or(false)
    }

    /// Clear an operator stop and restore ordinary queued intent.
    ///
    /// The only reducer-owned transition that may turn `TerminalState::Stopped`
    /// back into ordinary dispatch eligibility, and it moves exactly the state
    /// the stop itself produced: the terminal classification, the dequeue guard
    /// that keeps late events from reactivating the row, and the queue intent
    /// the stop cleared. Any other terminal outcome is refused as a no-op, so a
    /// resume can never erase Error, rejected, merged, or pushed evidence.
    ///
    /// Deliberately narrower than [`Self::retry_terminal_error`]: a stop is not
    /// a failure, so the change's error history and its retained Apply
    /// iteration-limit diagnostic explain something this command did not
    /// consume and are left exactly as they were. Scheduler-owned stall
    /// membership *is* cleared, because it is run-scoped bookkeeping that would
    /// otherwise keep skipping a row the operator just resumed.
    pub fn resume_stopped_change(&mut self, change_id: &str) -> ReduceOutcome {
        if !self.is_resumable_stopped(change_id) {
            return ReduceOutcome::NoOp;
        }
        {
            let rt = self.runtime_entry(change_id);
            rt.terminal = TerminalState::None;
            rt.clear_activity_wait_and_blocker();
            rt.queue_intent = QueueIntent::Queued;
            rt.dequeued = false;
            rt.observation = WorkspaceObservation::None;
            rt.commit_phase_attempt = None;
        }
        self.clear_stalled_change(change_id);
        self.clear_base_mutating_wait_queues(change_id);
        self.add_dynamic_change(change_id.to_string());
        ReduceOutcome::Changed(ReducerEffect::QueueIntentSet {
            change_id: change_id.to_string(),
            intent: QueueIntent::Queued,
        })
    }

    /// Clear a recoverable terminal error and restore ordinary queued retry intent.
    ///
    /// This is the only reducer-owned transition that may turn `TerminalState::Error`
    /// back into apply-dispatch eligibility. Final terminal states remain immutable.
    ///
    /// Reached only from `ReducerCommand::RetryError`, which only explicit retry
    /// intent — individual retry, bulk retry, retry-class Start, or the explicit
    /// per-target terminal-error queue alias — ever submits. That is what makes
    /// this the right place to consume the change's retained Apply
    /// iteration-limit diagnostic: the record stays observable for exactly as
    /// long as the terminal error it explains, and no automatic path can retire
    /// either one.
    pub fn retry_terminal_error(&mut self, change_id: &str) -> ReduceOutcome {
        {
            let rt = self.runtime_entry(change_id);
            if !matches!(rt.terminal, TerminalState::Error(_)) {
                return ReduceOutcome::NoOp;
            }
            rt.terminal = TerminalState::None;
            rt.clear_activity_wait_and_blocker();
            rt.queue_intent = QueueIntent::Queued;
            rt.dequeued = false;
            rt.observation = WorkspaceObservation::None;
        }
        self.clear_apply_iteration_limit(change_id);
        self.clear_stalled_change(change_id);
        self.clear_error_history(change_id);
        self.clear_base_mutating_wait_queues(change_id);
        self.add_dynamic_change(change_id.to_string());
        ReduceOutcome::Changed(ReducerEffect::QueueIntentSet {
            change_id: change_id.to_string(),
            intent: QueueIntent::Queued,
        })
    }

    // -----------------------------------------------------------------------
    // Reducer API (Phase 2)
    // -----------------------------------------------------------------------

    fn transition_change_to_dequeued(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        rt.terminal = TerminalState::None;
        rt.clear_activity_wait_and_blocker();
        rt.queue_intent = QueueIntent::NotQueued;
        rt.dequeued = true;
        // Cancellation can arrive anywhere inside finalization, including
        // before any terminal commit-phase event exists to observe, so the
        // ephemeral `[commit]` presentation is cleared here rather than being
        // left for a sequence that may never complete.
        rt.commit_phase_attempt = None;
        self.clear_base_mutating_wait_queues(change_id);
    }

    /// Reconcile reducer-owned transient state at a run's terminal stop boundary.
    ///
    /// The scheduler has already reached its cancellation-safe barrier by the
    /// time process-level `Stopped` is reduced, so any row still carrying
    /// activity, queue intent, or a wait/hold was interrupted by that stop and
    /// no longer has an owner. Each such row returns to the same non-terminal
    /// idle, not-queued shape an explicit dequeue produces, including the
    /// process-local dequeue guard that keeps a late lifecycle event or a
    /// same-process workspace observation from reactivating it before an
    /// explicit requeue.
    ///
    /// Deliberately *not* a per-change terminal outcome: a process stop is not a
    /// change result. Fresh idle rows were never part of the stopped run, and an
    /// existing terminal outcome — including recoverable `Error` — is a fact the
    /// stop must not erase.
    fn on_run_stopped(&mut self) {
        let interrupted: Vec<String> = self
            .change_runtime
            .iter()
            .filter(|(_, rt)| {
                !rt.is_terminal()
                    && (rt.is_active()
                        || matches!(rt.queue_intent, QueueIntent::Queued)
                        || !matches!(rt.wait_state, WaitState::None))
            })
            .map(|(change_id, _)| change_id.clone())
            .collect();

        for change_id in interrupted {
            self.transition_change_to_dequeued(&change_id);
            // Scheduler-owned stall/skip membership belongs to the run that just
            // ended; the retry budget it gates is re-derived from the workspace.
            self.stalled_change_ids.remove(&change_id);
            self.skipped_change_ids.remove(&change_id);
            if self.current_change_id.as_deref() == Some(change_id.as_str()) {
                self.current_change_id = None;
            }
        }
    }

    fn transition_change_to_stopped(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        rt.transition_to_terminal(TerminalState::Stopped);
        rt.queue_intent = QueueIntent::NotQueued;
        rt.commit_phase_attempt = None;
        self.clear_base_mutating_wait_queues(change_id);
    }

    fn transition_change_to_error(&mut self, change_id: &str, error: String) {
        let rt = self.runtime_entry(change_id);
        if !rt.is_terminal() {
            rt.transition_to_terminal(TerminalState::Error(error));
        }
        // Unconditional: a row that was already terminal must not keep
        // rendering a commit subphase either.
        rt.commit_phase_attempt = None;
        self.clear_base_mutating_wait_queues(change_id);
    }

    fn transition_change_to_merged(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        rt.transition_to_terminal(TerminalState::Merged);
        rt.queue_intent = QueueIntent::NotQueued;
        self.clear_base_mutating_wait_queues(change_id);
    }

    fn transition_change_to_stalled(
        &mut self,
        change_id: &str,
        blocker_reason: impl Into<String>,
        unblock_metadata: impl Into<String>,
        worktree_snapshot: impl Into<String>,
    ) {
        self.runtime_entry(change_id).transition_to_stalled(
            blocker_reason,
            unblock_metadata,
            worktree_snapshot,
        );
    }

    fn clear_recoverable_terminal_for_success(rt: &mut ChangeRuntimeState) -> bool {
        if rt.can_success_supersede_terminal() {
            rt.terminal = TerminalState::None;
            rt.clear_activity_wait_and_blocker();
            true
        } else {
            false
        }
    }

    fn apply_add_to_queue_command(&mut self, change_id: String) -> ReduceOutcome {
        self.clear_stalled_change(&change_id);
        {
            let rt = self.runtime_entry(&change_id);
            if rt.is_terminal() {
                return ReduceOutcome::NoOp;
            }
            // An explicit operator retry always releases a blocked/stalled hold
            // so the held phase can run again. The new execution result — not
            // the preserved metadata — is what reclassifies the change.
            if !rt.dequeued
                && !matches!(
                    rt.wait_state,
                    WaitState::Stalled | WaitState::ExternalBlocked
                )
                && (rt.is_active() || rt.queue_intent == QueueIntent::Queued)
            {
                return ReduceOutcome::NoOp;
            }
            rt.dequeued = false;
            rt.queue_intent = QueueIntent::Queued;
            rt.wait_state = WaitState::None;
            rt.clear_blocked_metadata();
        }
        self.clear_base_mutating_wait_queues(&change_id);
        self.add_dynamic_change(change_id.clone());
        ReduceOutcome::Changed(ReducerEffect::QueueIntentSet {
            change_id,
            intent: QueueIntent::Queued,
        })
    }

    fn apply_remove_from_queue_command(&mut self, change_id: String) -> ReduceOutcome {
        let rt = self.runtime_entry(&change_id);
        if rt.queue_intent == QueueIntent::NotQueued {
            return ReduceOutcome::NoOp;
        }
        rt.queue_intent = QueueIntent::NotQueued;
        ReduceOutcome::Changed(ReducerEffect::QueueIntentSet {
            change_id,
            intent: QueueIntent::NotQueued,
        })
    }

    fn apply_resolve_merge_command(&mut self, change_id: String) -> ReduceOutcome {
        {
            let rt = self.runtime_entry(&change_id);
            if !matches!(rt.wait_state, WaitState::MergeWait | WaitState::ResolveWait) {
                // TUI/manual resolve commands are already gated by a visible
                // `merge wait` row. In the observed stuck case the visible row
                // can be reconstructed from archive-complete workspace evidence
                // while the reducer still shows the row as idle. Treat explicit
                // operator retry there as scheduler-owned merge retry intent, but
                // continue to reject permanent terminal states such as Merged or
                // Rejected so stale clicks cannot reintroduce resolve work.
                if matches!(rt.activity, ActivityState::Idle)
                    && matches!(rt.terminal, TerminalState::None)
                {
                    rt.terminal = TerminalState::None;
                    rt.wait_state = WaitState::MergeWait;
                } else {
                    return ReduceOutcome::NoOp;
                }
            }
            rt.terminal = TerminalState::None;
            rt.activity = ActivityState::Idle;
            rt.wait_state = WaitState::ResolveWait;
            rt.queue_intent = QueueIntent::NotQueued;
            // Explicit operator intent is the only source of this permission.
            // Auto-resumable deferrals re-enter `ResolveWait` without it, so
            // they keep the unchanged generic preflight.
            rt.manual_resolve_retry = true;
            rt.clear_blocked_metadata();
        }
        self.remove_from_reject_wait_queue(&change_id);
        self.enqueue_unique_resolve_wait(&change_id);
        ReduceOutcome::Changed(ReducerEffect::WaitStateSet {
            change_id,
            wait: WaitState::ResolveWait,
        })
    }

    fn apply_dequeue_change_command(&mut self, change_id: String) -> ReduceOutcome {
        let rt = self.runtime_entry(&change_id);
        if matches!(
            rt.terminal,
            TerminalState::Merged | TerminalState::Pushed | TerminalState::Rejected(_)
        ) {
            return ReduceOutcome::NoOp;
        }
        self.transition_change_to_dequeued(&change_id);
        ReduceOutcome::Changed(ReducerEffect::QueueIntentSet {
            change_id,
            intent: QueueIntent::NotQueued,
        })
    }

    fn apply_stop_change_command(&mut self, change_id: String) -> ReduceOutcome {
        let rt = self.runtime_entry(&change_id);
        if rt.is_terminal() {
            return ReduceOutcome::NoOp;
        }
        self.transition_change_to_stopped(&change_id);
        ReduceOutcome::Changed(ReducerEffect::TerminalStateSet {
            change_id,
            terminal: TerminalState::Stopped,
        })
    }

    /// Apply a reducer command that expresses user intent (queue add/remove, resolve, stop).
    ///
    /// Returns the resulting `ReduceOutcome` describing what changed.
    pub fn apply_command(&mut self, cmd: ReducerCommand) -> ReduceOutcome {
        match cmd {
            ReducerCommand::AddToQueue(change_id) => self.apply_add_to_queue_command(change_id),
            ReducerCommand::RetryError(change_id) => self.retry_terminal_error(&change_id),
            ReducerCommand::RemoveFromQueue(change_id) => {
                self.apply_remove_from_queue_command(change_id)
            }
            ReducerCommand::ResolveMerge(change_id) => self.apply_resolve_merge_command(change_id),
            ReducerCommand::DequeueChange(change_id) => {
                self.apply_dequeue_change_command(change_id)
            }
            ReducerCommand::StopChange(change_id) => self.apply_stop_change_command(change_id),
            ReducerCommand::ResumeStopped(change_id) => self.resume_stopped_change(&change_id),
        }
    }

    /// Apply a workspace observation to reconcile wait states without overwriting
    /// active execution (Phase 2.4).
    pub fn apply_observation(&mut self, change_id: &str, obs: WorkspaceObservation) {
        // Scheduler-owned base-lane membership is reducer state the observation
        // must not duplicate, even if the runtime entry itself looks idle.
        let scheduler_pending = self.resolve_wait_queue.iter().any(|id| id == change_id)
            || self.reject_wait_queue.iter().any(|id| id == change_id);
        let rt = self.runtime_entry(change_id);

        // Never overwrite an active execution stage.
        if rt.is_active() {
            return;
        }
        // Never overwrite a terminal state.
        if rt.is_terminal() {
            return;
        }

        match obs {
            WorkspaceObservation::WorkspaceArchived => {
                // Restore MergeWait when the reducer already has explicit manual-wait
                // evidence. A bare archived workspace observation is otherwise only a
                // repository milestone: no-blocker archive completion
                // must remain in active merge handling until MergeCompleted or a concrete
                // MergeDeferred event explains manual wait.
                if matches!(rt.wait_state, WaitState::MergeWait) {
                    rt.observation = WorkspaceObservation::WorkspaceArchived;
                } else if rt.is_fresh_idle() && !scheduler_pending {
                    // Startup reconciliation: a fresh process has no reducer history at
                    // all, so archived-but-not-yet-merged workspace evidence is the only
                    // authority for this row. Restoring MergeWait here is what makes the
                    // reducer agree with the row the refresh scan already displays, and
                    // is what manual resolve admission is then evaluated against.
                    rt.wait_state = WaitState::MergeWait;
                    rt.clear_blocked_metadata();
                    rt.observation = WorkspaceObservation::WorkspaceArchived;
                }
            }
            WorkspaceObservation::WorktreeNotAhead => {
                // Clear MergeWait when worktree is no longer ahead.
                if matches!(rt.wait_state, WaitState::MergeWait) {
                    rt.wait_state = WaitState::None;
                    rt.clear_blocked_metadata();
                }
                rt.observation = WorkspaceObservation::WorktreeNotAhead;
            }
            WorkspaceObservation::None => {
                rt.observation = WorkspaceObservation::None;
            }
        }
    }

    fn on_processing_started(&mut self, change_id: &str) {
        if self.is_terminal_error_change(change_id) {
            return;
        }
        self.set_current_change(Some(change_id.to_string()));
        let rt = self.runtime_entry(change_id);
        if !rt.is_terminal() && !rt.dequeued {
            rt.queue_intent = QueueIntent::Queued;
        }
    }

    fn on_processing_error(&mut self, change_id: &str, error: String) {
        self.remove_from_pending(change_id);
        let rt = self.runtime_entry(change_id);
        // A failure raised between `ApplyCommitPhase::Started` and any terminal
        // commit-phase event — a WIP snapshot error, for instance — surfaces
        // only as a processing error, so this is the only place that can clear
        // the presentation it left behind.
        rt.commit_phase_attempt = None;
        if !rt.is_terminal() && !rt.dequeued {
            rt.transition_to_terminal(TerminalState::Error(error));
        }
    }

    /// Enter the ephemeral workspace-preparation activity.
    ///
    /// Only an idle, non-terminal, non-dequeued change may enter it: a change
    /// that is already running an operation must never be relabelled as if no
    /// agent had started, and a stopped or terminal change must never be
    /// reactivated by a late preparation event.
    fn on_workspace_preparation_started(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        if rt.is_terminal() || rt.dequeued || !matches!(rt.activity, ActivityState::Idle) {
            return;
        }
        rt.activity = ActivityState::Preparing;
        rt.wait_state = WaitState::None;
        rt.clear_blocked_metadata();
    }

    /// Leave workspace preparation without claiming any next phase.
    ///
    /// Deliberately narrow: it only undoes `Preparing`. Once another transition
    /// has taken over — an operation started, an error, a stop — this is a no-op,
    /// so a late or duplicate clear can never regress real state.
    fn on_workspace_preparation_ended(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        if matches!(rt.activity, ActivityState::Preparing) {
            rt.activity = ActivityState::Idle;
        }
    }

    fn on_apply_started(&mut self, change_id: &str) {
        let mut should_start = false;
        {
            let rt = self.runtime_entry(change_id);
            if !rt.is_terminal() && !rt.dequeued {
                rt.activity = ActivityState::Applying;
                rt.wait_state = WaitState::None;
                rt.clear_blocked_metadata();
                should_start = true;
            }
            // A repair iteration always restores apply presentation, including
            // on a row this reducer declined to reactivate: stale `[commit]`
            // must never outlive the finalization that set it.
            rt.commit_phase_attempt = None;
        }
        if should_start {
            self.set_current_change(Some(change_id.to_string()));
        }
    }

    fn on_apply_completed(&mut self, change_id: &str) {
        self.increment_apply_count(change_id);
        let rt = self.runtime_entry(change_id);
        rt.commit_phase_attempt = None;
        if matches!(rt.activity, ActivityState::Applying) {
            rt.activity = ActivityState::Idle;
        }
    }

    /// Record the ephemeral final-commit presentation phase.
    ///
    /// Nothing about the canonical lifecycle moves here: the change stays in
    /// whatever activity it was already in. Only a `Started` phase leaves
    /// presentation active, so completion, failure, and cancellation all clear
    /// it, and a terminal or dequeued row never adopts it at all.
    fn on_apply_commit_phase(
        &mut self,
        change_id: &str,
        phase: crate::events::ApplyCommitPhase,
        attempt: u32,
    ) {
        let rt = self.runtime_entry(change_id);
        rt.commit_phase_attempt = match phase.is_active() && !rt.is_terminal() && !rt.dequeued {
            true => Some(attempt),
            false => None,
        };
    }

    fn on_acceptance_started(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        if !rt.is_terminal() && !rt.dequeued {
            rt.activity = ActivityState::Accepting;
        }
    }

    fn on_acceptance_completed(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        if matches!(rt.activity, ActivityState::Accepting) {
            rt.activity = ActivityState::Idle;
        }
    }

    fn on_change_rejected(&mut self, change_id: &str, reason: String) {
        self.remove_from_pending(change_id);
        self.remove_from_reject_wait_queue(change_id);
        let rt = self.runtime_entry(change_id);
        if !rt.is_terminal() {
            rt.transition_to_terminal(TerminalState::Rejected(reason));
            rt.queue_intent = QueueIntent::NotQueued;
        }
    }

    fn on_rejection_review_completed(
        &mut self,
        change_id: &str,
        outcome: crate::events::RejectionOutcome,
    ) {
        if matches!(outcome, crate::events::RejectionOutcome::Confirm) {
            self.remove_from_pending(change_id);
        }
        self.remove_from_reject_wait_queue(change_id);

        let rt = self.runtime_entry(change_id);
        if rt.is_terminal() || rt.dequeued {
            return;
        }
        match outcome {
            crate::events::RejectionOutcome::Confirm => {
                rt.transition_to_terminal(TerminalState::Rejected(
                    "rejecting review confirmed rejection".to_string(),
                ));
                rt.queue_intent = QueueIntent::NotQueued;
            }
            crate::events::RejectionOutcome::Resume => {
                rt.activity = ActivityState::Applying;
                rt.wait_state = WaitState::None;
                rt.clear_blocked_metadata();
                rt.terminal = TerminalState::None;
            }
            crate::events::RejectionOutcome::Block => {
                rt.transition_to_stalled(
                    "rejection review returned block; unresolved blocker remains",
                    "resolve unresolved blocker tasks in openspec/changes/<change_id>/tasks.md, then trigger explicit resume",
                    "existing worktree and WIP context are preserved for stalled rejection review",
                );
            }
        }
    }

    fn on_rejection_review_failed(&mut self, change_id: &str, error: String) {
        self.remove_from_pending(change_id);
        self.remove_from_reject_wait_queue(change_id);
        let rt = self.runtime_entry(change_id);
        if !rt.is_terminal() && !rt.dequeued {
            rt.transition_to_terminal(TerminalState::Error(error));
        }
    }

    fn on_workspace_status_updated(
        &mut self,
        change_id: &str,
        status: &crate::vcs::WorkspaceStatus,
    ) {
        if !self
            .change_runtime
            .get(change_id)
            .is_some_and(|rt| !rt.is_terminal() && !rt.dequeued)
        {
            return;
        }
        match status {
            // Monotonic blocker precedence: a structured classification already
            // owns this wait, so the lower-fidelity workspace observation is
            // idempotent. Both structured transitions already left the change
            // idle, so confirming "still blocked" requires no mutation at all —
            // and any mutation here would downgrade the wait kind and replace
            // validated metadata with generic strings.
            crate::vcs::WorkspaceStatus::Blocked
                if self
                    .change_runtime
                    .get(change_id)
                    .is_some_and(ChangeRuntimeState::has_structured_blocker_hold) =>
            {
                if let Some(rt) = self.change_runtime.get(change_id) {
                    tracing::debug!(
                        change_id = %change_id,
                        current_status = rt.display_status(),
                        blocker_kind = ?rt.blocker_kind(),
                        "Ignoring generic blocked workspace status because a structured blocker hold owns this wait"
                    );
                }
            }
            crate::vcs::WorkspaceStatus::Blocked if self.stalled_change_ids.contains(change_id) => {
                self.transition_change_to_stalled(
                    change_id,
                    "change stalled with recoverable blocker",
                    "resolve blocker evidence and retry explicitly",
                    "workflow state is preserved in the workspace",
                );
            }
            crate::vcs::WorkspaceStatus::Rejecting => {
                self.mark_reject_wait(change_id);
            }
            crate::vcs::WorkspaceStatus::Applying => {
                self.runtime_entry(change_id).activity = ActivityState::Applying;
            }
            crate::vcs::WorkspaceStatus::Accepting => {
                self.runtime_entry(change_id).activity = ActivityState::Accepting;
            }
            crate::vcs::WorkspaceStatus::Blocked => {
                self.transition_change_to_stalled(
                    change_id,
                    "apply reported recoverable blocker; workspace remains stalled",
                    "resolve implementation blocker section and pending unblock tasks before explicit retry",
                    "existing worktree and WIP context are preserved while stalled",
                );
            }
            crate::vcs::WorkspaceStatus::Archiving => {
                self.runtime_entry(change_id).activity = ActivityState::Archiving;
            }
            crate::vcs::WorkspaceStatus::Resolving => {
                self.on_workspace_status_resolving(change_id);
            }
            crate::vcs::WorkspaceStatus::MergeWait => {
                self.on_workspace_status_merge_wait(change_id);
            }
            _ => {}
        }
    }

    fn on_workspace_status_resolving(&mut self, change_id: &str) {
        let has_other_lane_blocker = self.has_other_post_archive_lane_blocker(change_id);
        let rt = self.runtime_entry(change_id);
        if has_other_lane_blocker {
            rt.activity = ActivityState::Idle;
            rt.wait_state = WaitState::ResolveWait;
            self.enqueue_unique_resolve_wait(change_id);
        } else {
            rt.activity = ActivityState::Resolving;
            rt.wait_state = WaitState::None;
            self.remove_from_resolve_wait_queue(change_id);
        }
    }

    fn on_workspace_status_merge_wait(&mut self, change_id: &str) {
        let rt = self.runtime_entry(change_id);
        // WorkspaceStatus::MergeWait is manual-blocker evidence only when
        // the reducer does not already own a stronger active/pending state.
        // Auto-resumable deferrals are represented by MergeDeferred(true)
        // and scheduler ResolveWait; stale workspace display hints must not
        // downgrade that retry intent to manual merge wait.
        if matches!(
            rt.activity,
            ActivityState::Resolving | ActivityState::Rejecting
        ) || matches!(
            rt.wait_state,
            WaitState::ResolveWait | WaitState::RejectWait
        ) {
            tracing::debug!(
                change_id = %change_id,
                current_status = rt.display_status(),
                "Ignoring workspace MergeWait status because reducer owns stronger post-archive state"
            );
        } else {
            rt.activity = ActivityState::Idle;
            rt.wait_state = WaitState::MergeWait;
            rt.queue_intent = QueueIntent::NotQueued;
            self.remove_from_resolve_wait_queue(change_id);
        }
    }

    /// Apply an ExecutionEvent to update the shared state.
    ///
    /// This is the single source of truth for state mutations driven by execution events.
    ///
    /// ## Current Usage
    ///
    /// - **CLI Orchestrator**: Calls this method in `src/orchestrator.rs` to track:
    ///   - `ProcessingStarted` - When a change begins processing
    ///   - `ApplyStarted` - When apply operation starts
    ///   - `ApplyCompleted` - When apply operation completes (increments apply count)
    ///   - `ChangeArchived` - When a change is successfully archived
    ///
    /// - **TUI/Web**: Currently maintain their own ExecutionEvent-driven state independently.
    ///   Future refactoring can make them query this shared state for unified tracking.
    pub fn apply_execution_event(&mut self, event: &crate::events::ExecutionEvent) {
        use crate::events::ExecutionEvent;

        match event {
            // Processing lifecycle
            ExecutionEvent::ProcessingStarted(change_id) => self.on_processing_started(change_id),
            ExecutionEvent::ProcessingError { id, error } => {
                self.on_processing_error(id, error.clone());
            }

            // Apply events
            ExecutionEvent::ApplyStarted { change_id, .. } => self.on_apply_started(change_id),
            ExecutionEvent::ApplyCompleted { change_id, .. } => self.on_apply_completed(change_id),
            ExecutionEvent::ApplyFailed { change_id, error } => {
                self.runtime_entry(change_id).commit_phase_attempt = None;
                self.remove_from_pending(change_id);
                self.transition_change_to_error(change_id, error.clone());
            }
            ExecutionEvent::ApplyCommitPhase {
                change_id,
                phase,
                attempt,
            } => self.on_apply_commit_phase(change_id, *phase, *attempt),

            // Acceptance events
            ExecutionEvent::AcceptanceStarted { change_id, .. } => {
                self.on_acceptance_started(change_id);
            }
            ExecutionEvent::AcceptanceCompleted { change_id } => {
                self.on_acceptance_completed(change_id);
            }
            ExecutionEvent::AcceptanceFailed { change_id, error } => {
                self.transition_change_to_error(change_id, error.clone());
            }
            ExecutionEvent::ChangeRejected { change_id, reason } => {
                self.on_change_rejected(change_id, reason.clone());
            }
            ExecutionEvent::RejectionReviewCompleted { change_id, outcome } => {
                self.on_rejection_review_completed(change_id, *outcome);
            }
            ExecutionEvent::RejectionReviewFailed { change_id, error } => {
                self.on_rejection_review_failed(change_id, error.clone());
            }

            // Workspace preparation (process-local only)
            ExecutionEvent::WorkspacePreparationStarted { change_id } => {
                self.on_workspace_preparation_started(change_id);
            }
            ExecutionEvent::WorkspacePreparationEnded { change_id } => {
                self.on_workspace_preparation_ended(change_id);
            }

            // Workspace status synchronization events
            ExecutionEvent::WorkspaceStatusUpdated {
                change_id, status, ..
            } => {
                self.on_workspace_status_updated(change_id, status);
            }

            // Archive events
            ExecutionEvent::ArchiveStarted { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() && !rt.dequeued {
                    rt.activity = ActivityState::Archiving;
                }
            }
            ExecutionEvent::ArchiveResumed { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if matches!(rt.terminal, TerminalState::Error(_)) {
                    rt.terminal = TerminalState::None;
                }
                if !rt.is_terminal() && !rt.dequeued {
                    rt.activity = ActivityState::Archiving;
                    rt.wait_state = WaitState::None;
                    rt.clear_blocked_metadata();
                }
            }
            ExecutionEvent::ArchiveRetryScheduled { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() {
                    rt.activity = ActivityState::Archiving;
                }
            }
            ExecutionEvent::ChangeArchived(change_id) => {
                self.mark_archived(change_id);
                let has_other_resolve_lane_blocker =
                    self.change_runtime.iter().any(|(id, runtime)| {
                        id != change_id
                            && matches!(
                                runtime.activity,
                                ActivityState::Resolving | ActivityState::Rejecting
                            )
                            && !runtime.is_terminal()
                    });
                let rt = self.runtime_entry(change_id);
                if Self::clear_recoverable_terminal_for_success(rt) {
                    // Archive is never terminal on its own: it always enters
                    // post-archive handling.
                    rt.queue_intent = QueueIntent::NotQueued;
                    if has_other_resolve_lane_blocker {
                        // Lane occupied: keep this row scheduler-owned for automatic retry
                        // after the active merge/reject lane clears.
                        rt.activity = ActivityState::Idle;
                        rt.wait_state = WaitState::ResolveWait;
                        self.enqueue_unique_resolve_wait(change_id);
                    } else {
                        // No blocker known yet: make active merge handling truthful in the
                        // reducer. Manual MergeWait is set only by a later concrete
                        // MergeDeferred(auto_resumable=false) event from merge readiness.
                        rt.activity = ActivityState::Resolving;
                        rt.wait_state = WaitState::None;
                        self.remove_from_resolve_wait_queue(change_id);
                    }
                }
            }
            ExecutionEvent::ArchiveFailed {
                change_id, error, ..
            } => {
                self.transition_change_to_error(change_id, error.clone());
            }

            // Merge / resolve events
            ExecutionEvent::MergeDeferred {
                change_id,
                auto_resumable,
                ..
            } => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() {
                    if *auto_resumable {
                        // Auto-resumable: will be retried after a preceding merge/resolve
                        // completes. Use ResolveWait so workspace-refresh reconciliation
                        // does not regress it back to MergeWait. Do not interrupt an
                        // already-active merge attempt; that active state remains the
                        // truthful visible state until completion or manual deferral.
                        if !rt.is_active() {
                            rt.wait_state = WaitState::ResolveWait;
                        }
                        self.enqueue_unique_resolve_wait(change_id);
                    } else {
                        // Manual intervention required: the change must remain visible
                        // as merge-wait only, not as ordinary queued work. Manual
                        // deferral can be discovered during active post-archive merge
                        // handling, so clear Resolving and surface the concrete blocker.
                        rt.activity = ActivityState::Idle;
                        rt.wait_state = WaitState::MergeWait;
                        rt.queue_intent = QueueIntent::NotQueued;
                        self.remove_from_resolve_wait_queue(change_id);
                    }
                }
            }
            ExecutionEvent::MergeCompleted { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if rt.can_success_supersede_terminal() {
                    self.transition_change_to_merged(change_id);
                }
            }
            ExecutionEvent::PushStarted { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() && !rt.dequeued {
                    rt.activity = ActivityState::Resolving;
                    rt.wait_state = WaitState::None;
                }
            }
            ExecutionEvent::PushCompleted { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if rt.can_success_supersede_terminal() {
                    rt.transition_to_terminal(TerminalState::Pushed);
                    rt.queue_intent = QueueIntent::NotQueued;
                    // Confirmed publication ends this change's base-lane work, so
                    // any retry intent recorded while it was publishing must not
                    // survive as ordinary merge/resolve work.
                    self.clear_base_mutating_wait_queues(change_id);
                }
            }
            ExecutionEvent::PushFailed {
                change_id, error, ..
            } => {
                self.transition_change_to_error(change_id, error.clone());
            }
            ExecutionEvent::ResolveStarted { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() && !rt.dequeued {
                    rt.activity = ActivityState::Resolving;
                    rt.wait_state = WaitState::None;
                }
            }
            ExecutionEvent::ResolveCompleted { change_id, .. } => {
                let rt = self.runtime_entry(change_id);
                if rt.can_success_supersede_terminal()
                    && !rt.dequeued
                    && (matches!(rt.terminal, TerminalState::Error(_))
                        || matches!(rt.activity, ActivityState::Resolving)
                        || matches!(rt.wait_state, WaitState::ResolveWait))
                {
                    // Successful resolve means the change is now merged. Setting terminal
                    // prevents a subsequent ChangesRefreshed from resurrecting ResolveWait
                    // via apply_observation (which skips terminal entries).
                    self.transition_change_to_merged(change_id);
                } else {
                    self.clear_base_mutating_wait_queues(change_id);
                }
            }
            ExecutionEvent::ResolveFailed { change_id, .. } => {
                // Resolve failure does NOT regress a terminal state.
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() {
                    // Reset activity if we were actively resolving.
                    if matches!(rt.activity, ActivityState::Resolving) {
                        rt.activity = ActivityState::Idle;
                    }
                    // Restore MergeWait when the failure happened during Resolving,
                    // or before ResolveStarted was sent (early dirty-base check
                    // while still in ResolveWait from ReducerCommand::ResolveMerge).
                    if matches!(rt.activity, ActivityState::Idle)
                        && matches!(rt.wait_state, WaitState::ResolveWait | WaitState::None)
                    {
                        rt.wait_state = WaitState::MergeWait;
                        self.remove_from_resolve_wait_queue(change_id);
                    }
                }
            }

            // Dependency events
            ExecutionEvent::DependencyBlocked {
                change_id,
                dependency_ids,
            } => {
                let ids = dependency_ids.clone();
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() && !rt.is_active() {
                    rt.reconcile_dependency_blocker(&ids);
                }
            }
            ExecutionEvent::DependencyResolved { change_id } => {
                let rt = self.runtime_entry(change_id);
                if matches!(rt.wait_state, WaitState::DependencyBlocked) {
                    rt.wait_state = WaitState::None;
                    rt.clear_blocked_metadata();
                }
            }
            ExecutionEvent::AcceptanceGated { change_id, blocker } => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() && !rt.dequeued {
                    // The agent reported facts; the orchestrator decides the
                    // lifecycle. A complete, validated payload becomes external
                    // `blocked`; anything weaker stays an Acceptance-owned
                    // `stalled` hold rather than an inferred external wait.
                    let classification =
                        crate::orchestration::blocker_classification::classify_reported_facts(
                            blocker,
                        );
                    tracing::debug!(
                        change_id = %change_id,
                        reported_category = %blocker.category,
                        classified_as = ?classification.display_status(),
                        "classified acceptance blocker facts"
                    );
                    match classification {
                        crate::orchestration::blocker_classification::LifecycleClassification::ExternalBlocked(info) => {
                            rt.transition_to_external_blocked(&info, blocker.worktree_snapshot());
                        }
                        crate::orchestration::blocker_classification::LifecycleClassification::Stalled { detail, .. } => {
                            rt.transition_to_acceptance_stalled(
                                format!("acceptance-gated:{}", blocker.category),
                                detail,
                                blocker.worktree_snapshot(),
                                blocker.resumable,
                            );
                        }
                        crate::orchestration::blocker_classification::LifecycleClassification::ProtocolCorrection { .. } => {}
                    }
                }
            }
            ExecutionEvent::ExecutionBlocked { change_id, blocker } => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() && !rt.dequeued {
                    use crate::orchestration::blocker_classification as classification;
                    // A repeated permission/tool-policy denial is an execution
                    // stop, not a wait on a named prerequisite, so it is
                    // classified as one rather than being offered to the
                    // external-claim validator at all.
                    let classified = if classification::is_permission_denial(blocker) {
                        classification::classify_execution_stop(
                            classification::ExecutionStopReason::PermissionDenial,
                            blocker.summary(),
                        )
                    } else {
                        classification::classify_reported_facts(blocker)
                    };
                    let blocker_reason = format!("execution-blocked:{}", blocker.category);
                    match classified {
                        crate::orchestration::blocker_classification::LifecycleClassification::ExternalBlocked(info) => {
                            rt.transition_to_external_blocked(&info, blocker.worktree_snapshot());
                        }
                        // An Acceptance-phase execution hold (for example a
                        // repeated permission denial during acceptance) holds
                        // dispatch the same way a validated acceptance blocker
                        // does. Apply-phase holds keep their own conservative
                        // handling.
                        _ if blocker.phase == "acceptance" => {
                            rt.transition_to_acceptance_stalled(
                                blocker_reason,
                                blocker.summary(),
                                blocker.worktree_snapshot(),
                                blocker.resumable,
                            );
                        }
                        _ => {
                            rt.transition_to_stalled(
                                blocker_reason,
                                blocker.summary(),
                                blocker.worktree_snapshot(),
                            );
                        }
                    }
                }
            }

            // The `on_merged` hook owns the merged transition: while it fails the
            // change is not merged, so the reducer records the merge-wait
            // recovery state an operator has to act on. Recording it here is
            // also what makes the recovery an *edge* — a replayed hook failure
            // finds the row already in merge wait and changes nothing, so it
            // cannot revoke an execution mark the operator set in the meantime.
            ExecutionEvent::HookFailed {
                change_id,
                hook_type,
                ..
            } if hook_type == crate::hooks::HookType::OnMerged.config_key() => {
                let rt = self.runtime_entry(change_id);
                if !rt.is_terminal() && !rt.dequeued {
                    rt.activity = ActivityState::Idle;
                    rt.wait_state = WaitState::MergeWait;
                    rt.queue_intent = QueueIntent::NotQueued;
                    self.remove_from_resolve_wait_queue(change_id);
                }
            }

            // Stop/dequeue events
            ExecutionEvent::ChangeDequeued { change_id }
            | ExecutionEvent::ChangeStopped { change_id } => {
                let rt = self.runtime_entry(change_id);
                // `Stopped` is in the guard for the same reason the others are:
                // it is a settled outcome, and the dequeue edge a targeted
                // force-stop publishes after its own terminal settlement must
                // report that settlement rather than undo it back to an idle
                // `not queued` row an observer would keep waiting on.
                if matches!(
                    rt.terminal,
                    TerminalState::Merged
                        | TerminalState::Pushed
                        | TerminalState::Rejected(_)
                        | TerminalState::Stopped
                ) {
                    return;
                }
                self.transition_change_to_dequeued(change_id);
            }

            // Process-level run boundary: the reducer, not a frontend, owns the
            // transition every interrupted row makes when the run ends.
            ExecutionEvent::Stopped => self.on_run_stopped(),

            // Dynamic queue support
            ExecutionEvent::ChangesRefreshed {
                changes,
                merge_wait_ids,
                rejected_changes: _,
                worktree_not_ahead_ids,
                ..
            } => {
                // Refresh the initial snapshot if new changes appeared
                let new_ids: Vec<String> = changes
                    .iter()
                    .filter(|c| {
                        !self.initial_change_ids.contains(&c.id)
                            && !self.archived_changes.contains(&c.id)
                    })
                    .map(|c| c.id.clone())
                    .collect();
                for id in new_ids {
                    self.add_dynamic_change(id);
                }

                // If a previously rejected change reappears in the active listing,
                // treat it as reactivated and reset runtime state to defaults.
                for change in changes {
                    let rt = self.runtime_entry(&change.id);
                    if matches!(rt.terminal, TerminalState::Rejected(_)) {
                        rt.terminal = TerminalState::None;
                        rt.clear_activity_wait_and_blocker();
                        rt.queue_intent = QueueIntent::NotQueued;
                    }
                }

                // Apply workspace observations via the reconcile path.
                let mw: Vec<String> = merge_wait_ids.iter().cloned().collect();
                let nah: Vec<String> = worktree_not_ahead_ids.iter().cloned().collect();
                for id in mw {
                    self.apply_observation(&id.clone(), WorkspaceObservation::WorkspaceArchived);
                }
                for id in nah {
                    self.apply_observation(&id.clone(), WorkspaceObservation::WorktreeNotAhead);
                }
            }

            // Other events don't affect shared state directly
            _ => {}
        }
    }
}

impl Default for OrchestratorState {
    fn default() -> Self {
        Self::new(Vec::new(), 0)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_new_state() {
        let state =
            OrchestratorState::new(vec!["change-a".to_string(), "change-b".to_string()], 10);

        assert_eq!(state.total_changes(), 2);
        assert_eq!(state.remaining_changes(), 2);
        assert_eq!(state.changes_processed(), 0);
        assert_eq!(state.iteration(), 0);
        assert_eq!(state.max_iterations(), 10);
        assert!(state.current_change_id().is_none());
    }

    #[test]
    fn test_is_in_snapshot() {
        let state = OrchestratorState::new(vec!["change-a".to_string()], 0);

        assert!(state.is_in_snapshot("change-a"));
        assert!(!state.is_in_snapshot("change-b"));
    }

    #[test]
    fn test_apply_count_increment() {
        let mut state = OrchestratorState::new(vec!["change-a".to_string()], 0);

        assert_eq!(state.apply_count("change-a"), 0);
        assert_eq!(state.increment_apply_count("change-a"), 1);
        assert_eq!(state.increment_apply_count("change-a"), 2);
        assert_eq!(state.apply_count("change-a"), 2);
    }

    #[test]
    fn test_mark_archived() {
        let mut state =
            OrchestratorState::new(vec!["change-a".to_string(), "change-b".to_string()], 0);
        state.set_current_change(Some("change-a".to_string()));
        state.increment_apply_count("change-a");

        assert!(state.is_pending("change-a"));
        assert!(!state.is_archived("change-a"));
        assert_eq!(state.remaining_changes(), 2);

        state.mark_archived("change-a");

        assert!(!state.is_pending("change-a"));
        assert!(state.is_archived("change-a"));
        assert_eq!(state.remaining_changes(), 1);
        assert_eq!(state.changes_processed(), 1);
        assert!(state.current_change_id().is_none());
        assert_eq!(state.apply_count("change-a"), 0); // Cleared
    }

    #[test]
    fn test_is_complete() {
        let mut state = OrchestratorState::new(vec!["change-a".to_string()], 0);

        assert!(!state.is_complete());
        state.mark_archived("change-a");
        assert!(state.is_complete());
    }

    #[test]
    fn test_iteration_limit() {
        let mut state = OrchestratorState::new(vec![], 10);

        assert!(!state.is_iteration_limit_reached());

        for _ in 0..8 {
            state.increment_iteration();
        }
        assert!(state.is_approaching_iteration_limit()); // 80%
        assert!(!state.is_iteration_limit_reached());

        state.increment_iteration(); // 9
        assert!(!state.is_iteration_limit_reached());

        state.increment_iteration(); // 10
        assert!(state.is_iteration_limit_reached());
    }

    #[test]
    fn test_no_iteration_limit() {
        let mut state = OrchestratorState::new(vec![], 0);

        for _ in 0..100 {
            state.increment_iteration();
        }

        assert!(!state.is_iteration_limit_reached());
        assert!(!state.is_approaching_iteration_limit());
    }

    #[test]
    fn test_add_dynamic_change() {
        let mut state = OrchestratorState::new(vec!["change-a".to_string()], 0);

        assert_eq!(state.total_changes(), 1);
        assert!(!state.is_pending("change-b"));

        state.add_dynamic_change("change-b".to_string());

        assert_eq!(state.total_changes(), 2);
        assert!(state.is_pending("change-b"));
        assert!(state.is_in_snapshot("change-b"));
    }

    #[test]
    fn test_add_dynamic_change_idempotent() {
        let mut state = OrchestratorState::new(vec!["change-a".to_string()], 0);

        state.add_dynamic_change("change-a".to_string()); // Already exists
        state.add_dynamic_change("change-b".to_string());
        state.add_dynamic_change("change-b".to_string()); // Duplicate

        assert_eq!(state.total_changes(), 2); // Not 3 or 4
    }

    #[test]
    fn test_remove_from_pending() {
        let mut state = OrchestratorState::new(vec!["change-a".to_string()], 0);
        state.set_current_change(Some("change-a".to_string()));

        assert!(state.is_pending("change-a"));
        assert!(state.current_change_id().is_some());

        state.remove_from_pending("change-a");

        assert!(!state.is_pending("change-a"));
        assert!(!state.is_archived("change-a")); // Not archived, just removed
        assert!(state.current_change_id().is_none());
    }

    #[test]
    fn test_error_history_management_and_circuit_breaker() {
        let mut state = OrchestratorState::new(vec!["change-a".to_string()], 0);
        let config = CircuitBreakerConfig {
            enabled: true,
            threshold: 2,
        };

        assert!(!state.record_error_and_check_circuit_breaker(
            "change-a",
            "same error",
            config.clone()
        ));
        assert!(state.record_error_and_check_circuit_breaker("change-a", "same error", config));
        assert!(state.last_error("change-a").is_some());

        state.clear_error_history("change-a");
        assert!(state.last_error("change-a").is_none());
    }

    // -----------------------------------------------------------------------
    // Phase 1.2: change_runtime initialisation
    // -----------------------------------------------------------------------

    #[test]
    fn test_orchestrator_state_initializes_change_runtime() {
        let state = OrchestratorState::new(vec!["change-a".to_string(), "change-b".to_string()], 0);

        // Each change must have a runtime entry with the default (not-queued, idle) state.
        let rt_a = state
            .change_runtime("change-a")
            .expect("runtime for change-a");
        assert_eq!(rt_a.queue_intent, QueueIntent::NotQueued);
        assert_eq!(rt_a.activity, ActivityState::Idle);
        assert_eq!(rt_a.wait_state, WaitState::None);
        assert!(matches!(rt_a.terminal, TerminalState::None));

        let rt_b = state
            .change_runtime("change-b")
            .expect("runtime for change-b");
        assert_eq!(rt_b.queue_intent, QueueIntent::NotQueued);
    }

    // -----------------------------------------------------------------------
    // Phase 1.3: invariant helpers
    // -----------------------------------------------------------------------

    #[test]
    fn test_change_runtime_invariants() {
        // Valid: default state.
        let valid = ChangeRuntimeState::default();
        assert!(valid.invariants_hold());

        // Invalid: terminal + active activity.
        let invalid = ChangeRuntimeState {
            terminal: TerminalState::Merged,
            activity: ActivityState::Applying,
            ..Default::default()
        };
        assert!(!invalid.invariants_hold());

        // Invalid: ResolveWait + Resolving.
        let invalid2 = ChangeRuntimeState {
            wait_state: WaitState::ResolveWait,
            activity: ActivityState::Resolving,
            ..Default::default()
        };
        assert!(!invalid2.invariants_hold());

        // Valid: MergeWait + Idle.
        let ok = ChangeRuntimeState {
            wait_state: WaitState::MergeWait,
            ..Default::default()
        };
        assert!(ok.invariants_hold());

        // Rejecting + no terminal outcome is a valid in-flight state while review runs.
        let in_flight_rejecting = ChangeRuntimeState {
            activity: ActivityState::Rejecting,
            terminal: TerminalState::None,
            ..Default::default()
        };
        assert!(in_flight_rejecting.invariants_hold());

        // Invalid: RejectWait + Rejecting.
        let invalid3 = ChangeRuntimeState {
            wait_state: WaitState::RejectWait,
            activity: ActivityState::Rejecting,
            ..Default::default()
        };
        assert!(!invalid3.invariants_hold());
    }

    // -----------------------------------------------------------------------
    // Phase 1.4: display_status derivation
    // -----------------------------------------------------------------------

    #[test]
    fn test_lifecycle_display_distinguishes_rejected_stalled_blocked_and_error() {
        let rejected = ChangeRuntimeState {
            terminal: TerminalState::Rejected("terminal".to_string()),
            ..Default::default()
        };
        assert_eq!(rejected.display_status(), "rejected");

        let stalled = ChangeRuntimeState {
            wait_state: WaitState::Stalled,
            ..Default::default()
        };
        assert_eq!(stalled.display_status(), "stalled");

        let dependency_blocked = ChangeRuntimeState {
            wait_state: WaitState::DependencyBlocked,
            ..Default::default()
        };
        assert_eq!(dependency_blocked.display_status(), "blocked");

        let error = ChangeRuntimeState {
            terminal: TerminalState::Error("repo-fixable failure".to_string()),
            ..Default::default()
        };
        assert_eq!(error.display_status(), "error");
    }

    #[test]
    fn pushed_terminal_status_is_distinct_from_merged() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 1);
        state.apply_execution_event(&crate::events::ExecutionEvent::PushCompleted {
            change_id: "c".to_string(),
            remote: "origin".to_string(),
            branch: "c".to_string(),
        });
        assert_eq!(state.display_status("c"), "pushed");
        assert!(matches!(
            state.runtime_entry("c").terminal,
            TerminalState::Pushed
        ));
    }

    #[test]
    fn test_display_status_derivation() {
        let state = OrchestratorState::new(vec!["c".to_string()], 0);
        // Default is not queued.
        assert_eq!(state.display_status("c"), "not queued");
        assert_eq!(state.display_status("unknown"), "not queued");

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert_eq!(state.display_status("c"), "queued");

        // Applying takes priority over queued.
        let rt = state.runtime_entry("c");
        rt.activity = ActivityState::Applying;
        assert_eq!(state.display_status("c"), "applying");

        // Terminal takes highest priority.
        let rt = state.runtime_entry("c");
        rt.terminal = TerminalState::Merged;
        assert_eq!(state.display_status("c"), "merged");
    }

    #[test]
    fn test_display_color_derivation() {
        let mut rt = ChangeRuntimeState::default();
        assert_eq!(rt.display_color(), ratatui::style::Color::DarkGray);

        rt.queue_intent = QueueIntent::Queued;
        assert_eq!(rt.display_color(), ratatui::style::Color::Yellow);

        rt.wait_state = WaitState::DependencyBlocked;
        assert_eq!(rt.display_color(), ratatui::style::Color::Gray);

        rt.activity = ActivityState::Preparing;
        assert_eq!(rt.display_status(), "preparing");
        assert_eq!(rt.display_color(), ratatui::style::Color::Green);

        rt.activity = ActivityState::Applying;
        assert_eq!(rt.display_color(), ratatui::style::Color::Cyan);

        rt.activity = ActivityState::Accepting;
        assert_eq!(rt.display_color(), ratatui::style::Color::LightGreen);

        rt.activity = ActivityState::Archiving;
        assert_eq!(rt.display_color(), ratatui::style::Color::Magenta);

        rt.activity = ActivityState::Resolving;
        assert_eq!(rt.display_color(), ratatui::style::Color::LightCyan);

        rt.activity = ActivityState::Idle;
        rt.wait_state = WaitState::MergeWait;
        assert_eq!(rt.display_color(), ratatui::style::Color::LightMagenta);

        rt.wait_state = WaitState::ResolveWait;
        assert_eq!(rt.display_color(), ratatui::style::Color::Magenta);

        rt.wait_state = WaitState::RejectWait;
        assert_eq!(rt.display_status(), "reject pending");
        assert_eq!(rt.display_color(), ratatui::style::Color::LightMagenta);

        rt.wait_state = WaitState::None;
        rt.terminal = TerminalState::Merged;
        assert_eq!(rt.display_color(), ratatui::style::Color::LightBlue);

        rt.terminal = TerminalState::Rejected("blocked".to_string());
        assert_eq!(rt.display_color(), ratatui::style::Color::LightRed);

        rt.terminal = TerminalState::Stopped;
        assert_eq!(rt.display_color(), ratatui::style::Color::DarkGray);

        rt.terminal = TerminalState::Error("boom".to_string());
        assert_eq!(rt.display_color(), ratatui::style::Color::Red);
    }

    #[test]
    fn test_error_message_derivation() {
        let rt = ChangeRuntimeState {
            terminal: TerminalState::Error("fatal".to_string()),
            ..Default::default()
        };
        assert_eq!(rt.error_message(), Some("fatal"));

        let rt2 = ChangeRuntimeState {
            terminal: TerminalState::Merged,
            ..Default::default()
        };
        assert_eq!(rt2.error_message(), None);
    }

    // -----------------------------------------------------------------------
    // Phase 1.5: active/inactive classification
    // -----------------------------------------------------------------------

    #[test]
    fn test_runtime_state_active_classification() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Idle → not active.
        assert!(!state.is_active_change("c"));

        // Applying → active.
        state.runtime_entry("c").activity = ActivityState::Applying;
        assert!(state.is_active_change("c"));

        // Terminal → not active (invariant-wise, but is_active_change checks activity).
        state.runtime_entry("c").terminal = TerminalState::Merged;
        state.runtime_entry("c").activity = ActivityState::Idle;
        assert!(!state.is_active_change("c"));
        assert!(state.is_terminal_change("c"));
    }

    // -----------------------------------------------------------------------
    // Phase 2.2: apply_command queue intent
    // -----------------------------------------------------------------------

    #[test]
    fn final_terminal_dispatch_stop_excludes_recoverable_terminal_error() {
        let mut state = OrchestratorState::new(
            vec![
                "merged".to_string(),
                "archived".to_string(),
                "rejected".to_string(),
                "error".to_string(),
            ],
            0,
        );
        state.apply_execution_event(&crate::events::ExecutionEvent::MergeCompleted {
            change_id: "merged".to_string(),
            revision: "rev".to_string(),
        });
        state.runtime_entry("archived").terminal = TerminalState::Pushed;
        state.runtime_entry("rejected").terminal = TerminalState::Rejected("no".to_string());
        state.apply_execution_event(&crate::events::ExecutionEvent::ProcessingError {
            id: "error".to_string(),
            error: "boom".to_string(),
        });

        assert!(state.is_final_terminal_dispatch_stop("merged"));
        assert!(state.is_final_terminal_dispatch_stop("archived"));
        assert!(state.is_final_terminal_dispatch_stop("rejected"));
        assert!(!state.is_final_terminal_dispatch_stop("error"));
        assert!(state.is_terminal_error_change("error"));
        assert!(matches!(
            state.apply_command(ReducerCommand::AddToQueue("error".to_string())),
            ReduceOutcome::NoOp
        ));
        assert!(matches!(
            state.apply_command(ReducerCommand::RetryError("error".to_string())),
            ReduceOutcome::Changed(_)
        ));
        assert_eq!(state.display_status("error"), "queued");
    }

    #[test]
    fn test_apply_command_queue_intent() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // AddToQueue.
        let outcome = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(outcome, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "queued");

        // AddToQueue again → NoOp (idempotent).
        let outcome2 = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(outcome2, ReduceOutcome::NoOp));

        state.mark_stalled("c".to_string());
        state.runtime_entry("c").wait_state = WaitState::Stalled;
        let outcome_from_stalled = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(outcome_from_stalled, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "queued");
        assert!(!state.stalled_change_ids().contains("c"));

        // RemoveFromQueue.
        let outcome3 = state.apply_command(ReducerCommand::RemoveFromQueue("c".to_string()));
        assert!(matches!(outcome3, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "not queued");

        // RemoveFromQueue again → NoOp.
        let outcome4 = state.apply_command(ReducerCommand::RemoveFromQueue("c".to_string()));
        assert!(matches!(outcome4, ReduceOutcome::NoOp));

        // DequeueChange.
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        let outcome5 = state.apply_command(ReducerCommand::DequeueChange("c".to_string()));
        assert!(matches!(outcome5, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "not queued");

        // DequeueChange on already not queued keeps queue-off state.
        let outcome6 = state.apply_command(ReducerCommand::DequeueChange("c".to_string()));
        assert!(matches!(outcome6, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "not queued");

        // Terminal errors require the explicit retry transition, not ordinary queue intent.
        state.runtime_entry("c").terminal = TerminalState::Error("boom".to_string());
        let outcome_error_queue = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(outcome_error_queue, ReduceOutcome::NoOp));
        assert_eq!(state.display_status("c"), "error");
        let outcome_error_retry = state.apply_command(ReducerCommand::RetryError("c".to_string()));
        assert!(matches!(outcome_error_retry, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "queued");

        // Rejected cannot be re-queued until a refresh reactivates the change.
        state.runtime_entry("c").terminal = TerminalState::Rejected("blocked".to_string());
        let outcome7 = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(outcome7, ReduceOutcome::NoOp));
        let outcome_rejected_retry =
            state.apply_command(ReducerCommand::RetryError("c".to_string()));
        assert!(matches!(outcome_rejected_retry, ReduceOutcome::NoOp));

        // Reactivation path: when the change reappears in active listing after

        // REJECTED.md removal, ChangesRefreshed clears rejected terminal state.
        use crate::events::ExecutionEvent;
        use crate::openspec::{Change, ProposalMetadata};
        use std::collections::{HashMap, HashSet};
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![Change {
                id: "c".to_string(),
                completed_tasks: 0,
                total_tasks: 1,
                last_modified: "now".to_string(),
                dependencies: Vec::new(),
                metadata: ProposalMetadata::default(),
            }],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::new(),
        });
        assert_eq!(state.display_status("c"), "not queued");

        // AddToQueue must be accepted again after reactivation.
        let outcome8 = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(outcome8, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "queued");
    }

    #[test]
    fn retry_terminal_error_clears_error_gate_and_stale_retry_metadata() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        state.apply_execution_event(&crate::events::ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "apply".to_string(),
        });
        state.record_error_and_check_circuit_breaker(
            "c",
            "boom",
            CircuitBreakerConfig {
                enabled: true,
                threshold: 2,
            },
        );
        state.mark_stalled("c".to_string());
        state.apply_execution_event(&crate::events::ExecutionEvent::ProcessingError {
            id: "c".to_string(),
            error: "boom".to_string(),
        });

        assert_eq!(state.display_status("c"), "error");
        assert!(state.queued_change_ids().is_empty());
        assert!(state.last_error("c").is_some());

        let outcome = state.apply_command(ReducerCommand::RetryError("c".to_string()));

        assert!(matches!(outcome, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "queued");
        assert_eq!(state.queued_change_ids(), vec!["c".to_string()]);
        assert!(state.last_error("c").is_none());
        assert!(!state.stalled_change_ids().contains("c"));
        let runtime = state.change_runtime("c").expect("runtime should exist");
        assert!(matches!(runtime.terminal, TerminalState::None));
        assert_eq!(runtime.wait_state, WaitState::None);
        assert_eq!(runtime.blocked_metadata, BlockedMetadata::default());
    }

    #[test]
    fn late_success_supersedes_recoverable_error_without_requeueing_apply() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&crate::events::ExecutionEvent::ApplyFailed {
            change_id: "c".to_string(),
            error: "boom".to_string(),
        });
        assert_eq!(state.display_status("c"), "error");

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });

        assert_eq!(state.display_status("c"), "merged");
        assert!(state.queued_change_ids().is_empty());
        let runtime = state.change_runtime("c").expect("runtime should exist");
        assert_eq!(runtime.queue_intent, QueueIntent::NotQueued);
    }

    #[test]
    fn archive_resumed_clears_recoverable_archive_error_and_restores_archiving_activity() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ArchiveFailed {
            change_id: "c".to_string(),
            error: "Archive commit finalization failed".to_string(),
            reason: Some("verification_failed".to_string()),
            summary: Some("commit incomplete".to_string()),
        });
        assert_eq!(state.display_status("c"), "error");

        state.apply_execution_event(&ExecutionEvent::ArchiveResumed {
            change_id: "c".to_string(),
            reason: Some("archive_commit_incomplete".to_string()),
            summary: Some("resuming archived dirty repair".to_string()),
        });

        assert_eq!(state.display_status("c"), "archiving");
        let runtime = state.change_runtime("c").expect("runtime should exist");
        assert!(matches!(runtime.terminal, TerminalState::None));
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn test_base_mutating_lane_is_single_occupant_across_resolving_and_rejecting() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("a".to_string()));
        assert_eq!(state.display_status("a"), "resolving");

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "b".to_string(),
            workspace_name: "ws-b".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });

        assert_eq!(state.display_status("b"), "reject pending");
        assert_eq!(state.reject_wait_change_ids(), vec!["b".to_string()]);
        assert!(state.has_other_post_archive_lane_blocker("b"));
        assert!(state.global_invariants_hold());
    }

    /// Only explicit operator intent grants the scoped merge-continuation
    /// permission; an auto-resumable deferral re-enters `ResolveWait` without it
    /// and therefore keeps the unchanged generic dirty preflight.
    #[test]
    fn manual_resolve_permission_is_granted_only_by_explicit_operator_intent() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "merge lane busy".to_string(),
            auto_resumable: true,
        });
        assert!(
            !state.has_manual_resolve_retry("alpha"),
            "an auto-resumable deferral is not operator intent"
        );

        state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));
        assert!(state.has_manual_resolve_retry("alpha"));
    }

    /// The permission is consumed by the dispatch that owns the base lane, and a
    /// later ordinary scheduled attempt must not inherit it.
    #[test]
    fn manual_resolve_permission_is_consumed_once_and_is_not_sticky() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);
        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "bounded resolve exhausted".to_string(),
            auto_resumable: false,
        });
        state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));

        // Promotion only reads the permission: a dispatch deferred by lane
        // occupancy has to find it again.
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("alpha".to_string(), WaitState::ResolveWait))
        );
        assert!(state.has_manual_resolve_retry("alpha"));

        assert!(state.consume_manual_resolve_retry("alpha"));
        assert!(!state.has_manual_resolve_retry("alpha"));
        assert!(
            !state.consume_manual_resolve_retry("alpha"),
            "a second dispatch gets the unchanged generic preflight"
        );
        assert!(!state.consume_manual_resolve_retry("unknown-change"));
    }

    /// Losing the wait loses the retry the permission authorized, so a dequeue
    /// must not leave a live permission behind for a later requeue.
    #[test]
    fn manual_resolve_permission_does_not_survive_dequeue() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);
        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "bounded resolve exhausted".to_string(),
            auto_resumable: false,
        });
        state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));
        assert!(state.has_manual_resolve_retry("alpha"));

        state.apply_command(ReducerCommand::DequeueChange("alpha".to_string()));

        assert!(!state.has_manual_resolve_retry("alpha"));
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn release_base_mutating_lane_after_retry_restores_resolve_wait_uniquely() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "conflict".to_string(),
            auto_resumable: false,
        });
        state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("alpha".to_string(), WaitState::ResolveWait))
        );
        assert!(state.is_base_mutating_lane_occupied());

        assert!(state.release_base_mutating_lane_after_retry("alpha", WaitState::ResolveWait));
        assert!(!state.is_base_mutating_lane_occupied());
        assert_eq!(state.display_status("alpha"), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec!["alpha".to_string()]);
        assert!(state.global_invariants_hold());

        assert!(!state.release_base_mutating_lane_after_retry("alpha", WaitState::ResolveWait));
        assert_eq!(state.resolve_wait_queue, vec!["alpha".to_string()]);
        assert!(state.reject_wait_queue.is_empty());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn abandon_base_mutating_lane_occupant_releases_resolve_without_requeueing() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string(), "beta".to_string()], 0);

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "conflict".to_string(),
            auto_resumable: false,
        });
        state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("alpha".to_string(), WaitState::ResolveWait))
        );
        state.resolve_wait_queue.push("alpha".to_string());
        state.reject_wait_queue.push("alpha".to_string());
        state
            .runtime_entry("alpha")
            .set_blocked_metadata("reason", "unblock", "snapshot");

        assert!(state.abandon_base_mutating_lane_occupant("alpha"));

        let runtime = state.change_runtime("alpha").expect("runtime for alpha");
        assert_eq!(runtime.activity, ActivityState::Idle);
        assert_eq!(runtime.wait_state, WaitState::None);
        assert_eq!(runtime.blocked_metadata, BlockedMetadata::default());
        assert!(!state
            .resolve_wait_change_ids()
            .contains(&"alpha".to_string()));
        assert!(!state
            .reject_wait_change_ids()
            .contains(&"alpha".to_string()));
        assert!(!state.resolve_wait_queue.contains(&"alpha".to_string()));
        assert!(!state.reject_wait_queue.contains(&"alpha".to_string()));
        assert!(!state.is_base_mutating_lane_occupied());
        assert!(state.global_invariants_hold());

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "beta".to_string(),
            reason: "conflict".to_string(),
            auto_resumable: false,
        });
        state.apply_command(ReducerCommand::ResolveMerge("beta".to_string()));
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("beta".to_string(), WaitState::ResolveWait))
        );
    }

    #[test]
    fn abandon_base_mutating_lane_occupant_releases_reject_and_noops_terminal_or_non_occupant() {
        let mut state = OrchestratorState::new(
            vec![
                "lane".to_string(),
                "reject".to_string(),
                "terminal".to_string(),
            ],
            0,
        );

        state.apply_execution_event(&crate::events::ExecutionEvent::ChangeArchived(
            "lane".to_string(),
        ));
        state.mark_reject_wait("reject");
        state.apply_execution_event(&crate::events::ExecutionEvent::MergeCompleted {
            change_id: "lane".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("reject".to_string(), WaitState::RejectWait))
        );

        assert!(state.abandon_base_mutating_lane_occupant("reject"));
        let runtime = state.change_runtime("reject").expect("runtime for reject");
        assert_eq!(runtime.activity, ActivityState::Idle);
        assert_eq!(runtime.wait_state, WaitState::None);
        assert!(state.reject_wait_change_ids().is_empty());
        assert!(!state.is_base_mutating_lane_occupied());
        assert!(state.global_invariants_hold());

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeCompleted {
            change_id: "terminal".to_string(),
            revision: "rev".to_string(),
        });
        assert!(!state.abandon_base_mutating_lane_occupant("terminal"));
        assert!(!state.abandon_base_mutating_lane_occupant("unknown"));
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn release_base_mutating_lane_after_retry_restores_reject_wait_and_noops_terminal() {
        let mut state = OrchestratorState::new(vec!["lane".to_string(), "reject".to_string()], 0);

        state.apply_execution_event(&crate::events::ExecutionEvent::ChangeArchived(
            "lane".to_string(),
        ));
        state.mark_reject_wait("reject");
        state.apply_execution_event(&crate::events::ExecutionEvent::MergeCompleted {
            change_id: "lane".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("reject".to_string(), WaitState::RejectWait))
        );

        assert!(state.release_base_mutating_lane_after_retry("reject", WaitState::RejectWait));
        assert!(!state.is_base_mutating_lane_occupied());
        assert_eq!(state.display_status("reject"), "reject pending");
        assert_eq!(state.reject_wait_change_ids(), vec!["reject".to_string()]);
        assert!(state.global_invariants_hold());

        state.apply_execution_event(&crate::events::ExecutionEvent::ChangeRejected {
            change_id: "reject".to_string(),
            reason: "confirmed".to_string(),
        });
        assert!(!state.release_base_mutating_lane_after_retry("reject", WaitState::RejectWait));
        assert!(state.reject_wait_change_ids().is_empty());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn test_reject_wait_queue_membership_and_clear_on_start_completion() {
        use crate::events::{ExecutionEvent, RejectionOutcome};

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("a".to_string()));
        state.mark_reject_wait("b");

        assert_eq!(state.display_status("b"), "reject pending");
        assert_eq!(state.reject_wait_change_ids(), vec!["b".to_string()]);

        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "a".to_string(),
            revision: "rev".to_string(),
        });
        let promoted = state.promote_next_base_mutating_lane_waiter();
        assert_eq!(promoted, Some(("b".to_string(), WaitState::RejectWait)));
        assert_eq!(state.display_status("b"), "rejecting");
        assert!(state.reject_wait_change_ids().is_empty());

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "b".to_string(),
            outcome: RejectionOutcome::Confirm,
        });
        assert_eq!(state.display_status("b"), "rejected");
        assert!(state.reject_wait_change_ids().is_empty());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn command_side_effects_update_terminal_wait_and_base_lane_queues() {
        let mut state = OrchestratorState::new(vec!["c".to_string(), "d".to_string()], 0);

        state.resolve_wait_queue.push("c".to_string());
        state.reject_wait_queue.push("c".to_string());
        let add_outcome = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(add_outcome, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "queued");
        assert!(state.resolve_wait_queue.is_empty());
        assert!(state.reject_wait_queue.is_empty());

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "manual conflict".to_string(),
            auto_resumable: false,
        });
        let resolve_outcome = state.apply_command(ReducerCommand::ResolveMerge("c".to_string()));
        assert!(matches!(
            resolve_outcome,
            ReduceOutcome::Changed(ReducerEffect::WaitStateSet {
                wait: WaitState::ResolveWait,
                ..
            })
        ));
        assert_eq!(state.display_status("c"), "resolve pending");
        assert_eq!(state.resolve_wait_queue, vec!["c".to_string()]);
        assert!(state.reject_wait_queue.is_empty());

        let dequeue_outcome = state.apply_command(ReducerCommand::DequeueChange("c".to_string()));
        assert!(matches!(dequeue_outcome, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "not queued");
        assert!(state.resolve_wait_queue.is_empty());
        assert!(state.reject_wait_queue.is_empty());
        let runtime = state.change_runtime("c").expect("runtime for c");
        assert!(matches!(runtime.terminal, TerminalState::None));
        assert_eq!(runtime.activity, ActivityState::Idle);
        assert_eq!(runtime.wait_state, WaitState::None);
        assert_eq!(runtime.queue_intent, QueueIntent::NotQueued);
        assert!(runtime.dequeued);

        state.apply_command(ReducerCommand::AddToQueue("d".to_string()));
        let stop_outcome = state.apply_command(ReducerCommand::StopChange("d".to_string()));
        assert!(matches!(stop_outcome, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("d"), "stopped");
        let runtime = state.change_runtime("d").expect("runtime for d");
        assert!(matches!(runtime.terminal, TerminalState::Stopped));
        assert_eq!(runtime.queue_intent, QueueIntent::NotQueued);
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn execution_event_side_effects_update_wait_queues_and_blocked_metadata() {
        use crate::events::{ExecutionEvent, RejectionOutcome};

        let mut state = OrchestratorState::new(
            vec![
                "resolving".to_string(),
                "archived".to_string(),
                "reject".to_string(),
            ],
            0,
        );

        state.apply_execution_event(&ExecutionEvent::ResolveStarted {
            change_id: "resolving".to_string(),
            command: "resolve".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("archived".to_string()));
        assert_eq!(state.display_status("archived"), "resolve pending");
        assert_eq!(state.resolve_wait_queue, vec!["archived".to_string()]);

        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "archived".to_string(),
            error: "base dirty".to_string(),
        });
        assert_eq!(state.display_status("archived"), "merge wait");
        assert!(state.resolve_wait_queue.is_empty());

        state.mark_reject_wait("reject");
        assert_eq!(state.reject_wait_queue, vec!["reject".to_string()]);
        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "reject".to_string(),
            outcome: RejectionOutcome::Block,
        });
        assert!(state.reject_wait_queue.is_empty());
        let runtime = state.change_runtime("reject").expect("runtime for reject");
        assert_eq!(runtime.wait_state, WaitState::Stalled);
        assert!(runtime.blocked_metadata.blocker_reason.is_some());
        assert!(matches!(runtime.terminal, TerminalState::None));

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "reject".to_string(),
            outcome: RejectionOutcome::Confirm,
        });
        assert_eq!(state.display_status("reject"), "rejected");
        assert!(state.reject_wait_queue.is_empty());
        assert!(state.global_invariants_hold());
    }

    // -----------------------------------------------------------------------
    // Phase 2.3: apply_execution_event transitions
    // -----------------------------------------------------------------------

    // === Ephemeral Apply commit subphase ===

    /// The commit subphase is presentation only: it changes the Apply lane's
    /// rendered operation and nothing else. The canonical lifecycle, the queue
    /// intent, and every routing input stay exactly where they were.
    #[test]
    fn commit_presentation_never_changes_the_canonical_lifecycle() {
        use crate::events::{ApplyCommitPhase, ExecutionEvent};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        let before = state.change_runtime("c").cloned().expect("runtime");

        state.apply_execution_event(&ExecutionEvent::ApplyCommitPhase {
            change_id: "c".to_string(),
            phase: ApplyCommitPhase::Started,
            attempt: 4,
        });

        let during = state.change_runtime("c").expect("runtime");
        assert_eq!(
            state.display_status("c"),
            "applying",
            "the canonical status must not become a commit state"
        );
        assert_eq!(during.apply_operation_label(), "commit");
        assert_eq!(during.commit_phase_attempt, Some(4));
        assert_eq!(during.activity, before.activity);
        assert_eq!(during.wait_state, before.wait_state);
        assert_eq!(during.queue_intent, before.queue_intent);
        assert!(state.global_invariants_hold());
    }

    /// Completion, failure, and a repair `ApplyStarted` all clear presentation,
    /// so no row can be left rendering a `[commit]` that is no longer running.
    #[test]
    fn every_terminal_commit_phase_clears_presentation() {
        use crate::events::{ApplyCommitPhase, ExecutionEvent};

        for closing in [ApplyCommitPhase::Completed, ApplyCommitPhase::Failed] {
            let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
            state.apply_execution_event(&ExecutionEvent::ApplyStarted {
                change_id: "c".to_string(),
                command: "cmd".to_string(),
            });
            state.apply_execution_event(&ExecutionEvent::ApplyCommitPhase {
                change_id: "c".to_string(),
                phase: ApplyCommitPhase::Started,
                attempt: 1,
            });
            state.apply_execution_event(&ExecutionEvent::ApplyCommitPhase {
                change_id: "c".to_string(),
                phase: closing,
                attempt: 1,
            });

            let runtime = state.change_runtime("c").expect("runtime");
            assert_eq!(
                runtime.commit_phase_attempt, None,
                "{closing:?} must clear commit presentation"
            );
            assert_eq!(runtime.apply_operation_label(), "apply");
            assert_eq!(state.display_status("c"), "applying");
        }
    }

    /// Cancellation can land anywhere inside finalization, including before any
    /// terminal commit-phase event was emitted — a WIP snapshot failure between
    /// `Started` and the commit surfaces only as a processing error. Every one
    /// of those routes must clear the ephemeral presentation, or the row keeps
    /// rendering `[commit]` for a sequence that is no longer running.
    #[test]
    fn cancelling_outside_the_commit_sequence_clears_commit_presentation() {
        use crate::events::{ApplyCommitPhase, ExecutionEvent};

        /// Put a change into an active commit phase, then cancel it the way the
        /// named route does.
        fn active_commit_then(cancel: impl Fn(&mut OrchestratorState)) -> Option<u32> {
            let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
            state.apply_execution_event(&ExecutionEvent::ApplyStarted {
                change_id: "c".to_string(),
                command: "cmd".to_string(),
            });
            state.apply_execution_event(&ExecutionEvent::ApplyCommitPhase {
                change_id: "c".to_string(),
                phase: ApplyCommitPhase::Started,
                attempt: 3,
            });
            assert_eq!(
                state
                    .change_runtime("c")
                    .expect("runtime")
                    .commit_phase_attempt,
                Some(3),
                "precondition: the row is presenting an active commit phase"
            );

            cancel(&mut state);
            assert!(state.global_invariants_hold());
            state
                .change_runtime("c")
                .expect("runtime")
                .commit_phase_attempt
        }

        assert_eq!(
            active_commit_then(|state| {
                state.apply_execution_event(&ExecutionEvent::ChangeDequeued {
                    change_id: "c".to_string(),
                });
            }),
            None,
            "ChangeDequeued must clear commit presentation"
        );
        assert_eq!(
            active_commit_then(|state| {
                state.apply_execution_event(&ExecutionEvent::ChangeStopped {
                    change_id: "c".to_string(),
                });
            }),
            None,
            "ChangeStopped must clear commit presentation"
        );
        assert_eq!(
            active_commit_then(|state| {
                state.apply_command(ReducerCommand::DequeueChange("c".to_string()));
            }),
            None,
            "the DequeueChange command must clear commit presentation"
        );
        assert_eq!(
            active_commit_then(|state| {
                state.apply_command(ReducerCommand::StopChange("c".to_string()));
            }),
            None,
            "the StopChange command must clear commit presentation"
        );
        assert_eq!(
            active_commit_then(|state| {
                state.apply_execution_event(&ExecutionEvent::ProcessingError {
                    id: "c".to_string(),
                    error: "WIP snapshot failed".to_string(),
                });
            }),
            None,
            "a processing error must clear commit presentation"
        );
    }

    /// A repair iteration restores `[apply]` even if the closing phase event was
    /// never observed.
    #[test]
    fn a_repair_iteration_restores_apply_presentation() {
        use crate::events::{ApplyCommitPhase, ExecutionEvent};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ApplyCommitPhase {
            change_id: "c".to_string(),
            phase: ApplyCommitPhase::Started,
            attempt: 2,
        });
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });

        assert_eq!(
            state
                .change_runtime("c")
                .expect("runtime")
                .commit_phase_attempt,
            None,
            "ApplyStarted always clears commit presentation"
        );
        assert_eq!(
            state.all_apply_operation_labels().get("c").copied(),
            Some("apply")
        );
    }

    /// A terminal or dequeued row must not adopt commit presentation from a
    /// late in-flight event.
    #[test]
    fn a_terminal_row_never_adopts_commit_presentation() {
        use crate::events::{ApplyCommitPhase, ExecutionEvent};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ApplyFailed {
            change_id: "c".to_string(),
            error: "boom".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ApplyCommitPhase {
            change_id: "c".to_string(),
            phase: ApplyCommitPhase::Started,
            attempt: 1,
        });

        assert_eq!(state.display_status("c"), "error");
        assert_eq!(
            state
                .change_runtime("c")
                .expect("runtime")
                .commit_phase_attempt,
            None
        );
    }

    /// Constitutional law 1: a restart re-derives routing from the workspace, so
    /// commit presentation must simply be absent in a fresh reducer and must not
    /// be reachable from any persisted input.
    #[test]
    fn a_fresh_reducer_has_no_commit_presentation() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        assert_eq!(
            state.all_apply_operation_labels().get("c").copied(),
            Some("apply"),
            "a fresh reducer presents the Apply lane, never a commit subphase"
        );

        state.apply_execution_event(&crate::events::ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        assert_eq!(
            state
                .change_runtime("c")
                .expect("runtime")
                .commit_phase_attempt,
            None,
            "process-local commit presentation starts empty"
        );
    }

    #[test]
    fn test_apply_execution_event_transitions() {
        use crate::events::{ExecutionEvent, RejectionOutcome};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        assert_eq!(state.display_status("c"), "applying");

        state.apply_execution_event(&ExecutionEvent::ApplyCompleted {
            change_id: "c".to_string(),
            revision: "rev1".to_string(),
        });
        assert_eq!(state.display_status("c"), "not queued");

        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        assert_eq!(state.display_status("c"), "accepting");

        state.apply_execution_event(&ExecutionEvent::AcceptanceCompleted {
            change_id: "c".to_string(),
        });

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "c".to_string(),
            workspace_name: "ws-c".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });
        assert_eq!(state.display_status("c"), "rejecting");

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "c".to_string(),
            outcome: RejectionOutcome::Resume,
        });
        assert_eq!(state.display_status("c"), "applying");
        assert_ne!(
            state
                .change_runtime("c")
                .expect("runtime for c after rejecting resume")
                .activity,
            ActivityState::Rejecting
        );

        state.apply_execution_event(&ExecutionEvent::RejectionReviewFailed {
            change_id: "c".to_string(),
            error: "rejecting failed".to_string(),
        });
        assert_eq!(state.display_status("c"), "error");
        assert_ne!(
            state
                .change_runtime("c")
                .expect("runtime for c after rejecting failure")
                .activity,
            ActivityState::Rejecting
        );

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ProcessingStarted("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "c".to_string(),
            workspace_name: "ws-c".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });
        assert_eq!(state.display_status("c"), "rejecting");

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "c".to_string(),
            outcome: RejectionOutcome::Confirm,
        });
        assert_eq!(state.display_status("c"), "rejected");
        assert_ne!(
            state
                .change_runtime("c")
                .expect("runtime for c after rejecting confirm")
                .activity,
            ActivityState::Rejecting
        );
    }

    /// The explicit category reaches reducer state verbatim, and prose never
    /// overrides it — the same narrative under a different category must produce
    /// a different external blocker classification.
    #[test]
    fn test_explicit_blocker_categories_reach_blocked_metadata_without_prose_inference() {
        // Every message below contains words the old classifier keyed on
        // (docker, daemon, timeout, credential, still running). None of them may
        // change the category.
        let cases = [
            (
                "credential",
                "docker image pull failed: lookup registry-1.docker.io i/o timeout",
            ),
            ("infrastructure", "missing non-mockable external credential"),
            (
                "human_decision",
                "agent-exec managed verification job still running",
            ),
            ("external_approval", "Cannot connect to the Docker daemon"),
            (
                "schema_incompatibility",
                "package registry timeout fetching crate",
            ),
        ];

        for (idx, (explicit_category, message)) in cases.iter().enumerate() {
            let change_id = format!("c-{idx}");
            let mut state = OrchestratorState::new(vec![change_id.clone()], 0);
            state.apply_execution_event(&crate::events::ExecutionEvent::AcceptanceGated {
                change_id: change_id.clone(),
                blocker: crate::events::StalledBlocker::acceptance_external(
                    *explicit_category,
                    *message,
                ),
            });

            let runtime = state
                .change_runtime(&change_id)
                .expect("runtime after blocker classification");
            // A validated non-repository prerequisite is `blocked`, and its
            // blocker kind says why — never a dependency edge, never a stall.
            assert_eq!(state.display_status(&change_id), "blocked");
            assert_eq!(runtime.blocker_kind(), BlockerKind::External);
            assert!(matches!(runtime.terminal, TerminalState::None));
            assert_eq!(
                runtime.blocked_metadata.blocker_reason.as_deref(),
                Some(format!("external-blocked:{explicit_category}").as_str()),
                "the explicit category must survive verbatim for {message}"
            );
            assert!(
                runtime
                    .blocked_metadata
                    .unblock_metadata
                    .as_deref()
                    .unwrap_or_default()
                    .contains(message),
                "metadata should preserve observed error summary for {message}"
            );
        }
    }

    #[test]
    fn test_execution_blocked_permission_denial_transitions_to_stalled_with_operator_guidance() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let denial = crate::permission::classify_permission_denial(&[Some(
            "Read permission denied for /private/secret.txt",
        )])
        .expect("denial should classify");
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "c".to_string(),
            command: "accept".to_string(),
        });

        state.apply_execution_event(&ExecutionEvent::ExecutionBlocked {
            change_id: "c".to_string(),
            blocker: StalledBlocker::permission_denial("acceptance", &denial),
        });

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert_eq!(state.display_status("c"), "stalled");
        assert_eq!(runtime.activity, ActivityState::Idle);
        assert_eq!(runtime.wait_state, WaitState::Stalled);
        assert!(matches!(runtime.terminal, TerminalState::None));
        let metadata = runtime
            .blocked_metadata
            .unblock_metadata
            .as_deref()
            .expect("operator guidance metadata");
        assert!(metadata.contains("permission/tool policy denial"));
        assert!(metadata.contains("operator action"));
        assert!(!metadata.to_ascii_lowercase().contains("dependency blocked"));
    }

    #[test]
    fn test_acceptance_gated_transitions_to_external_blocked_with_structured_metadata() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "c".to_string(),
            command: "accept".to_string(),
        });

        state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
            change_id: "c".to_string(),
            blocker: StalledBlocker::acceptance_external(
                "pending_verification",
                "docker image pull failed: lookup registry-1.docker.io i/o timeout",
            ),
        });

        let runtime = state
            .change_runtime("c")
            .expect("runtime for c after acceptance gated");
        assert_eq!(state.display_status("c"), "blocked");
        assert_eq!(runtime.blocker_kind(), BlockerKind::External);
        assert_eq!(runtime.activity, ActivityState::Idle);
        assert_eq!(runtime.wait_state, WaitState::ExternalBlocked);
        assert!(matches!(runtime.terminal, TerminalState::None));
        assert_eq!(
            runtime.blocked_metadata.blocker_reason.as_deref(),
            Some("external-blocked:pending_verification")
        );
        let unblock = runtime
            .blocked_metadata
            .unblock_metadata
            .as_deref()
            .expect("unblock metadata");
        assert!(unblock.contains("external blocker (pending_verification)"));
        assert!(unblock.contains("reported by acceptance"));
        assert!(unblock.contains("unblock when"));
        assert!(unblock.contains("next action"));
        assert!(unblock.contains("docker image pull failed"));
        assert!(runtime.blocked_metadata.unblock_condition.is_some());
        assert_eq!(
            runtime.blocked_metadata.blocker_origin.as_deref(),
            Some("acceptance")
        );
        assert_eq!(
            runtime.blocked_metadata.worktree_snapshot.as_deref(),
            Some("existing worktree and WIP context are preserved while stalled")
        );
    }

    /// Acceptance facts that do not validate as a non-repository prerequisite
    /// stay `stalled`. An unsupported category and a missing unblock condition
    /// must not be promoted into an inferred external wait.
    #[test]
    fn test_unvalidated_acceptance_blocker_facts_remain_stalled() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
            change_id: "c".to_string(),
            blocker: StalledBlocker {
                category: "acceptance_finding".to_string(),
                phase: "acceptance".to_string(),
                gate: "acceptance".to_string(),
                error_summary: "unresolved finding".to_string(),
                evidence: vec!["src/lib.rs:1 missing coverage".to_string()],
                unblock_condition: None,
                prerequisite_owner: None,
                next_action: "resolve the finding and retry".to_string(),
                resumable: true,
                worktree_preserved: true,
            },
        });

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert_eq!(state.display_status("c"), "stalled");
        assert_eq!(runtime.wait_state, WaitState::Stalled);
        assert_eq!(runtime.blocker_kind(), BlockerKind::None);
        assert!(runtime.blocked_metadata.unblock_condition.is_none());
        // The unsupported category is reported first, and the detail keeps the
        // reason so an operator can see exactly what disqualified the claim.
        assert!(runtime
            .blocked_metadata
            .unblock_metadata
            .as_deref()
            .unwrap_or_default()
            .contains("'acceptance_finding' is not one of"));
    }

    /// Dependency waits and external prerequisite waits are both `blocked` and
    /// stay distinguishable, and neither is confused with an execution stall.
    #[test]
    fn test_dependency_and_external_waits_share_blocked_but_keep_their_kind() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let mut state = OrchestratorState::new(
            vec![
                "dependency-wait".to_string(),
                "external-wait".to_string(),
                "execution-stall".to_string(),
            ],
            0,
        );
        state.apply_execution_event(&ExecutionEvent::DependencyBlocked {
            change_id: "dependency-wait".to_string(),
            dependency_ids: vec!["alpha".to_string()],
        });
        state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
            change_id: "external-wait".to_string(),
            blocker: StalledBlocker::acceptance_external("credential", "STAGING_API_KEY is unset"),
        });
        let denial = crate::permission::classify_permission_denial(&[Some(
            "Read permission denied for /private/secret.txt",
        )])
        .expect("denial should classify");
        state.apply_execution_event(&ExecutionEvent::ExecutionBlocked {
            change_id: "execution-stall".to_string(),
            blocker: StalledBlocker::permission_denial("acceptance", &denial),
        });

        assert_eq!(state.display_status("dependency-wait"), "blocked");
        assert_eq!(state.display_status("external-wait"), "blocked");
        assert_eq!(state.display_status("execution-stall"), "stalled");

        assert_eq!(
            state
                .change_runtime("dependency-wait")
                .unwrap()
                .blocker_kind(),
            BlockerKind::Dependency
        );
        assert_eq!(
            state
                .change_runtime("external-wait")
                .unwrap()
                .blocker_kind(),
            BlockerKind::External
        );
        assert_eq!(
            state
                .change_runtime("execution-stall")
                .unwrap()
                .blocker_kind(),
            BlockerKind::None
        );

        // Every surface reads one reducer-derived view, so none of them has to
        // re-derive `blocked` versus `stalled` or the blocker kind.
        let views = state.all_blocker_views();
        assert_eq!(views["dependency-wait"].status, "blocked");
        assert_eq!(views["dependency-wait"].kind, BlockerKind::Dependency);
        assert_eq!(views["external-wait"].status, "blocked");
        assert_eq!(views["external-wait"].kind, BlockerKind::External);
        assert_eq!(views["external-wait"].origin.as_deref(), Some("acceptance"));
        assert!(views["external-wait"].unblock_condition.is_some());
        assert_eq!(views["execution-stall"].status, "stalled");
        assert_eq!(views["execution-stall"].kind, BlockerKind::None);
        assert!(views["execution-stall"].unblock_condition.is_none());

        // Only the external wait is suppressed as an external prerequisite hold.
        assert_eq!(
            state.externally_blocked_change_ids(),
            HashSet::from(["external-wait".to_string()])
        );

        // The external wait never becomes a dependency edge.
        assert!(state
            .change_runtime("external-wait")
            .unwrap()
            .blocked_metadata
            .blocker_reason
            .as_deref()
            .unwrap_or_default()
            .starts_with("external-blocked:"));
    }

    /// The whole Acceptance-owned hold is this reducer entry. It suppresses
    /// ordinary dispatch, carries resumability, is dropped by an explicit retry,
    /// and — because it is only in memory — does not exist in a fresh reducer.
    #[test]
    fn acceptance_hold_is_in_memory_only_and_clears_on_retry_or_restart() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
            change_id: "c".to_string(),
            blocker: StalledBlocker::acceptance_external("credential", "STAGING_API_KEY is unset"),
        });

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert!(runtime.is_external_blocked());
        assert!(runtime.is_acceptance_stalled());
        assert!(runtime.is_resumable_acceptance_stall());
        assert_eq!(
            state.acceptance_stalled_change_ids(),
            HashSet::from(["c".to_string()]),
            "the hold must be visible to queue classification"
        );
        assert_eq!(state.display_status("c"), "blocked");

        // An explicit operator retry consumes the hold in the same process, so
        // the blocked phase runs again and its fresh result reclassifies.
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(state.acceptance_stalled_change_ids().is_empty());
        let runtime = state.change_runtime("c").expect("runtime for c");
        assert!(!runtime.is_acceptance_stalled());
        assert!(!runtime.is_external_blocked());
        assert_eq!(runtime.blocker_kind(), BlockerKind::None);
        assert!(runtime.blocked_metadata.unblock_condition.is_none());
        assert_eq!(state.display_status("c"), "queued");

        // A restart is a fresh reducer: no hold survives it, so the change is
        // dispatchable again and repository evidence decides the next action.
        let restarted = OrchestratorState::new(vec!["c".to_string()], 0);
        assert!(restarted.acceptance_stalled_change_ids().is_empty());
        assert!(!restarted
            .change_runtime("c")
            .map(ChangeRuntimeState::is_external_blocked)
            .unwrap_or(false));
    }

    /// Only Acceptance-phase blockers create an Acceptance hold. An apply-origin
    /// blocker still stalls, but must not be mistaken for an acceptance-only
    /// retry candidate.
    #[test]
    fn apply_phase_blocker_stalls_without_creating_an_acceptance_hold() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let denial = crate::permission::classify_permission_denial(&[Some(
            "Read permission denied for /private/secret.txt",
        )])
        .expect("denial should classify");
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ExecutionBlocked {
            change_id: "c".to_string(),
            blocker: StalledBlocker::permission_denial("apply", &denial),
        });

        assert_eq!(state.display_status("c"), "stalled");
        assert!(state.acceptance_stalled_change_ids().is_empty());
    }

    /// A non-resumable Acceptance blocker holds dispatch but must not authorize
    /// an acceptance-only retry past evidence the operator still owns.
    #[test]
    fn non_resumable_acceptance_hold_is_not_retry_eligible() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
            change_id: "c".to_string(),
            blocker: StalledBlocker {
                resumable: false,
                ..StalledBlocker::acceptance_external("human_decision", "owner must decide")
            },
        });

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert!(runtime.is_acceptance_stalled());
        assert!(!runtime.is_resumable_acceptance_stall());
        assert_eq!(
            state.acceptance_stalled_change_ids(),
            HashSet::from(["c".to_string()])
        );
    }

    // -----------------------------------------------------------------------
    // Monotonic blocker precedence: a structured classification outranks the
    // generic blocked workspace observation the same producer emits right after
    // it. Every test below applies the exact ordered pair the external-blocker
    // path produces.
    // -----------------------------------------------------------------------

    /// Apply the compatibility workspace status the external-blocker producers
    /// emit immediately after their structured event.
    fn observe_generic_blocked_workspace(state: &mut OrchestratorState, change_id: &str) {
        state.apply_execution_event(&crate::events::ExecutionEvent::WorkspaceStatusUpdated {
            change_id: change_id.to_string(),
            workspace_name: format!("ws-{change_id}"),
            status: crate::vcs::WorkspaceStatus::Blocked,
        });
    }

    /// Reported facts that validate as an Apply-observed external prerequisite.
    fn apply_external_blocker() -> crate::events::StalledBlocker {
        crate::events::StalledBlocker {
            category: "infrastructure".to_string(),
            phase: "apply".to_string(),
            gate: "apply".to_string(),
            error_summary: "the build cache service is unreachable".to_string(),
            evidence: vec!["cargo test: connection refused to cache.internal".to_string()],
            unblock_condition: Some("cache.internal accepts connections again".to_string()),
            prerequisite_owner: Some("platform".to_string()),
            next_action: "restore cache.internal then retry apply".to_string(),
            resumable: true,
            worktree_preserved: true,
        }
    }

    /// An Acceptance-observed external wait keeps every routing-relevant field
    /// after the generic blocked workspace status the same dispatch branch
    /// emits. Before monotonic precedence this second event rebuilt the metadata
    /// as a generic stall: kind `none`, no owner, no unblock condition, and
    /// `resumable: false`.
    #[test]
    fn structured_blocker_metadata_survives_workspace_blocked_for_an_acceptance_external_wait() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "c".to_string(),
            command: "accept".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
            change_id: "c".to_string(),
            blocker: StalledBlocker {
                prerequisite_owner: Some("platform".to_string()),
                ..StalledBlocker::acceptance_external("credential", "STAGING_API_KEY is unset")
            },
        });
        let before = state
            .blocker_view("c")
            .expect("the structured event must establish a blocker view");

        observe_generic_blocked_workspace(&mut state, "c");

        assert_eq!(state.display_status("c"), "blocked");
        let after = state
            .blocker_view("c")
            .expect("the hold must survive the generic observation");
        assert_eq!(
            after, before,
            "a lower-fidelity observation must not change any projected blocker field"
        );
        assert_eq!(after.kind, BlockerKind::External);
        assert_eq!(after.origin.as_deref(), Some("acceptance"));
        assert_eq!(after.prerequisite_owner.as_deref(), Some("platform"));
        assert_eq!(
            after.category.as_deref(),
            Some("external-blocked:credential")
        );
        assert!(after
            .unblock_condition
            .as_deref()
            .unwrap_or_default()
            .contains("STAGING_API_KEY is unset"));
        assert!(after
            .detail
            .as_deref()
            .unwrap_or_default()
            .contains("next action"));
        assert!(after.resumable, "resumability drives explicit retry");

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert_eq!(runtime.wait_state, WaitState::ExternalBlocked);
        assert_eq!(runtime.activity, ActivityState::Idle);
        assert!(matches!(runtime.terminal, TerminalState::None));
        assert!(runtime.blocked_metadata.acceptance_stall);
        assert!(runtime.is_resumable_acceptance_stall());

        // Queue classification and operator routing read these sets.
        assert_eq!(
            state.externally_blocked_change_ids(),
            HashSet::from(["c".to_string()]),
            "dispatch suppression must survive the generic observation"
        );
        assert_eq!(
            state.acceptance_stalled_change_ids(),
            HashSet::from(["c".to_string()])
        );
    }

    /// The same precedence protects an Apply-observed external prerequisite:
    /// origin and every structured field survive, and the wait never degrades
    /// into an Acceptance-owned hold it never was.
    #[test]
    fn structured_blocker_metadata_survives_workspace_blocked_for_an_apply_external_wait() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ExecutionBlocked {
            change_id: "c".to_string(),
            blocker: apply_external_blocker(),
        });
        let before = state
            .blocker_view("c")
            .expect("a validated apply claim must establish a blocker view");

        observe_generic_blocked_workspace(&mut state, "c");

        assert_eq!(state.display_status("c"), "blocked");
        let after = state
            .blocker_view("c")
            .expect("the hold must survive the generic observation");
        assert_eq!(after, before);
        assert_eq!(after.kind, BlockerKind::External);
        assert_eq!(after.origin.as_deref(), Some("apply"));
        assert_eq!(after.prerequisite_owner.as_deref(), Some("platform"));
        assert_eq!(
            after.category.as_deref(),
            Some("external-blocked:infrastructure")
        );
        assert_eq!(
            after.unblock_condition.as_deref(),
            Some("cache.internal accepts connections again")
        );
        assert!(after.resumable);

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert_eq!(runtime.wait_state, WaitState::ExternalBlocked);
        assert!(
            !runtime.blocked_metadata.acceptance_stall,
            "an apply-origin external wait is not an Acceptance-owned hold"
        );
        assert_eq!(
            state.externally_blocked_change_ids(),
            HashSet::from(["c".to_string()]),
            "queue suppression uses the external set for either origin"
        );
        assert!(state.acceptance_stalled_change_ids().is_empty());
    }

    /// An Acceptance-owned non-external execution hold stays `stalled` and keeps
    /// its Acceptance ownership and resumability. Losing either would let the
    /// applied workspace be dispatched back through Acceptance automatically.
    #[test]
    fn structured_blocker_metadata_survives_workspace_blocked_for_an_acceptance_owned_stall() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let denial = crate::permission::classify_permission_denial(&[Some(
            "Read permission denied for /private/secret.txt",
        )])
        .expect("denial should classify");
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ExecutionBlocked {
            change_id: "c".to_string(),
            blocker: StalledBlocker::permission_denial("acceptance", &denial),
        });
        let before = state
            .blocker_view("c")
            .expect("an acceptance-owned hold must establish a blocker view");

        observe_generic_blocked_workspace(&mut state, "c");

        assert_eq!(state.display_status("c"), "stalled");
        let after = state
            .blocker_view("c")
            .expect("the hold must survive the generic observation");
        assert_eq!(after, before);
        assert_eq!(after.kind, BlockerKind::None);
        assert!(after.resumable);
        assert_eq!(
            after.category.as_deref(),
            Some("execution-blocked:permission:file_read")
        );
        assert!(after
            .detail
            .as_deref()
            .unwrap_or_default()
            .contains("permission"));

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert_eq!(runtime.wait_state, WaitState::Stalled);
        assert!(runtime.blocked_metadata.acceptance_stall);
        assert!(runtime.is_resumable_acceptance_stall());
        assert_eq!(
            state.acceptance_stalled_change_ids(),
            HashSet::from(["c".to_string()]),
            "ordinary dispatch stays suppressed until explicit retry or restart"
        );
        assert!(state.externally_blocked_change_ids().is_empty());
    }

    /// Without a structured classification the generic observation still owns
    /// the wait: the conservative `stalled` fallback that rejection and legacy
    /// apply handoffs depend on is unchanged, and it invents no external facts.
    #[test]
    fn structured_blocker_metadata_survives_workspace_blocked_fallback_stays_generic_without_a_hold(
    ) {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&crate::events::ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "apply".to_string(),
        });

        observe_generic_blocked_workspace(&mut state, "c");

        assert_eq!(state.display_status("c"), "stalled");
        let view = state
            .blocker_view("c")
            .expect("the fallback still records a blocker view");
        assert_eq!(view.kind, BlockerKind::None);
        assert!(view.unblock_condition.is_none());
        assert!(view.prerequisite_owner.is_none());
        assert!(view.origin.is_none());
        assert!(!view.resumable);
        assert_eq!(
            view.category.as_deref(),
            Some("apply reported recoverable blocker; workspace remains stalled")
        );

        let runtime = state.change_runtime("c").expect("runtime for c");
        assert_eq!(runtime.wait_state, WaitState::Stalled);
        assert!(!runtime.blocked_metadata.acceptance_stall);
        assert!(state.acceptance_stalled_change_ids().is_empty());
        assert!(state.externally_blocked_change_ids().is_empty());

        // The `mark_stalled` fallback keeps its own generic message too.
        let mut marked = OrchestratorState::new(vec!["d".to_string()], 0);
        marked.mark_stalled("d".to_string());
        observe_generic_blocked_workspace(&mut marked, "d");
        assert_eq!(marked.display_status("d"), "stalled");
        assert_eq!(
            marked
                .change_runtime("d")
                .unwrap()
                .blocked_metadata
                .blocker_reason
                .as_deref(),
            Some("change stalled with recoverable blocker")
        );
    }

    /// Duplicate and replayed generic observations stay idempotent, and the
    /// existing explicit-retry boundary still releases the hold afterwards.
    #[test]
    fn structured_blocker_metadata_survives_workspace_blocked_under_duplicate_delivery() {
        use crate::events::{ExecutionEvent, StalledBlocker};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
            change_id: "c".to_string(),
            blocker: StalledBlocker::acceptance_external("credential", "STAGING_API_KEY is unset"),
        });
        let before = state.blocker_view("c").expect("blocker view");

        for _ in 0..3 {
            observe_generic_blocked_workspace(&mut state, "c");
        }
        assert_eq!(state.blocker_view("c").as_ref(), Some(&before));
        assert_eq!(state.display_status("c"), "blocked");

        // Precedence protects metadata; it does not make a hold permanent.
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert_eq!(state.display_status("c"), "queued");
        assert!(state.blocker_view("c").is_none());
        assert!(state.externally_blocked_change_ids().is_empty());
    }

    #[test]
    fn test_rejection_review_block_transitions_to_blocked_with_metadata() {
        use crate::events::{ExecutionEvent, RejectionOutcome};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ProcessingStarted("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "c".to_string(),
            workspace_name: "ws-c".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });
        assert_eq!(state.display_status("c"), "rejecting");

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "c".to_string(),
            outcome: RejectionOutcome::Block,
        });

        let runtime = state
            .change_runtime("c")
            .expect("runtime for c after rejecting block");
        assert_eq!(state.display_status("c"), "stalled");
        assert_eq!(runtime.activity, ActivityState::Idle);
        assert_eq!(runtime.wait_state, WaitState::Stalled);
        assert!(matches!(runtime.terminal, TerminalState::None));
        assert_eq!(
            runtime.blocked_metadata.blocker_reason.as_deref(),
            Some("rejection review returned block; unresolved blocker remains")
        );
        assert_eq!(
            runtime.blocked_metadata.unblock_metadata.as_deref(),
            Some(
                "resolve unresolved blocker tasks in openspec/changes/<change_id>/tasks.md, then trigger explicit resume"
            )
        );
        assert_eq!(
            runtime.blocked_metadata.worktree_snapshot.as_deref(),
            Some("existing worktree and WIP context are preserved for stalled rejection review")
        );
    }

    #[test]
    fn test_rejection_review_resume_from_archived_workspace_context_sets_applying() {
        use crate::events::{ExecutionEvent, RejectionOutcome};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Resume path can start from an archived workspace that re-entered rejecting review.
        state.runtime_entry("c").wait_state = WaitState::MergeWait;
        state.apply_observation("c", WorkspaceObservation::WorkspaceArchived);
        assert_eq!(state.display_status("c"), "merge wait");

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "c".to_string(),
            workspace_name: "ws-c".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });
        assert_eq!(state.display_status("c"), "rejecting");

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "c".to_string(),
            outcome: RejectionOutcome::Resume,
        });

        assert_eq!(state.display_status("c"), "applying");
    }

    #[test]
    fn test_rejection_review_block_from_archived_workspace_context_sets_stalled() {
        use crate::events::{ExecutionEvent, RejectionOutcome};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.runtime_entry("c").wait_state = WaitState::MergeWait;
        state.apply_observation("c", WorkspaceObservation::WorkspaceArchived);
        assert_eq!(state.display_status("c"), "merge wait");

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "c".to_string(),
            workspace_name: "ws-c".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });
        assert_eq!(state.display_status("c"), "rejecting");

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "c".to_string(),
            outcome: RejectionOutcome::Block,
        });

        assert_eq!(state.display_status("c"), "stalled");
    }

    // -----------------------------------------------------------------------
    // Phase 2.4: apply_observation reconcile
    // -----------------------------------------------------------------------

    #[test]
    fn test_apply_observation_reconcile_merge_wait() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // WorkspaceArchived preserves an already-established concrete MergeWait.
        state.runtime_entry("c").wait_state = WaitState::MergeWait;
        state.apply_observation("c", WorkspaceObservation::WorkspaceArchived);
        assert_eq!(state.display_status("c"), "merge wait");

        // WorktreeNotAhead → clears MergeWait.
        state.apply_observation("c", WorkspaceObservation::WorktreeNotAhead);
        assert_eq!(state.display_status("c"), "not queued");

        // Active execution prevents observation from overwriting.
        state.runtime_entry("c").activity = ActivityState::Applying;
        state.apply_observation("c", WorkspaceObservation::WorkspaceArchived);
        // Still applying (not overwritten).
        assert_eq!(state.display_status("c"), "applying");
    }

    // -----------------------------------------------------------------------
    // Phase 2.5: idempotency and late-event precedence
    // -----------------------------------------------------------------------

    // -----------------------------------------------------------------------
    // Phase 5.2: changes_refreshed uses reducer observation path
    // -----------------------------------------------------------------------

    #[test]
    fn test_changes_refreshed_uses_reducer_observation_path() {
        use crate::events::ExecutionEvent;
        use std::collections::{HashMap, HashSet};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Simulate a ChangesRefreshed with c in merge_wait_ids after a concrete manual
        // MergeDeferred(auto_resumable=false) had already established MergeWait.
        state.runtime_entry("c").wait_state = WaitState::MergeWait;
        let mut merge_wait_ids = HashSet::new();
        merge_wait_ids.insert("c".to_string());
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids,
        });

        // The reducer should have set MergeWait via apply_observation.
        assert_eq!(state.display_status("c"), "merge wait");
    }

    #[test]
    fn test_change_rejected_clears_only_target_queue_intent() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);

        state.apply_command(ReducerCommand::AddToQueue("a".to_string()));
        state.apply_command(ReducerCommand::AddToQueue("b".to_string()));
        assert_eq!(state.display_status("a"), "queued");
        assert_eq!(state.display_status("b"), "queued");

        state.apply_execution_event(&ExecutionEvent::ChangeRejected {
            change_id: "a".to_string(),
            reason: "blocked".to_string(),
        });

        assert_eq!(state.display_status("a"), "rejected");
        assert_eq!(state.display_status("b"), "queued");
    }

    #[test]
    fn test_changes_refreshed_reactivates_rejected_change() {
        use crate::events::ExecutionEvent;
        use crate::openspec::{Change, ProposalMetadata};
        use std::collections::{HashMap, HashSet};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Previously rejected and with stale non-default runtime fields.
        let rt = state.runtime_entry("c");
        rt.terminal = TerminalState::Rejected("blocked".to_string());
        rt.activity = ActivityState::Rejecting;
        rt.wait_state = WaitState::MergeWait;
        rt.queue_intent = QueueIntent::Queued;

        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![Change {
                id: "c".to_string(),
                completed_tasks: 0,
                total_tasks: 1,
                last_modified: "now".to_string(),
                dependencies: Vec::new(),
                metadata: ProposalMetadata::default(),
            }],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::new(),
        });

        let rt = state.change_runtime("c").expect("runtime for c");
        assert!(matches!(rt.terminal, TerminalState::None));
        assert!(matches!(rt.activity, ActivityState::Idle));
        assert!(matches!(rt.wait_state, WaitState::None));
        assert!(matches!(rt.queue_intent, QueueIntent::NotQueued));
        assert_eq!(state.display_status("c"), "not queued");
    }

    // -----------------------------------------------------------------------
    // Phase 5.3: merge wait release after external merge
    // -----------------------------------------------------------------------

    #[test]
    fn test_merge_wait_release_after_external_merge() {
        use crate::events::ExecutionEvent;
        use std::collections::{HashMap, HashSet};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Put into MergeWait.
        state.runtime_entry("c").wait_state = WaitState::MergeWait;
        state.apply_observation("c", WorkspaceObservation::WorkspaceArchived);
        assert_eq!(state.display_status("c"), "merge wait");

        // Refreshed with c in worktree_not_ahead_ids → clears MergeWait.
        let mut not_ahead = HashSet::new();
        not_ahead.insert("c".to_string());
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: not_ahead,
            merge_wait_ids: HashSet::new(),
        });
        assert_eq!(state.display_status("c"), "not queued");
    }

    // -----------------------------------------------------------------------
    // Startup reconciliation: archived-but-not-yet-merged workspace evidence
    // -----------------------------------------------------------------------

    /// A `ChangesRefreshed` carrying only archived-but-not-yet-merged evidence.
    fn startup_refresh(merge_wait_ids: &[&str]) -> crate::events::ExecutionEvent {
        use std::collections::{HashMap, HashSet};

        crate::events::ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: merge_wait_ids.iter().map(|id| id.to_string()).collect(),
        }
    }

    #[test]
    fn startup_refresh_restores_merge_wait_for_a_fresh_idle_change() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);
        assert_eq!(state.display_status("alpha"), "not queued");

        state.apply_execution_event(&startup_refresh(&["alpha"]));

        assert_eq!(state.display_status("alpha"), "merge wait");
        assert!(matches!(
            state.change_runtime("alpha").unwrap().wait_state,
            WaitState::MergeWait
        ));
        assert!(
            state.resolve_wait_change_ids().is_empty(),
            "restoring the wait must not enqueue scheduler-owned resolve work"
        );
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn startup_refresh_merge_wait_restoration_is_idempotent() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&startup_refresh(&["alpha"]));
        state.apply_execution_event(&startup_refresh(&["alpha"]));

        assert_eq!(state.display_status("alpha"), "merge wait");
        assert!(state.resolve_wait_change_ids().is_empty());
    }

    #[test]
    fn startup_refresh_does_not_regress_stronger_reducer_state_to_merge_wait() {
        use crate::events::ExecutionEvent;

        // Drives a fresh state through the reducer transitions a case needs.
        type CaseSetup = Box<dyn Fn(&mut OrchestratorState)>;

        // Each case is set up through the same reducer transitions production uses,
        // then re-observed with the change still reported as archived-but-not-merged.
        let cases: Vec<(&str, &str, CaseSetup)> = vec![
            (
                "resolving",
                "resolving",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_execution_event(&ExecutionEvent::ChangeArchived("x".to_string()));
                }),
            ),
            (
                "resolve pending",
                "resolve pending",
                Box::new(|state: &mut OrchestratorState| {
                    state.runtime_entry("x").wait_state = WaitState::MergeWait;
                    state.apply_command(ReducerCommand::ResolveMerge("x".to_string()));
                }),
            ),
            (
                "rejecting",
                "rejecting",
                Box::new(|state: &mut OrchestratorState| {
                    state.runtime_entry("x").activity = ActivityState::Rejecting;
                }),
            ),
            (
                "reject pending",
                "reject pending",
                Box::new(|state: &mut OrchestratorState| {
                    // Rejection review only waits while another change holds the
                    // single base-mutating lane.
                    state.runtime_entry("lane").activity = ActivityState::Resolving;
                    state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
                        change_id: "x".to_string(),
                        workspace_name: "ws-x".to_string(),
                        status: crate::vcs::WorkspaceStatus::Rejecting,
                    });
                }),
            ),
            (
                "queued",
                "queued",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_command(ReducerCommand::AddToQueue("x".to_string()));
                }),
            ),
            (
                "merged",
                "merged",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_execution_event(&ExecutionEvent::MergeCompleted {
                        change_id: "x".to_string(),
                        revision: "rev-x".to_string(),
                    });
                }),
            ),
            (
                "rejected",
                "rejected",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_execution_event(&ExecutionEvent::ChangeRejected {
                        change_id: "x".to_string(),
                        reason: "blocked".to_string(),
                    });
                }),
            ),
            (
                "error",
                "error",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_execution_event(&ExecutionEvent::ProcessingError {
                        id: "x".to_string(),
                        error: "boom".to_string(),
                    });
                }),
            ),
            (
                "stopped",
                "stopped",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_command(ReducerCommand::StopChange("x".to_string()));
                }),
            ),
            (
                "blocked",
                "blocked",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_execution_event(&ExecutionEvent::DependencyBlocked {
                        change_id: "x".to_string(),
                        dependency_ids: vec!["dep".to_string()],
                    });
                }),
            ),
            (
                "dequeued",
                "not queued",
                Box::new(|state: &mut OrchestratorState| {
                    state.apply_command(ReducerCommand::AddToQueue("x".to_string()));
                    state.apply_command(ReducerCommand::DequeueChange("x".to_string()));
                }),
            ),
        ];

        for (label, expected, setup) in cases {
            // `lane` exists only so a case can occupy the single base-mutating lane.
            let mut state = OrchestratorState::new(vec!["x".to_string(), "lane".to_string()], 0);
            setup(&mut state);
            assert_eq!(state.display_status("x"), expected, "setup for {}", label);
            let resolve_wait_before = state.resolve_wait_change_ids();
            let reject_wait_before = state.reject_wait_change_ids();

            state.apply_execution_event(&startup_refresh(&["x"]));

            assert_eq!(
                state.display_status("x"),
                expected,
                "refresh evidence must not regress {} to merge wait",
                label
            );
            assert_eq!(
                state.resolve_wait_change_ids(),
                resolve_wait_before,
                "refresh must not create a duplicate manual resolve reservation for {}",
                label
            );
            assert_eq!(
                state.reject_wait_change_ids(),
                reject_wait_before,
                "refresh must not disturb reject-wait membership for {}",
                label
            );
            assert!(state.global_invariants_hold(), "invariants for {}", label);
        }
    }

    // -----------------------------------------------------------------------
    // Phase 5.4: WorkspaceState::Archived recovers MergeWait (not ResolveWait)
    // -----------------------------------------------------------------------

    #[test]
    fn test_workspace_archived_recovers_merge_wait() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // WorkspaceArchived observation preserves concrete MergeWait evidence.
        state.runtime_entry("c").wait_state = WaitState::MergeWait;
        state.apply_observation("c", WorkspaceObservation::WorkspaceArchived);
        assert_eq!(state.display_status("c"), "merge wait");
        assert!(
            matches!(
                state.change_runtime("c").unwrap().wait_state,
                WaitState::MergeWait
            ),
            "observation should set MergeWait, not ResolveWait"
        );
    }

    // -----------------------------------------------------------------------
    // Phase 5.5: queue-added change not overwritten by MergeWait refresh
    // -----------------------------------------------------------------------

    #[test]
    fn test_queue_add_not_overwritten_by_merge_wait_refresh() {
        use crate::events::ExecutionEvent;
        use std::collections::{HashMap, HashSet};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Queue the change.
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert_eq!(state.display_status("c"), "queued");

        // A bare archived workspace observation is only a milestone and must not
        // create MergeWait without concrete MergeDeferred(auto_resumable=false) evidence.
        state.apply_observation("c", WorkspaceObservation::WorkspaceArchived);
        assert_eq!(state.display_status("c"), "queued");

        // WorktreeNotAhead remains idempotent when no MergeWait exists.
        let mut not_ahead = HashSet::new();
        not_ahead.insert("c".to_string());
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: not_ahead,
            merge_wait_ids: HashSet::new(),
        });
        // After clearing MergeWait, the queue intent (Queued) is visible again.
        assert_eq!(state.display_status("c"), "queued");
    }

    #[test]
    fn test_base_mutating_lane_exclusivity_and_wait_membership() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(
            vec![
                "resolving-a".to_string(),
                "rejecting-b".to_string(),
                "archive-c".to_string(),
            ],
            0,
        );

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "resolving-a".to_string(),
            workspace_name: "ws-a".to_string(),
            status: crate::vcs::WorkspaceStatus::Resolving,
        });
        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "rejecting-b".to_string(),
            workspace_name: "ws-b".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });

        assert_eq!(state.display_status("resolving-a"), "resolving");
        assert_eq!(state.display_status("rejecting-b"), "reject pending");
        assert_eq!(
            state.reject_wait_change_ids(),
            vec!["rejecting-b".to_string()]
        );
        assert!(state.resolve_wait_change_ids().is_empty());
        assert!(state.has_other_post_archive_lane_blocker("archive-c"));
        assert!(state.global_invariants_hold());

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("archive-c".to_string()));
        assert_eq!(state.display_status("archive-c"), "resolve pending");
        assert_eq!(
            state.resolve_wait_change_ids(),
            vec!["archive-c".to_string()]
        );
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn test_reject_wait_clear_and_deterministic_single_promotion() {
        use crate::events::{ExecutionEvent, RejectionOutcome};

        let mut state = OrchestratorState::new(
            vec![
                "lane-a".to_string(),
                "resolve-b".to_string(),
                "reject-c".to_string(),
            ],
            0,
        );

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "lane-a".to_string(),
            workspace_name: "ws-a".to_string(),
            status: crate::vcs::WorkspaceStatus::Resolving,
        });
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("resolve-b".to_string()));
        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "reject-c".to_string(),
            workspace_name: "ws-c".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });

        assert_eq!(state.display_status("resolve-b"), "resolve pending");
        assert_eq!(state.display_status("reject-c"), "reject pending");

        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "lane-a".to_string(),
            error: "manual blocker".to_string(),
        });
        let promoted = state.promote_next_base_mutating_lane_waiter();
        assert_eq!(
            promoted,
            Some(("resolve-b".to_string(), WaitState::ResolveWait))
        );
        assert_eq!(state.display_status("resolve-b"), "resolving");
        assert_eq!(state.display_status("reject-c"), "reject pending");
        assert!(state.global_invariants_hold());

        assert_eq!(state.promote_next_base_mutating_lane_waiter(), None);
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "resolve-b".to_string(),
            revision: "rev-b".to_string(),
        });
        let promoted = state.promote_next_base_mutating_lane_waiter();
        assert_eq!(
            promoted,
            Some(("reject-c".to_string(), WaitState::RejectWait))
        );
        assert_eq!(state.display_status("reject-c"), "rejecting");
        assert!(state.reject_wait_change_ids().is_empty());
        assert!(state.global_invariants_hold());

        state.apply_execution_event(&ExecutionEvent::RejectionReviewCompleted {
            change_id: "reject-c".to_string(),
            outcome: RejectionOutcome::Resume,
        });
        assert!(state.reject_wait_change_ids().is_empty());
        assert_eq!(state.display_status("reject-c"), "applying");
    }

    // -----------------------------------------------------------------------
    // Phase 4.3: parallel merge events drive reducer wait states
    // -----------------------------------------------------------------------

    #[test]
    fn test_parallel_change_archived_no_blocker_enters_resolving_not_merge_wait() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));

        assert_eq!(state.display_status("c"), "resolving");
        assert!(
            state.resolve_wait_change_ids().is_empty(),
            "no-blocker archive completion must enter immediate merge handling, not resolve wait"
        );
        assert!(
            state.queued_change_ids().is_empty(),
            "no-blocker archive completion must not be reintroduced as ordinary queued work"
        );
    }

    #[test]
    fn test_no_blocker_merge_wait_to_merged_vibration_regression() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));
        assert_eq!(state.display_status("c"), "resolving");

        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["c".to_string()]),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::from(["c".to_string()]),
        });
        assert_eq!(
            state.display_status("c"),
            "resolving",
            "workspace refresh must not turn active no-blocker merge handling into merge wait"
        );

        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");
    }

    #[test]
    fn test_merged_archived_vibration_regression() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));
        assert_eq!(state.display_status("c"), "merged");

        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["c".to_string()]),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::from(["c".to_string()]),
        });
        assert_eq!(state.display_status("c"), "merged");
    }

    #[test]
    fn test_manual_merge_deferred_clears_normal_queue_intent() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert_eq!(state.queued_change_ids(), vec!["c".to_string()]);

        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });

        assert_eq!(state.display_status("c"), "merge wait");
        assert!(
            state.queued_change_ids().is_empty(),
            "manual merge deferral must consume ordinary queue intent"
        );
        assert!(
            state.resolve_wait_change_ids().is_empty(),
            "manual merge deferral must not remain scheduler-owned resolve retry intent"
        );
    }

    #[test]
    fn test_manual_merge_deferred_clears_existing_resolve_wait_queue_membership() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "another merge is in progress".to_string(),
            auto_resumable: true,
        });
        assert_eq!(state.display_status("c"), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec!["c".to_string()]);

        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });

        assert_eq!(state.display_status("c"), "merge wait");
        assert!(state.queued_change_ids().is_empty());
        assert!(state.resolve_wait_change_ids().is_empty());
    }

    #[test]
    fn test_manual_merge_deferred_resolve_merge_explicit_retry_sets_resolve_wait() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status("c"), "merge wait");
        assert!(state.queued_change_ids().is_empty());

        let outcome = state.apply_command(ReducerCommand::ResolveMerge("c".to_string()));

        assert!(matches!(outcome, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec!["c".to_string()]);
        assert!(
            state.queued_change_ids().is_empty(),
            "explicit merge retry must use resolve-wait intent, not normal queue intent"
        );
    }

    /// Archive is never terminal on its own: it enters active post-archive
    /// handling, so an explicit manual retry there changes nothing, while a
    /// concrete manual wait is still admitted as scheduler-consumable intent.
    #[test]
    fn test_archived_change_enters_active_merge_handling_before_manual_retry() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&crate::events::ExecutionEvent::ChangeArchived(
            "alpha".to_string(),
        ));
        assert_eq!(
            state.display_status("alpha"),
            "resolving",
            "archive alone must never be terminal"
        );

        assert!(
            matches!(
                state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string())),
                ReduceOutcome::NoOp
            ),
            "a row already in active merge handling needs no manual retry"
        );
        assert_eq!(state.display_status("alpha"), "resolving");

        // A concrete manual deferral is what makes the retry meaningful.
        state.apply_execution_event(&crate::events::ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "manual resolution required".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status("alpha"), "merge wait");

        let outcome = state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));

        assert!(
            matches!(outcome, ReduceOutcome::Changed(_)),
            "explicit manual retry for a merge-wait row must not be dropped"
        );
        assert_eq!(state.display_status("alpha"), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec!["alpha".to_string()]);
        assert!(state.queued_change_ids().is_empty());
    }

    #[test]
    fn test_merged_manual_retry_remains_noop() {
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&crate::events::ExecutionEvent::MergeCompleted {
            change_id: "alpha".to_string(),
            revision: "rev-alpha".to_string(),
        });

        let outcome = state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));

        assert!(matches!(outcome, ReduceOutcome::NoOp));
        assert_eq!(state.display_status("alpha"), "merged");
        assert!(state.resolve_wait_change_ids().is_empty());
    }

    #[test]
    fn test_auto_resumable_merge_deferred_keeps_scheduler_retry_intent() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "another merge is in progress".to_string(),
            auto_resumable: true,
        });

        assert_eq!(state.display_status("c"), "resolve pending");
        assert_eq!(state.queued_change_ids(), vec!["c".to_string()]);
        assert_eq!(state.resolve_wait_change_ids(), vec!["c".to_string()]);
    }

    #[test]
    fn test_archive_merge_defers_to_resolve_pending_when_rejecting_active() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "a".to_string(),
            workspace_name: "ws-a".to_string(),
            status: crate::vcs::WorkspaceStatus::Rejecting,
        });
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("b".to_string()));

        assert_eq!(state.display_status("a"), "rejecting");
        assert_eq!(state.display_status("b"), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec!["b".to_string()]);
        assert!(state.reject_wait_change_ids().is_empty());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn test_active_applying_does_not_create_resolve_pending_on_archive() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "a".to_string(),
            command: "apply".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("b".to_string()));

        assert_eq!(state.display_status("a"), "applying");
        assert_eq!(state.display_status("b"), "resolving");
        assert!(state.resolve_wait_change_ids().is_empty());
    }

    #[test]
    fn test_parallel_merge_events_drive_reducer_wait_states() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // MergeDeferred (manual, not auto-resumable) → MergeWait
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status("c"), "merge wait");

        // ResolveStarted → Resolving (clears MergeWait)
        state.apply_execution_event(&ExecutionEvent::ResolveStarted {
            change_id: "c".to_string(),
            command: "resolve".to_string(),
        });
        assert_eq!(state.display_status("c"), "resolving");

        // ResolveCompleted → Merged (terminal): successful resolve means the change is merged.
        state.apply_execution_event(&ExecutionEvent::ResolveCompleted {
            change_id: "c".to_string(),
            worktree_change_ids: None,
        });
        assert_eq!(state.display_status("c"), "merged");

        // MergeCompleted is idempotent once already terminal (parallel orchestrator path).
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");

        // ResolveFailed after Merged must NOT regress.
        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "c".to_string(),
            error: "late".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");
    }

    // -----------------------------------------------------------------------
    // Regression: stale ResolveWait must not survive ResolveCompleted + refresh
    // -----------------------------------------------------------------------
    //
    // Scenario: base dirtiness caused MergeWait, user triggered manual resolve
    // (ResolveMerge command → ResolveWait), resolve succeeded, but the row was
    // previously stuck at "resolve pending" because ResolveCompleted was not
    // applied to the shared reducer before the next ChangesRefreshed.

    #[test]
    fn test_merge_completed_ignores_later_stale_manual_merge_deferred() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "alpha".to_string(),
            revision: "rev-alpha".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "Archive incomplete for 'alpha': worktree may be dirty".to_string(),
            auto_resumable: false,
        });

        assert_eq!(state.display_status("alpha"), "merged");
        assert!(state.queued_change_ids().is_empty());
        assert!(state.resolve_wait_change_ids().is_empty());
    }

    #[test]
    fn test_merge_completed_clears_resolve_wait_intent() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "Resolve in progress for another change".to_string(),
            auto_resumable: true,
        });
        assert_eq!(state.resolve_wait_change_ids(), vec!["alpha".to_string()]);

        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "alpha".to_string(),
            revision: "rev-alpha".to_string(),
        });

        assert_eq!(state.display_status("alpha"), "merged");
        assert!(
            state.resolve_wait_change_ids().is_empty(),
            "MergeCompleted must clear reducer-owned ResolveWait retry intent"
        );
    }

    #[test]
    fn test_clear_resolve_wait_intent_removes_retry_without_terminal_transition() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "Resolve in progress for another change".to_string(),
            auto_resumable: true,
        });
        assert_eq!(state.resolve_wait_change_ids(), vec!["alpha".to_string()]);

        state.clear_resolve_wait_intent("alpha");

        assert!(state.resolve_wait_change_ids().is_empty());
        // Releasing the reservation records no outcome, so the change falls back
        // to the retryable manual wait instead of an idle `not queued` row.
        assert_eq!(state.display_status("alpha"), "merge wait");
        assert!(!state.is_terminal_change("alpha"));
    }

    /// Reducer settlement table for a stale deferred resolve retry.
    ///
    /// The failure this guards is concrete: `ReducerCommand::ResolveMerge` leaves
    /// `QueueIntent::NotQueued` behind the `ResolveWait`, so clearing the wait on
    /// its own used to display `not queued` for a change the base tree already
    /// contains.
    fn manual_resolve_retry_state(change_id: &str) -> OrchestratorState {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec![change_id.to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: change_id.to_string(),
            reason: "manual conflict".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status(change_id), "merge wait");
        state.apply_command(ReducerCommand::ResolveMerge(change_id.to_string()));
        assert_eq!(state.display_status(change_id), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec![change_id.to_string()]);
        state
    }

    #[test]
    fn stale_deferred_merge_retry_with_proven_base_integration_settles_merged() {
        let mut state = manual_resolve_retry_state("alpha");

        let settlement = state.settle_stale_resolve_retry("alpha", StaleResolveEvidence::Proven);

        assert_eq!(settlement, StaleResolveSettlement::Merged);
        assert_eq!(state.display_status("alpha"), "merged");
        assert_ne!(state.display_status("alpha"), "not queued");
        assert!(state.is_terminal_change("alpha"));
        assert!(state.resolve_wait_change_ids().is_empty());
        assert!(state.resolve_wait_queue.is_empty());
        assert!(!state.is_base_mutating_lane_occupied());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn stale_deferred_merge_retry_without_proven_integration_retains_merge_wait() {
        for evidence in [StaleResolveEvidence::Absent, StaleResolveEvidence::Unknown] {
            let mut state = manual_resolve_retry_state("alpha");

            let settlement = state.settle_stale_resolve_retry("alpha", evidence);

            assert_eq!(
                settlement,
                StaleResolveSettlement::MergeWaitRetained,
                "{evidence:?} proves nothing and must stay retryable"
            );
            assert_eq!(
                state.display_status("alpha"),
                "merge wait",
                "{evidence:?} must not expose not queued or merged"
            );
            assert!(!state.is_terminal_change("alpha"));
            assert!(state.resolve_wait_change_ids().is_empty());
            assert!(state.resolve_wait_queue.is_empty());
            assert!(state.global_invariants_hold());

            // The operator can still retry explicitly from the retained state.
            state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));
            assert_eq!(state.display_status("alpha"), "resolve pending");
        }
    }

    #[test]
    fn stale_deferred_merge_retry_settles_promoted_lane_occupant() {
        let mut state = manual_resolve_retry_state("alpha");
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("alpha".to_string(), WaitState::ResolveWait))
        );
        assert!(state.is_base_mutating_lane_occupied());

        let settlement = state.settle_stale_resolve_retry("alpha", StaleResolveEvidence::Absent);

        assert_eq!(settlement, StaleResolveSettlement::MergeWaitRetained);
        assert_eq!(state.display_status("alpha"), "merge wait");
        assert!(
            !state.is_base_mutating_lane_occupied(),
            "settlement releases base-lane ownership once the reducer state agrees"
        );
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn stale_deferred_merge_retry_preserves_immutable_terminal_outcomes() {
        let mut state = manual_resolve_retry_state("alpha");
        state.apply_command(ReducerCommand::StopChange("alpha".to_string()));
        assert_eq!(state.display_status("alpha"), "stopped");

        let settlement = state.settle_stale_resolve_retry("alpha", StaleResolveEvidence::Proven);

        assert_eq!(settlement, StaleResolveSettlement::AlreadySettled);
        assert_eq!(state.display_status("alpha"), "stopped");
        assert!(state.resolve_wait_change_ids().is_empty());
        assert!(state.resolve_wait_queue.is_empty());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn stale_deferred_merge_retry_keeps_bounded_resolve_failure_in_merge_wait() {
        use crate::events::ExecutionEvent;

        let mut state = manual_resolve_retry_state("alpha");
        assert_eq!(
            state.promote_next_base_mutating_lane_waiter(),
            Some(("alpha".to_string(), WaitState::ResolveWait))
        );

        // Ordinary bounded exhaustion is unchanged by stale settlement: one
        // change-scoped ResolveFailed returns the change to manual merge wait.
        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "alpha".to_string(),
            error: "bounded resolve exhausted".to_string(),
        });

        assert_eq!(state.display_status("alpha"), "merge wait");
        assert!(!state.is_terminal_change("alpha"));
        assert!(state.resolve_wait_change_ids().is_empty());
        assert!(!state.is_base_mutating_lane_occupied());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn test_resolve_completed_clears_resolve_wait_and_survives_refresh() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Step 1: concrete manual merge blocker -> enters MergeWait.
        // A bare ChangeArchived event now truthfully enters active post-archive
        // merge handling; only MergeDeferred(auto_resumable=false) represents
        // operator-owned manual merge wait.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status("c"), "merge wait");

        // Step 2: user triggers manual resolve → reducer transitions to ResolveWait.
        state.apply_command(ReducerCommand::ResolveMerge("c".to_string()));
        assert_eq!(state.display_status("c"), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec!["c".to_string()]);

        // Step 3: a refresh still sees merge_wait/worktree branch and emits observations.
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: std::collections::HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: std::collections::HashSet::new(),
            worktree_change_ids: std::collections::HashSet::from(["c".to_string()]),
            worktree_paths: std::collections::HashMap::new(),
            worktree_not_ahead_ids: std::collections::HashSet::new(),
            merge_wait_ids: std::collections::HashSet::from(["c".to_string()]),
        });
        assert_eq!(state.display_status("c"), "resolve pending");

        // Step 4: resolve succeeds → must transition to merged, not stall at resolve pending.
        state.apply_execution_event(&ExecutionEvent::ResolveCompleted {
            change_id: "c".to_string(),
            worktree_change_ids: None,
        });
        assert_eq!(state.display_status("c"), "merged");
        assert!(
            state.resolve_wait_change_ids().is_empty(),
            "resolve completion must clear queued resolve intent"
        );

        // Step 5: another refresh still seeing stale merge_wait/worktree signals
        // must NOT regress the row back to "resolve pending" or "merge wait".
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: std::collections::HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: std::collections::HashSet::new(),
            worktree_change_ids: std::collections::HashSet::from(["c".to_string()]),
            worktree_paths: std::collections::HashMap::new(),
            worktree_not_ahead_ids: std::collections::HashSet::new(),
            merge_wait_ids: std::collections::HashSet::from(["c".to_string()]),
        });
        assert_eq!(state.display_status("c"), "merged");
        assert_ne!(
            state.display_status("c"),
            "resolve pending",
            "row must not regress to resolve pending after successful resolve + refresh"
        );
    }

    #[test]
    fn test_resolve_wait_manual_merge_deferred_demotes_to_merge_wait() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["change-a".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "change-a".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        state.apply_command(ReducerCommand::ResolveMerge("change-a".to_string()));
        assert_eq!(state.display_status("change-a"), "resolve pending");
        assert_eq!(
            state.resolve_wait_change_ids(),
            vec!["change-a".to_string()]
        );

        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "change-a".to_string(),
            reason: "Working tree has uncommitted changes".to_string(),
            auto_resumable: false,
        });

        assert_eq!(state.display_status("change-a"), "merge wait");
        assert!(
            state.resolve_wait_change_ids().is_empty(),
            "manual retry deferral must remove reducer-owned ResolveWait membership"
        );
        assert!(state.queued_change_ids().is_empty());
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn test_resolve_wait_clean_lane_promotion_promotes_exactly_one_waiter() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["alpha".to_string(), "beta".to_string()], 0);

        for change_id in ["alpha", "beta"] {
            state.apply_execution_event(&ExecutionEvent::MergeDeferred {
                change_id: change_id.to_string(),
                reason: "base dirty".to_string(),
                auto_resumable: false,
            });
            state.apply_command(ReducerCommand::ResolveMerge(change_id.to_string()));
        }

        let promoted = state.promote_next_base_mutating_lane_waiter();

        assert_eq!(
            promoted,
            Some(("alpha".to_string(), WaitState::ResolveWait))
        );
        assert_eq!(state.display_status("alpha"), "resolving");
        assert_eq!(state.display_status("beta"), "resolve pending");
        assert_eq!(state.resolve_wait_change_ids(), vec!["beta".to_string()]);
        assert_eq!(state.promote_next_base_mutating_lane_waiter(), None);
        assert!(state.global_invariants_hold());
    }

    #[test]
    fn test_resolve_failed_restores_merge_wait() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Reach MergeWait via manual-intervention deferral, then promote to ResolveWait.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        state.apply_command(ReducerCommand::ResolveMerge("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::ResolveStarted {
            change_id: "c".to_string(),
            command: "resolve".to_string(),
        });

        // Resolve fails → must return to merge wait, not stay queued/idle.
        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "c".to_string(),
            error: "conflict".to_string(),
        });
        assert_eq!(state.display_status("c"), "merge wait");

        // A subsequent RefreshFailed-after-terminal must not regress a Merged entry.
        let mut state2 = OrchestratorState::new(vec!["c".to_string()], 0);
        state2.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        state2.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "c".to_string(),
            error: "late".to_string(),
        });
        assert_eq!(state2.display_status("c"), "merged");
    }

    // -----------------------------------------------------------------------
    // Phase 4.4: late events after stop do not regress state
    // -----------------------------------------------------------------------

    #[test]
    fn test_workspace_status_update_targets_explicit_change_id() {
        use crate::events::ExecutionEvent;
        use crate::vcs::WorkspaceStatus;

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "a".to_string(),
            command: "apply-a".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "b".to_string(),
            command: "apply-b".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "b".to_string(),
            workspace_name: "ws-b".to_string(),
            status: WorkspaceStatus::Rejecting,
        });

        assert_eq!(state.display_status("a"), "applying");
        assert_eq!(state.display_status("b"), "rejecting");
    }

    #[test]
    fn test_late_events_after_stop_do_not_regress_state() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Dequeue the change.
        state.apply_execution_event(&ExecutionEvent::ChangeDequeued {
            change_id: "c".to_string(),
        });
        assert_eq!(state.display_status("c"), "not queued");

        // Late ApplyStarted must not regress from dequeued state.
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        assert_eq!(state.display_status("c"), "not queued");

        // Late AcceptanceStarted must not regress from dequeued state.
        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        assert_eq!(state.display_status("c"), "not queued");

        // Late ProcessingError must not regress from dequeued state.
        state.apply_execution_event(&ExecutionEvent::ProcessingError {
            id: "c".to_string(),
            error: "late error".to_string(),
        });
        assert_eq!(state.display_status("c"), "not queued");
    }

    // -----------------------------------------------------------------------
    // Phase 2.5: idempotency and late-event precedence
    // -----------------------------------------------------------------------

    #[test]
    fn test_reducer_idempotency_and_precedence() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Archive, then complete the post-archive merge that terminates it.
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));
        assert_eq!(state.display_status("c"), "resolving");
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");

        // Late ResolveFailed must NOT regress the terminal state.
        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "c".to_string(),
            error: "late".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");

        // Duplicate ApplyStarted on a terminal change must be no-op.
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        // Terminal wins: still merged.
        assert_eq!(state.display_status("c"), "merged");
    }

    /// Phase 7.2: reducer runtime state and legacy aggregates (pending_changes, archived_changes,
    /// current_change_id, apply_count) must agree on the canonical status of each change
    /// after a representative lifecycle sequence.
    #[test]
    fn test_reducer_runtime_and_legacy_aggregates_stay_consistent() {
        use crate::events::ExecutionEvent;

        let mut state =
            OrchestratorState::new(vec!["a".to_string(), "b".to_string(), "c".to_string()], 5);

        // ── Initial state ──────────────────────────────────────────────────
        // All changes start in pending_changes (from new()), but reducer says "not queued"
        // because no AddToQueue command has been issued yet.
        assert_eq!(state.display_status("a"), "not queued");
        assert_eq!(state.display_status("b"), "not queued");
        assert_eq!(state.display_status("c"), "not queued");
        assert!(state.is_pending("a"));
        assert!(state.is_pending("b"));
        assert!(state.is_pending("c"));

        // ── Queue a and b via reducer ──────────────────────────────────────
        state.apply_command(ReducerCommand::AddToQueue("a".to_string()));
        state.apply_command(ReducerCommand::AddToQueue("b".to_string()));

        assert_eq!(state.display_status("a"), "queued");
        assert_eq!(state.display_status("b"), "queued");
        assert!(state.is_pending("a"));
        assert!(state.is_pending("b"));

        // ── Start applying 'a' ─────────────────────────────────────────────
        state.set_current_change(Some("a".to_string()));
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "a".to_string(),
            command: "cmd".to_string(),
        });
        state.increment_apply_count("a");

        assert_eq!(state.display_status("a"), "applying");
        assert_eq!(state.current_change_id(), Some(&"a".to_string()));
        assert_eq!(state.apply_count("a"), 1);

        // ── Archive 'a' ────────────────────────────────────────────────────
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("a".to_string()));
        state.mark_archived("a");

        // Archive enters post-archive handling; the legacy archived set agrees
        // that the archive itself happened.
        assert_eq!(state.display_status("a"), "resolving");
        assert!(state.is_archived("a"));
        assert!(!state.is_pending("a"));

        // ── Dequeue 'b' ────────────────────────────────────────────────────
        state.apply_command(ReducerCommand::DequeueChange("b".to_string()));

        // Reducer returns b to not queued; legacy pending membership is unchanged.
        assert_eq!(state.display_status("b"), "not queued");
        // 'c' (never explicitly queued in reducer) is still "not queued" in reducer.
        assert_eq!(state.display_status("c"), "not queued");
        assert!(state.is_pending("c")); // legacy: still in pending set
        assert!(!state.is_archived("c"));
    }

    // -----------------------------------------------------------------------
    // Archive-to-post-archive transitions (the sole execution path)
    // -----------------------------------------------------------------------

    #[test]
    fn test_parallel_mode_change_archived_transitions_to_resolving_without_blocker() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Apply → Archive
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ApplyCompleted {
            change_id: "c".to_string(),
            revision: "rev1".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ArchiveStarted {
            change_id: "c".to_string(),
            command: "archive".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));

        // ChangeArchived is a merge-routing milestone, not a
        // stable terminal display. With no lane blocker or manual deferral
        // evidence, the reducer must show active merge handling immediately.
        assert_eq!(
            state.display_status("c"),
            "resolving",
            "Parallel: no-blocker ChangeArchived must transition to active merge handling, not merge wait"
        );
        assert!(
            !state.is_terminal_change("c"),
            "Parallel: change must not be terminal after archive"
        );

        // MergeCompleted should now succeed.
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "merge-rev".to_string(),
        });
        assert_eq!(
            state.display_status("c"),
            "merged",
            "Parallel: MergeCompleted must transition to merged terminal"
        );
        assert!(state.is_terminal_change("c"));
    }

    #[test]
    fn test_parallel_mode_archive_success_clears_prior_acceptance_error() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::AcceptanceFailed {
            change_id: "alpha".to_string(),
            error: "transient acceptance failure".to_string(),
        });
        assert_eq!(state.display_status("alpha"), "error");

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("alpha".to_string()));

        assert_eq!(
            state.display_status("alpha"),
            "resolving",
            "same-change archive success must supersede a recoverable acceptance error and enter active no-blocker merge handling"
        );
        assert!(!state.is_terminal_change("alpha"));

        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "alpha".to_string(),
            revision: "merge-rev".to_string(),
        });
        assert_eq!(state.display_status("alpha"), "merged");
        assert!(state.is_terminal_change("alpha"));
    }

    #[test]
    fn test_parallel_mode_acceptance_error_archive_then_merge_finishes_merged() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["add-skill-secret-ingestion".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::AcceptanceFailed {
            change_id: "add-skill-secret-ingestion".to_string(),
            error: "acceptance failed before fix".to_string(),
        });
        assert_eq!(state.display_status("add-skill-secret-ingestion"), "error");

        state.apply_execution_event(&ExecutionEvent::ChangeArchived(
            "add-skill-secret-ingestion".to_string(),
        ));
        assert_ne!(state.display_status("add-skill-secret-ingestion"), "error");

        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "add-skill-secret-ingestion".to_string(),
            revision: "merge-rev".to_string(),
        });

        assert_eq!(state.display_status("add-skill-secret-ingestion"), "merged");
        assert!(state.is_terminal_change("add-skill-secret-ingestion"));
    }

    #[test]
    fn test_merge_and_resolve_success_clear_prior_processing_error_but_not_rejected() {
        use crate::events::ExecutionEvent;

        let mut merge_state = OrchestratorState::new(vec!["alpha".to_string()], 0);
        merge_state.apply_execution_event(&ExecutionEvent::ProcessingError {
            id: "alpha".to_string(),
            error: "recoverable process failure".to_string(),
        });
        assert_eq!(merge_state.display_status("alpha"), "error");
        merge_state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "alpha".to_string(),
            revision: "merge-rev".to_string(),
        });
        assert_eq!(merge_state.display_status("alpha"), "merged");

        let mut resolve_state = OrchestratorState::new(vec!["alpha".to_string()], 0);
        resolve_state.apply_execution_event(&ExecutionEvent::ProcessingError {
            id: "alpha".to_string(),
            error: "recoverable process failure".to_string(),
        });
        assert_eq!(resolve_state.display_status("alpha"), "error");
        resolve_state.apply_execution_event(&ExecutionEvent::ResolveCompleted {
            change_id: "alpha".to_string(),
            worktree_change_ids: None,
        });
        assert_eq!(resolve_state.display_status("alpha"), "merged");

        let mut rejected_state = OrchestratorState::new(vec!["alpha".to_string()], 0);
        rejected_state.apply_execution_event(&ExecutionEvent::ChangeRejected {
            change_id: "alpha".to_string(),
            reason: "final rejection".to_string(),
        });
        rejected_state.apply_execution_event(&ExecutionEvent::ChangeArchived("alpha".to_string()));
        rejected_state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "alpha".to_string(),
            revision: "stale-merge-rev".to_string(),
        });
        rejected_state.apply_execution_event(&ExecutionEvent::ResolveCompleted {
            change_id: "alpha".to_string(),
            worktree_change_ids: None,
        });
        assert_eq!(rejected_state.display_status("alpha"), "rejected");
    }

    #[test]
    fn test_parallel_mode_change_archived_uses_resolve_pending_when_other_change_is_resolving() {
        use crate::events::ExecutionEvent;

        let mut state =
            OrchestratorState::new(vec!["resolving".to_string(), "archived".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::ResolveStarted {
            change_id: "resolving".to_string(),
            command: "resolve resolving".to_string(),
        });
        assert_eq!(state.display_status("resolving"), "resolving");

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("archived".to_string()));
        assert_eq!(
            state.display_status("archived"),
            "resolve pending",
            "Parallel: ChangeArchived must transition to resolve pending while another change is resolving"
        );
        assert!(!state.is_terminal_change("archived"));
    }

    #[test]
    fn test_parallel_mode_change_archived_uses_resolve_pending_when_other_change_is_rejecting() {
        use crate::events::ExecutionEvent;
        use crate::vcs::WorkspaceStatus;

        let mut state =
            OrchestratorState::new(vec!["rejecting".to_string(), "archived".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
            change_id: "rejecting".to_string(),
            workspace_name: "ws-rejecting".to_string(),
            status: WorkspaceStatus::Rejecting,
        });
        assert_eq!(state.display_status("rejecting"), "rejecting");

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("archived".to_string()));
        assert_eq!(
            state.display_status("archived"),
            "resolve pending",
            "Parallel: ChangeArchived must transition to resolve pending while another change is rejecting"
        );
        assert!(!state.is_terminal_change("archived"));
    }

    #[test]
    fn test_parallel_mode_change_archived_enters_resolving_when_other_change_is_accepting() {
        use crate::events::ExecutionEvent;

        let mut state =
            OrchestratorState::new(vec!["accepting".to_string(), "archived".to_string()], 0);

        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "accepting".to_string(),
            command: "accept accepting".to_string(),
        });
        assert_eq!(state.display_status("accepting"), "accepting");

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("archived".to_string()));
        assert_eq!(
            state.display_status("archived"),
            "resolving",
            "Parallel: accepting activity is not a merge/resolve lane blocker, so no-blocker archive handling must enter resolving instead of merge wait or resolve pending"
        );
    }

    #[test]
    fn test_parallel_mode_full_lifecycle() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);

        // Queue and process 'a'
        state.apply_command(ReducerCommand::AddToQueue("a".to_string()));
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "a".to_string(),
            command: "cmd".to_string(),
        });
        assert_eq!(state.display_status("a"), "applying");

        state.apply_execution_event(&ExecutionEvent::ChangeArchived("a".to_string()));
        assert_eq!(
            state.display_status("a"),
            "resolving",
            "Parallel: no-blocker archive completion must enter active resolving before merged"
        );
        assert!(!state.is_terminal_change("a"));

        // Merge 'a'
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "a".to_string(),
            revision: "rev-a".to_string(),
        });
        assert_eq!(state.display_status("a"), "merged");
        assert!(state.is_terminal_change("a"));

        // 'b' is still not queued
        assert_eq!(state.display_status("b"), "not queued");
        assert!(!state.is_terminal_change("b"));
    }

    #[test]
    fn test_parallel_mode_merge_deferred_then_completed() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Archive completion without a concrete blocker enters active merge handling.
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));
        assert_eq!(
            state.display_status("c"),
            "resolving",
            "Parallel: ChangeArchived alone must not imply merge wait before manual blocker evidence exists"
        );

        // Merge deferred (manual, not auto-resumable)
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        // Manual deferral: stays in merge wait.
        assert_eq!(state.display_status("c"), "merge wait");

        // Eventually merge succeeds
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");
        assert!(state.is_terminal_change("c"));
    }

    #[test]
    fn test_parallel_mode_late_events_do_not_regress_merged() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Full lifecycle to merged
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");

        // Late events must not regress
        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "c".to_string(),
            error: "late".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");

        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "cmd".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");
    }

    // -----------------------------------------------------------------------
    // Regression: auto-resumable MergeDeferred must not stay in MergeWait
    // -----------------------------------------------------------------------

    /// `MergeDeferred(auto_resumable=true)` must set ResolveWait, not MergeWait.
    /// This prevents the "stuck after prior merge" scenario where a change
    /// appears to need manual M-press even though it can be resolved automatically.
    #[test]
    fn test_auto_resumable_merge_deferred_sets_resolve_wait() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["b".to_string()], 0);

        // change-b gets a MergeDeferred that is auto-resumable (dirty base caused by
        // change-a's merge being in progress).
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "b".to_string(),
            reason: "Merge in progress (MERGE_HEAD exists)".to_string(),
            auto_resumable: true,
        });

        // Must NOT land in merge wait (which would require manual M press).
        assert_eq!(
            state.display_status("b"),
            "resolve pending",
            "auto-resumable deferred change must enter ResolveWait, not MergeWait"
        );
    }

    /// Workspace refresh after auto-resumable deferral must not regress to MergeWait.
    #[test]
    fn test_auto_resumable_deferred_survives_workspace_refresh() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["b".to_string()], 0);

        // Auto-resumable deferral → ResolveWait.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "b".to_string(),
            reason: "Working tree has uncommitted changes".to_string(),
            auto_resumable: true,
        });
        assert_eq!(state.display_status("b"), "resolve pending");

        // Subsequent ChangesRefreshed sees the workspace as archived (still waiting).
        // It must NOT regress the auto-resolve intent back to merge wait.
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: Default::default(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: Default::default(),
            worktree_change_ids: Default::default(),
            worktree_paths: Default::default(),
            worktree_not_ahead_ids: Default::default(),
            merge_wait_ids: ["b".to_string()].into_iter().collect(),
        });

        assert_eq!(
            state.display_status("b"),
            "resolve pending",
            "workspace refresh must not regress auto-resumable deferred change to merge wait"
        );
    }

    /// After auto-resumable deferral, MergeCompleted (from retry) drives change to Merged.
    #[test]
    fn test_auto_resumable_deferred_then_merge_completed() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["b".to_string()], 0);

        // Auto-resumable deferral.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "b".to_string(),
            reason: "Merge in progress (MERGE_HEAD exists)".to_string(),
            auto_resumable: true,
        });
        assert_eq!(state.display_status("b"), "resolve pending");

        // Scheduler retries and merge succeeds.
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "b".to_string(),
            revision: "abc123".to_string(),
        });
        assert_eq!(state.display_status("b"), "merged");
        assert!(state.is_terminal_change("b"));
    }

    // -----------------------------------------------------------------------
    // Fix: manual resolve dirty-base classification (fix-resolve-dirty-base-transition)
    // -----------------------------------------------------------------------

    /// Manual resolve blocked by another merge in progress → MergeDeferred(auto_resumable=true)
    /// → ResolveWait ("resolve pending"), eligible for automatic retry.
    #[test]
    fn test_manual_resolve_blocked_by_merge_in_progress_becomes_resolve_pending() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Change starts in MergeWait.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status("c"), "merge wait");

        // User requests resolve → ResolveWait.
        state.apply_command(ReducerCommand::ResolveMerge("c".to_string()));
        assert_eq!(state.display_status("c"), "resolve pending");

        // Dirty-base check discovers MERGE_HEAD → auto-resumable MergeDeferred.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "Merge in progress (MERGE_HEAD exists)".to_string(),
            auto_resumable: true,
        });
        assert_eq!(
            state.display_status("c"),
            "resolve pending",
            "auto-resumable dirty base must stay as resolve pending"
        );
    }

    /// Manual resolve blocked by uncommitted changes → ResolveFailed → MergeWait.
    #[test]
    fn test_manual_resolve_blocked_by_uncommitted_changes_stays_merge_wait() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Change starts in MergeWait.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "c".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status("c"), "merge wait");

        // User requests resolve → ResolveWait.
        state.apply_command(ReducerCommand::ResolveMerge("c".to_string()));
        assert_eq!(state.display_status("c"), "resolve pending");

        // Dirty-base check discovers uncommitted changes → ResolveFailed.
        // Note: No ResolveStarted was sent (early exit).
        state.apply_execution_event(&ExecutionEvent::ResolveFailed {
            change_id: "c".to_string(),
            error: "Base is dirty: Working tree has uncommitted changes".to_string(),
        });
        assert_eq!(
            state.display_status("c"),
            "merge wait",
            "uncommitted-changes dirty base must revert to merge wait"
        );
    }

    /// After auto-resumable deferral from manual resolve, a preceding resolve
    /// completes (ResolveCompleted for another change) → the deferred change is
    /// retried via retry_deferred_merges → MergeCompleted → Merged.
    #[test]
    fn test_auto_resumable_deferred_resolve_auto_retries_after_preceding_completes() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["a".to_string(), "b".to_string()], 0);

        // change-a is being resolved actively.
        state.apply_execution_event(&ExecutionEvent::ResolveStarted {
            change_id: "a".to_string(),
            command: "resolve a".to_string(),
        });
        assert_eq!(state.display_status("a"), "resolving");

        // change-b tries manual resolve → blocked by MERGE_HEAD → auto-resumable.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "b".to_string(),
            reason: "Merge in progress (MERGE_HEAD exists)".to_string(),
            auto_resumable: true,
        });
        assert_eq!(state.display_status("b"), "resolve pending");

        // change-a resolve completes.
        state.apply_execution_event(&ExecutionEvent::ResolveCompleted {
            change_id: "a".to_string(),
            worktree_change_ids: None,
        });
        assert_eq!(state.display_status("a"), "merged");

        // retry_deferred_merges retries change-b and succeeds → MergeCompleted.
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "b".to_string(),
            revision: "rev-b".to_string(),
        });
        assert_eq!(state.display_status("b"), "merged");
        assert!(state.is_terminal_change("b"));
    }

    // -----------------------------------------------------------------------
    // Fix: parallel TUI queued/blocked state regression – reducer unit tests
    // -----------------------------------------------------------------------

    /// Explicit RetryError on a change in Error terminal state must clear the terminal and
    /// set queue_intent = Queued so that the TUI retry path works correctly.
    #[test]
    fn test_explicit_retry_retries_error_terminal() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Drive the change to an error terminal state.
        state.apply_execution_event(&ExecutionEvent::ProcessingError {
            id: "c".to_string(),
            error: "apply failed".to_string(),
        });
        assert_eq!(state.display_status("c"), "error");
        assert!(state.is_terminal_change("c"));

        // Ordinary AddToQueue must not clear the error terminal.
        let ordinary_outcome = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(ordinary_outcome, ReduceOutcome::NoOp));
        assert_eq!(state.display_status("c"), "error");

        // Explicit RetryError must clear the error terminal.
        let outcome = state.apply_command(ReducerCommand::RetryError("c".to_string()));
        assert!(
            matches!(outcome, ReduceOutcome::Changed(_)),
            "RetryError on error change must be Changed, not NoOp"
        );
        assert_eq!(
            state.display_status("c"),
            "queued",
            "after retry, change must display as queued"
        );
        assert!(
            !state.is_terminal_change("c"),
            "error terminal must be cleared by AddToQueue"
        );
    }

    /// DequeueChange should clear runtime state to a non-terminal not-queued idle state.
    #[test]
    fn test_dequeue_change_resets_to_not_queued_idle() {
        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);

        // Start in queued/active-ish flow, then dequeue.
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        let outcome = state.apply_command(ReducerCommand::DequeueChange("c".to_string()));
        assert!(matches!(outcome, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "not queued");
        assert!(!state.is_terminal_change("c"));

        // Explicit re-queue remains possible only by user action.
        let outcome2 = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(matches!(outcome2, ReduceOutcome::Changed(_)));
        assert_eq!(state.display_status("c"), "queued");
    }

    /// AddToQueue on a change that reached its post-archive terminal state must
    /// be a no-op: archive alone is not terminal, the merge that follows it is.
    #[test]
    fn test_add_to_queue_noop_on_merged() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("c".to_string()));
        assert_eq!(
            state.display_status("c"),
            "resolving",
            "archive enters post-archive handling instead of terminating"
        );

        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "c".to_string(),
            revision: "rev".to_string(),
        });
        assert_eq!(state.display_status("c"), "merged");

        let outcome = state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert!(
            matches!(outcome, ReduceOutcome::NoOp),
            "AddToQueue on a merged change must be NoOp"
        );
        assert_eq!(state.display_status("c"), "merged");
    }

    /// After AddToQueue + DependencyBlocked + DependencyResolved, the display must
    /// return to "queued" (queue_intent is preserved through the block/resolve cycle).
    #[test]
    fn test_dependency_resolved_restores_queued_after_block() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert_eq!(state.display_status("c"), "queued");

        state.apply_execution_event(&ExecutionEvent::DependencyBlocked {
            change_id: "c".to_string(),
            dependency_ids: vec!["dep".to_string()],
        });
        assert_eq!(state.display_status("c"), "blocked");

        state.apply_execution_event(&ExecutionEvent::DependencyResolved {
            change_id: "c".to_string(),
        });
        assert_eq!(
            state.display_status("c"),
            "queued",
            "DependencyResolved must restore queued (not not-queued)"
        );
    }

    #[test]
    fn test_dependency_blocked_and_resolved_preserve_queue_intent_until_user_dequeue() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));

        state.apply_execution_event(&ExecutionEvent::DependencyBlocked {
            change_id: "c".to_string(),
            dependency_ids: vec!["dep-a".to_string()],
        });
        assert_eq!(state.display_status("c"), "blocked");

        state.apply_execution_event(&ExecutionEvent::DependencyResolved {
            change_id: "c".to_string(),
        });
        assert_eq!(state.display_status("c"), "queued");

        state.apply_command(ReducerCommand::RemoveFromQueue("c".to_string()));
        assert_eq!(
            state.display_status("c"),
            "not queued",
            "queue intent should only clear on explicit dequeue command"
        );
    }

    /// ChangesRefreshed must not overwrite queue_intent = Queued with "not queued".
    #[test]
    fn test_changes_refreshed_preserves_queue_intent() {
        use crate::events::ExecutionEvent;
        use std::collections::{HashMap, HashSet};

        let mut state = OrchestratorState::new(vec!["c".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("c".to_string()));
        assert_eq!(state.display_status("c"), "queued");

        // Simulate initial parallel ChangesRefreshed with no special observations.
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: HashSet::new(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::new(),
        });

        assert_eq!(
            state.display_status("c"),
            "queued",
            "ChangesRefreshed must not overwrite queue_intent = Queued"
        );
    }

    /// Fast-forward resolve: terminal Merged set via ResolveCompleted must survive
    /// a subsequent ChangesRefreshed that still reports the workspace as archived
    /// (the workspace scan may lag behind the actual merge).
    #[test]
    fn test_fast_forward_merged_survives_archived_observation() {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec!["ff".to_string()], 0);

        // Change is archived and deferred to merge wait (manual, not auto-resumable).
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "ff".to_string(),
            reason: "base dirty".to_string(),
            auto_resumable: false,
        });
        assert_eq!(state.display_status("ff"), "merge wait");

        // Resolve starts.
        state.apply_command(ReducerCommand::ResolveMerge("ff".to_string()));
        assert_eq!(state.display_status("ff"), "resolve pending");
        state.apply_execution_event(&ExecutionEvent::ResolveStarted {
            change_id: "ff".to_string(),
            command: "resolve-cmd".to_string(),
        });
        assert_eq!(state.display_status("ff"), "resolving");

        // Resolve succeeds (fast-forward merge detected by the caller).
        state.apply_execution_event(&ExecutionEvent::ResolveCompleted {
            change_id: "ff".to_string(),
            worktree_change_ids: None,
        });
        assert_eq!(state.display_status("ff"), "merged");

        // Refresh still sees the worktree as archived (workspace scan lag).
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![],
            committed_change_ids: Default::default(),
            rejected_changes: Vec::new(),
            uncommitted_file_change_ids: Default::default(),
            worktree_change_ids: Default::default(),
            worktree_paths: Default::default(),
            worktree_not_ahead_ids: Default::default(),
            merge_wait_ids: ["ff".to_string()].into_iter().collect(),
        });
        assert_eq!(
            state.display_status("ff"),
            "merged",
            "Terminal Merged from fast-forward resolve must not regress to merge wait"
        );
    }

    // ── Opted-in per-change upstream publication ────────────────────────────

    /// Set a parallel change up to the point where its archived result has just
    /// been integrated into cumulative base.
    fn per_change_upstream_state(change_id: &str) -> OrchestratorState {
        use crate::events::ExecutionEvent;

        let mut state = OrchestratorState::new(vec![change_id.to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::ChangeArchived(change_id.to_string()));
        state
    }

    fn per_change_upstream_push_started(change_id: &str) -> crate::events::ExecutionEvent {
        crate::events::ExecutionEvent::PushStarted {
            change_id: change_id.to_string(),
            remote: "origin".to_string(),
            branch: "main".to_string(),
        }
    }

    fn per_change_upstream_push_completed(change_id: &str) -> crate::events::ExecutionEvent {
        crate::events::ExecutionEvent::PushCompleted {
            change_id: change_id.to_string(),
            remote: "origin".to_string(),
            branch: "main".to_string(),
        }
    }

    fn per_change_upstream_push_failed(change_id: &str) -> crate::events::ExecutionEvent {
        crate::events::ExecutionEvent::PushFailed {
            change_id: change_id.to_string(),
            remote: "origin".to_string(),
            branch: "main".to_string(),
            error: "upstream publication incomplete: verification failed".to_string(),
        }
    }

    #[test]
    fn per_change_upstream_disabled_cumulative_merge_stays_merged() {
        use crate::events::ExecutionEvent;

        // Without the option the execution path still emits MergeCompleted, and
        // that must keep its existing terminal meaning.
        let mut state = per_change_upstream_state("alpha");
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "alpha".to_string(),
            revision: "merge-rev".to_string(),
        });

        assert_eq!(state.display_status("alpha"), "merged");
        assert!(state.is_terminal_change("alpha"));
    }

    #[test]
    fn per_change_upstream_local_integration_is_not_terminal_merged() {
        // With the option enabled the execution path suppresses MergeCompleted
        // and reports publication progress instead, so nothing can observe a
        // final `merged` while publication is still owed.
        let mut state = per_change_upstream_state("alpha");
        state.apply_execution_event(&per_change_upstream_push_started("alpha"));

        assert_ne!(state.display_status("alpha"), "merged");
        assert!(
            !state.is_terminal_change("alpha"),
            "publication progress must remain non-terminal"
        );
        assert!(
            !state.queued_change_ids().contains(&"alpha".to_string()),
            "a publishing change must not be ordinary queued apply work"
        );
    }

    /// Replay the events a *fresh* opted-in cumulative integration actually
    /// emits, in order, up to the point where publication takes over.
    ///
    /// The git-backed merge path emits `MergeStarted` and `ResolveStarted`, then
    /// the conflict resolver reports `ConflictResolutionStarted` /
    /// `ConflictResolutionCompleted`. It must *not* emit the per-change
    /// `ResolveCompleted`, because that finalizes the reducer as terminal
    /// `merged` before anything has been published.
    fn per_change_upstream_apply_fresh_integration(state: &mut OrchestratorState, change_id: &str) {
        use crate::events::ExecutionEvent;

        state.apply_execution_event(&ExecutionEvent::MergeStarted {
            revisions: vec!["ws-alpha".to_string()],
        });
        state.apply_execution_event(&ExecutionEvent::ResolveStarted {
            change_id: change_id.to_string(),
            command: "merge archived change into base branch (1 revision(s))".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ConflictResolutionStarted);
        state.apply_execution_event(&ExecutionEvent::ConflictResolutionCompleted);
    }

    #[test]
    fn per_change_upstream_fresh_integration_never_displays_merged() {
        let mut state = per_change_upstream_state("alpha");
        per_change_upstream_apply_fresh_integration(&mut state, "alpha");

        assert_ne!(
            state.display_status("alpha"),
            "merged",
            "local integration must not finalize an opted-in change"
        );
        assert!(
            !state.is_terminal_change("alpha"),
            "publication is still owed, so nothing is terminal yet"
        );

        state.apply_execution_event(&per_change_upstream_push_started("alpha"));
        assert_ne!(state.display_status("alpha"), "merged");
        assert!(!state.is_terminal_change("alpha"));

        state.apply_execution_event(&per_change_upstream_push_completed("alpha"));
        assert_eq!(
            state.display_status("alpha"),
            "pushed",
            "remote confirmation is the opted-in terminal state"
        );
        assert!(state.is_terminal_change("alpha"));
    }

    #[test]
    fn per_change_upstream_fresh_integration_failure_stays_recoverable() {
        // Same fresh sequence, but publication fails. A terminal `merged` here
        // would swallow PushFailed and leave no recoverable Error for F5 retry.
        let mut state = per_change_upstream_state("alpha");
        per_change_upstream_apply_fresh_integration(&mut state, "alpha");
        state.apply_execution_event(&per_change_upstream_push_started("alpha"));
        state.apply_execution_event(&per_change_upstream_push_failed("alpha"));

        assert_eq!(
            state.display_status("alpha"),
            "error",
            "a failed publication after a fresh integration must surface as recoverable error"
        );
        assert!(
            state.is_terminal_error_change("alpha"),
            "explicit F5 retry needs a recoverable error to act on"
        );
        assert!(matches!(
            state.apply_command(ReducerCommand::RetryError("alpha".to_string())),
            ReduceOutcome::Changed(_)
        ));

        // Retry succeeds: the recoverable error is superseded by confirmation.
        state.apply_execution_event(&per_change_upstream_push_started("alpha"));
        state.apply_execution_event(&per_change_upstream_push_completed("alpha"));
        assert_eq!(state.display_status("alpha"), "pushed");
    }

    #[test]
    fn per_change_upstream_remote_confirmation_becomes_pushed_terminal() {
        let mut state = per_change_upstream_state("alpha");
        state.apply_execution_event(&per_change_upstream_push_started("alpha"));
        state.apply_execution_event(&per_change_upstream_push_completed("alpha"));

        assert_eq!(state.display_status("alpha"), "pushed");
        assert!(state.is_terminal_change("alpha"));
        assert_ne!(state.display_status("alpha"), "merged");
    }

    #[test]
    fn per_change_upstream_publication_failure_is_recoverable_not_merged() {
        let mut state = per_change_upstream_state("alpha");
        state.apply_execution_event(&per_change_upstream_push_started("alpha"));
        state.apply_execution_event(&per_change_upstream_push_failed("alpha"));

        assert_eq!(
            state.display_status("alpha"),
            "error",
            "a failed publication projects into the existing recoverable error flow"
        );
        assert!(
            state.is_terminal_error_change("alpha"),
            "the recoverable error must gate ordinary apply dispatch"
        );

        // Explicit retry (F5 or the equivalent local web control) is accepted.
        let outcome = state.apply_command(ReducerCommand::RetryError("alpha".to_string()));
        assert!(matches!(outcome, ReduceOutcome::Changed(_)));
    }

    #[test]
    fn per_change_upstream_late_confirmation_supersedes_publication_failure() {
        let mut state = per_change_upstream_state("alpha");
        state.apply_execution_event(&per_change_upstream_push_started("alpha"));
        state.apply_execution_event(&per_change_upstream_push_failed("alpha"));
        assert_eq!(state.display_status("alpha"), "error");

        state.apply_execution_event(&per_change_upstream_push_completed("alpha"));

        assert_eq!(state.display_status("alpha"), "pushed");
        assert!(
            !state.queued_change_ids().contains(&"alpha".to_string()),
            "confirmed publication must not leave ordinary apply dispatch behind"
        );
    }

    #[test]
    fn per_change_upstream_pushed_terminal_is_not_retryable() {
        let mut state = per_change_upstream_state("alpha");
        state.apply_execution_event(&per_change_upstream_push_started("alpha"));
        state.apply_execution_event(&per_change_upstream_push_completed("alpha"));

        assert!(matches!(
            state.apply_command(ReducerCommand::RetryError("alpha".to_string())),
            ReduceOutcome::NoOp
        ));
        assert_eq!(state.display_status("alpha"), "pushed");

        // Nor may it be reintroduced as merge/resolve work.
        assert!(matches!(
            state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string())),
            ReduceOutcome::NoOp
        ));
        assert_eq!(state.display_status("alpha"), "pushed");
    }

    #[test]
    fn per_change_upstream_confirmation_clears_base_lane_retry_intent() {
        use crate::events::ExecutionEvent;

        let mut state = per_change_upstream_state("alpha");
        // A publication that had to wait behind the lane records resolve-wait
        // retry intent; confirmation must clear it rather than leave it as
        // ordinary merge work. The first deferral makes the change idle, which
        // is the state an auto-resumable lane wait is recorded from.
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "base lane busy".to_string(),
            auto_resumable: false,
        });
        state.apply_execution_event(&ExecutionEvent::MergeDeferred {
            change_id: "alpha".to_string(),
            reason: "waiting for publication of beta".to_string(),
            auto_resumable: true,
        });
        assert!(state
            .resolve_wait_change_ids()
            .contains(&"alpha".to_string()));

        state.apply_execution_event(&per_change_upstream_push_completed("alpha"));

        assert_eq!(state.display_status("alpha"), "pushed");
        assert!(
            !state
                .resolve_wait_change_ids()
                .contains(&"alpha".to_string()),
            "confirmed publication must clear base-lane retry intent"
        );
    }

    // ------------------------------------------------------------------
    // Explicit-intent boundary
    // ------------------------------------------------------------------

    /// Catalog refresh is registration, not intent: an all-change refresh may
    /// make an unselected change visible but must leave every execution lane
    /// clear for it.
    #[test]
    fn changes_refreshed_registers_without_creating_queue_or_lane_eligibility() {
        use crate::events::ExecutionEvent;
        use std::collections::{HashMap, HashSet};

        let mut state = OrchestratorState::new(vec!["fresh".to_string()], 1);
        state.apply_command(ReducerCommand::AddToQueue("fresh".to_string()));

        let change = |id: &str| crate::openspec::Change {
            id: id.to_string(),
            completed_tasks: 0,
            total_tasks: 1,
            last_modified: "now".to_string(),
            dependencies: Vec::new(),
            metadata: crate::openspec::ProposalMetadata::default(),
        };

        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![change("fresh"), change("stale")],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::from(["fresh".to_string(), "stale".to_string()]),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["stale".to_string()]),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::new(),
        });

        assert!(
            state.is_in_snapshot("stale"),
            "refresh may register a newly observed change"
        );
        assert_eq!(state.display_status("stale"), "not queued");
        assert_eq!(
            state
                .change_runtime("stale")
                .expect("refresh registers runtime state")
                .queue_intent,
            QueueIntent::NotQueued
        );
        assert_eq!(state.queued_change_ids(), vec!["fresh".to_string()]);
        assert!(state.merge_wait_change_ids().is_empty());
        assert!(state.resolve_wait_change_ids().is_empty());
        assert!(state.reject_wait_change_ids().is_empty());
        assert!(state.active_change_ids().is_empty());
        assert!(
            !state.is_ordinary_queue_eligible("stale"),
            "a registered but unqueued change must not be dispatchable"
        );
    }

    /// Revocation is immediate, and a later refresh cannot undo it.
    #[test]
    fn removal_and_dequeue_revoke_ordinary_eligibility_until_explicit_requeue() {
        use crate::events::ExecutionEvent;
        use std::collections::{HashMap, HashSet};

        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 1);
        state.apply_command(ReducerCommand::AddToQueue("alpha".to_string()));
        assert!(state.is_ordinary_queue_eligible("alpha"));

        state.apply_command(ReducerCommand::RemoveFromQueue("alpha".to_string()));
        assert!(!state.is_ordinary_queue_eligible("alpha"));
        assert!(state.queued_change_ids().is_empty());

        // A refresh that re-observes the change and its worktree must not
        // resurrect eligibility.
        state.apply_execution_event(&ExecutionEvent::ChangesRefreshed {
            changes: vec![crate::openspec::Change {
                id: "alpha".to_string(),
                completed_tasks: 0,
                total_tasks: 1,
                last_modified: "now".to_string(),
                dependencies: Vec::new(),
                metadata: crate::openspec::ProposalMetadata::default(),
            }],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::from(["alpha".to_string()]),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["alpha".to_string()]),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::new(),
        });
        assert!(!state.is_ordinary_queue_eligible("alpha"));

        state.apply_command(ReducerCommand::AddToQueue("alpha".to_string()));
        state.apply_command(ReducerCommand::DequeueChange("alpha".to_string()));
        assert!(
            !state.is_ordinary_queue_eligible("alpha"),
            "stop-and-dequeue revokes ordinary eligibility"
        );

        state.apply_command(ReducerCommand::AddToQueue("alpha".to_string()));
        assert!(
            state.is_ordinary_queue_eligible("alpha"),
            "explicit requeue restores eligibility"
        );
        assert_eq!(state.queued_change_ids(), vec!["alpha".to_string()]);
    }

    // -----------------------------------------------------------------------
    // Ephemeral workspace preparation
    // -----------------------------------------------------------------------

    /// A queued change admitted to a slot, ready to receive preparation events.
    fn queued_state(change_id: &str) -> OrchestratorState {
        let mut state = OrchestratorState::new(vec![change_id.to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue(change_id.to_string()));
        assert_eq!(state.display_status(change_id), "queued");
        state
    }

    fn prepare(state: &mut OrchestratorState, change_id: &str) {
        state.apply_execution_event(
            &crate::events::ExecutionEvent::WorkspacePreparationStarted {
                change_id: change_id.to_string(),
            },
        );
    }

    fn end_preparation(state: &mut OrchestratorState, change_id: &str) {
        state.apply_execution_event(&crate::events::ExecutionEvent::WorkspacePreparationEnded {
            change_id: change_id.to_string(),
        });
    }

    #[test]
    fn preparing_replaces_queued_and_then_yields_to_the_repository_derived_phase() {
        use crate::events::ExecutionEvent;

        // Apply route: queued -> preparing -> applying.
        let mut state = queued_state("c");
        prepare(&mut state, "c");
        assert_eq!(state.display_status("c"), "preparing");
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "apply".to_string(),
        });
        assert_eq!(state.display_status("c"), "applying");

        // A resumed workspace routes straight to acceptance, and preparation
        // must not have manufactured an Apply transition on the way.
        let mut state = queued_state("c");
        prepare(&mut state, "c");
        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "c".to_string(),
            command: "accept".to_string(),
        });
        assert_eq!(state.display_status("c"), "accepting");
        assert_eq!(state.apply_count("c"), 0);

        // Archive resume.
        let mut state = queued_state("c");
        prepare(&mut state, "c");
        state.apply_execution_event(&ExecutionEvent::ArchiveStarted {
            change_id: "c".to_string(),
            command: "archive".to_string(),
        });
        assert_eq!(state.display_status("c"), "archiving");
    }

    #[test]
    fn preparing_failure_becomes_error_with_the_preparation_diagnostic() {
        use crate::events::ExecutionEvent;

        let mut state = queued_state("c");
        prepare(&mut state, "c");
        state.apply_execution_event(&ExecutionEvent::ProcessingError {
            id: "c".to_string(),
            error: "Worktree setup failed after 1.0s: .wt/setup exited with code 3".to_string(),
        });

        assert_eq!(state.display_status("c"), "error");
        let rt = state.change_runtime("c").expect("runtime entry");
        assert!(matches!(rt.activity, ActivityState::Idle));
        assert!(rt
            .error_message()
            .is_some_and(|message| message.contains(".wt/setup")));
    }

    #[test]
    fn preparing_is_cleared_by_a_pre_operation_exit() {
        let mut state = queued_state("c");
        prepare(&mut state, "c");
        assert_eq!(state.display_status("c"), "preparing");

        // Global cancellation or a terminal resume route ends dispatch without
        // any operation-started event.
        end_preparation(&mut state, "c");
        assert_eq!(
            state.display_status("c"),
            "queued",
            "preparation must not outlive the dispatch that announced it"
        );
    }

    #[test]
    fn preparing_leaves_through_a_stop_before_an_operation_agent_starts() {
        use crate::events::ExecutionEvent;

        let mut state = queued_state("c");
        prepare(&mut state, "c");
        state.apply_execution_event(&ExecutionEvent::ChangeDequeued {
            change_id: "c".to_string(),
        });

        assert_eq!(state.display_status("c"), "not queued");
        let rt = state.change_runtime("c").expect("runtime entry");
        assert!(matches!(rt.activity, ActivityState::Idle));

        // A clear that arrives after the stop cannot resurrect anything.
        end_preparation(&mut state, "c");
        assert_eq!(state.display_status("c"), "not queued");
    }

    #[test]
    fn preparing_never_overwrites_a_running_operation_or_a_terminal_change() {
        use crate::events::ExecutionEvent;

        // A late preparation event must not relabel a running Apply as if no
        // agent had started.
        let mut state = queued_state("c");
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "apply".to_string(),
        });
        prepare(&mut state, "c");
        assert_eq!(state.display_status("c"), "applying");

        // Nor may it reactivate a terminal change.
        let mut state = queued_state("c");
        state.apply_execution_event(&ExecutionEvent::ProcessingError {
            id: "c".to_string(),
            error: "boom".to_string(),
        });
        prepare(&mut state, "c");
        assert_eq!(state.display_status("c"), "error");

        // Nor a change an operator force-stopped.
        let mut state = queued_state("c");
        state.apply_command(ReducerCommand::DequeueChange("c".to_string()));
        prepare(&mut state, "c");
        assert_eq!(state.display_status("c"), "not queued");
    }

    #[test]
    fn preparing_clearing_is_narrow_and_idempotent() {
        use crate::events::ExecutionEvent;

        // Clearing a change that is applying leaves the operation alone: the
        // completion funnel emits it for every task, including ones that did
        // reach a real phase.
        let mut state = queued_state("c");
        state.apply_execution_event(&ExecutionEvent::ApplyStarted {
            change_id: "c".to_string(),
            command: "apply".to_string(),
        });
        end_preparation(&mut state, "c");
        assert_eq!(state.display_status("c"), "applying");

        // Repeating the clear on an already-cleared change changes nothing.
        let mut state = queued_state("c");
        prepare(&mut state, "c");
        end_preparation(&mut state, "c");
        end_preparation(&mut state, "c");
        assert_eq!(state.display_status("c"), "queued");
    }

    #[test]
    fn preparing_counts_as_active_execution_but_not_as_a_running_agent() {
        let mut state = queued_state("c");
        prepare(&mut state, "c");

        assert!(
            state.is_active_change("c"),
            "an admitted change mutating its worktree is active execution"
        );
        assert!(
            !state.is_agent_execution_active(),
            "preparation must never justify a force-stopped-process claim"
        );
        assert!(
            crate::orchestration::operator_command::is_active_status(state.display_status("c")),
            "every operator surface must treat preparing as active"
        );
    }

    #[test]
    fn preparing_is_not_durable_routing_state() {
        let mut state = queued_state("c");
        prepare(&mut state, "c");
        assert_eq!(state.display_status("c"), "preparing");

        // A restart rebuilds the reducer from the same workspace contents. The
        // ephemeral preparation observation is gone, and nothing about the next
        // action was derived from it.
        let restarted = OrchestratorState::new(vec!["c".to_string()], 0);
        assert_eq!(restarted.display_status("c"), "not queued");
        assert!(!restarted.is_active_change("c"));
    }

    // -----------------------------------------------------------------------
    // Process-level stop: reducer-owned run-boundary reconciliation
    // -----------------------------------------------------------------------

    /// Drive one change into a named reducer state through ordinary events.
    fn enter_runtime_family(state: &mut OrchestratorState, change_id: &str, family: &str) {
        use crate::events::ExecutionEvent;
        use crate::vcs::WorkspaceStatus;

        match family {
            "queued" => {
                state.apply_command(ReducerCommand::AddToQueue(change_id.to_string()));
            }
            "preparing" => {
                state.apply_command(ReducerCommand::AddToQueue(change_id.to_string()));
                state.apply_execution_event(&ExecutionEvent::WorkspacePreparationStarted {
                    change_id: change_id.to_string(),
                });
            }
            "applying" => {
                state.apply_execution_event(&ExecutionEvent::ApplyStarted {
                    change_id: change_id.to_string(),
                    command: "apply".to_string(),
                });
            }
            "accepting" => {
                state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
                    change_id: change_id.to_string(),
                    command: "accept".to_string(),
                });
            }
            "archiving" => {
                state.apply_execution_event(&ExecutionEvent::ArchiveStarted {
                    change_id: change_id.to_string(),
                    command: "archive".to_string(),
                });
            }
            "resolving" => {
                state.apply_execution_event(&ExecutionEvent::ResolveStarted {
                    change_id: change_id.to_string(),
                    command: "resolve".to_string(),
                });
            }
            "rejecting" => {
                state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
                    change_id: change_id.to_string(),
                    workspace_name: format!("ws-{change_id}"),
                    status: WorkspaceStatus::Rejecting,
                });
            }
            "merge wait" => {
                state.apply_execution_event(&ExecutionEvent::MergeDeferred {
                    change_id: change_id.to_string(),
                    reason: "base branch is dirty".to_string(),
                    auto_resumable: false,
                });
            }
            "resolve pending" => {
                state.apply_execution_event(&ExecutionEvent::MergeDeferred {
                    change_id: change_id.to_string(),
                    reason: "another merge is in progress".to_string(),
                    auto_resumable: true,
                });
            }
            "reject pending" => {
                // Rejection review is a base-mutating lane: an occupied lane is
                // what produces the queued `reject pending` wait.
                state.apply_execution_event(&ExecutionEvent::ResolveStarted {
                    change_id: "lane-occupant".to_string(),
                    command: "resolve".to_string(),
                });
                state.apply_execution_event(&ExecutionEvent::WorkspaceStatusUpdated {
                    change_id: change_id.to_string(),
                    workspace_name: format!("ws-{change_id}"),
                    status: WorkspaceStatus::Rejecting,
                });
            }
            "blocked (dependency)" => {
                state.apply_execution_event(&ExecutionEvent::DependencyBlocked {
                    change_id: change_id.to_string(),
                    dependency_ids: vec!["dep".to_string()],
                });
            }
            "blocked (external)" => {
                state.apply_execution_event(&ExecutionEvent::AcceptanceGated {
                    change_id: change_id.to_string(),
                    blocker: crate::events::StalledBlocker::acceptance_external(
                        "credential",
                        "STAGING_API_KEY is unset",
                    ),
                });
            }
            "stalled" => {
                state.mark_stalled(change_id.to_string());
                state.apply_execution_event(&ExecutionEvent::ExecutionBlocked {
                    change_id: change_id.to_string(),
                    blocker: crate::events::StalledBlocker {
                        category: "no_progress".to_string(),
                        phase: "apply".to_string(),
                        gate: "apply".to_string(),
                        error_summary: "no semantic progress".to_string(),
                        evidence: vec!["tasks.md unchanged".to_string()],
                        unblock_condition: None,
                        prerequisite_owner: None,
                        next_action: "operator review".to_string(),
                        resumable: true,
                        worktree_preserved: true,
                    },
                });
            }
            other => panic!("unknown runtime family: {other}"),
        }
    }

    /// A terminal process stop returns every interrupted row to resumable
    /// `not queued`, and touches nothing it did not own.
    ///
    /// The regression: `ExecutionEvent::Stopped` was a reducer no-op, so a row
    /// the operator had just stopped stayed `accepting` in the one state both
    /// frontends read back.
    #[test]
    fn global_stopped_reconciles_interrupted_runtime() {
        use crate::events::ExecutionEvent;

        // ── Every interrupted activity and wait family returns to not queued ──
        let families = [
            "queued",
            "preparing",
            "applying",
            "accepting",
            "archiving",
            "resolving",
            "rejecting",
            "merge wait",
            "resolve pending",
            "reject pending",
            "blocked (dependency)",
            "blocked (external)",
            "stalled",
        ];

        for family in families {
            let mut state =
                OrchestratorState::new(vec!["target".to_string(), "lane-occupant".to_string()], 0);
            enter_runtime_family(&mut state, "target", family);
            assert_ne!(
                state.display_status("target"),
                "not queued",
                "{family} setup did not produce interrupted runtime state"
            );

            state.apply_execution_event(&ExecutionEvent::Stopped);

            assert_eq!(
                state.display_status("target"),
                "not queued",
                "{family} was not reconciled by the run-boundary stop"
            );
            let rt = state.change_runtime.get("target").expect("runtime entry");
            assert_eq!(rt.activity, ActivityState::Idle, "{family} stayed active");
            assert_eq!(rt.wait_state, WaitState::None, "{family} stayed waiting");
            assert_eq!(
                rt.queue_intent,
                QueueIntent::NotQueued,
                "{family} kept queue intent"
            );
            assert_eq!(
                rt.terminal,
                TerminalState::None,
                "{family} was given a terminal outcome by a process stop"
            );
            assert_eq!(rt.blocked_metadata, BlockedMetadata::default());
            assert!(rt.commit_phase_attempt.is_none());
            assert!(rt.dequeued, "{family} did not get the reactivation guard");
            assert!(
                state.resolve_wait_change_ids().is_empty(),
                "{family} left scheduler resolve membership behind"
            );
            assert!(
                state.reject_wait_change_ids().is_empty(),
                "{family} left scheduler reject membership behind"
            );
            assert!(
                !state.stalled_change_ids().contains("target"),
                "{family} left stall membership behind"
            );
            assert!(
                !state.is_ordinary_queue_eligible("target"),
                "{family} remained dispatchable after the run ended"
            );
        }

        // ── Terminal outcomes and fresh idle rows are not the stop's to change ──
        let mut state = OrchestratorState::new(
            vec![
                "err".to_string(),
                "merged".to_string(),
                "pushed".to_string(),
                "rejected".to_string(),
                "fresh".to_string(),
            ],
            0,
        );
        state.apply_execution_event(&ExecutionEvent::ProcessingError {
            id: "err".to_string(),
            error: "apply exited 1".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ChangeArchived("merged".to_string()));
        state.apply_execution_event(&ExecutionEvent::MergeCompleted {
            change_id: "merged".to_string(),
            revision: "rev".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::PushCompleted {
            change_id: "pushed".to_string(),
            remote: "origin".to_string(),
            branch: "main".to_string(),
        });
        state.apply_execution_event(&ExecutionEvent::ChangeRejected {
            change_id: "rejected".to_string(),
            reason: "acceptance confirmed rejection".to_string(),
        });

        state.apply_execution_event(&ExecutionEvent::Stopped);

        assert_eq!(state.display_status("err"), "error");
        assert_eq!(state.display_status("merged"), "merged");
        assert_eq!(state.display_status("pushed"), "pushed");
        assert_eq!(state.display_status("rejected"), "rejected");
        assert_eq!(state.display_status("fresh"), "not queued");
        assert!(
            !state
                .change_runtime
                .get("fresh")
                .expect("fresh runtime entry")
                .dequeued,
            "an unrelated idle row must not be claimed by the stopped run"
        );

        // ── Duplicate stop, late lifecycle events, and same-process refresh ──
        let mut state = OrchestratorState::new(vec!["alpha".to_string()], 0);
        state.apply_execution_event(&ExecutionEvent::AcceptanceStarted {
            change_id: "alpha".to_string(),
            command: "accept".to_string(),
        });
        assert_eq!(state.display_status("alpha"), "accepting");

        state.apply_execution_event(&ExecutionEvent::Stopped);
        let after_first_stop = format!("{:?}", state.change_runtime.get("alpha"));
        state.apply_execution_event(&ExecutionEvent::Stopped);
        assert_eq!(
            format!("{:?}", state.change_runtime.get("alpha")),
            after_first_stop,
            "a duplicate stop must not change reconciled state"
        );
        assert_eq!(state.display_status("alpha"), "not queued");

        for late in [
            ExecutionEvent::AcceptanceStarted {
                change_id: "alpha".to_string(),
                command: "accept".to_string(),
            },
            ExecutionEvent::ArchiveStarted {
                change_id: "alpha".to_string(),
                command: "archive".to_string(),
            },
            ExecutionEvent::ResolveStarted {
                change_id: "alpha".to_string(),
                command: "resolve".to_string(),
            },
            ExecutionEvent::WorkspacePreparationStarted {
                change_id: "alpha".to_string(),
            },
        ] {
            state.apply_execution_event(&late);
            assert_eq!(
                state.display_status("alpha"),
                "not queued",
                "a late event from the stopped run reactivated the row"
            );
        }

        // A same-process refresh observes the archived workspace the stopped run
        // released; the guard is what keeps it from restoring `merge wait`.
        state.apply_execution_event(&stop_refresh_event("alpha"));
        assert_eq!(
            state.display_status("alpha"),
            "not queued",
            "a same-process workspace observation resurrected stopped work"
        );

        // ── Explicit requeue releases the guard ──────────────────────────────
        state.apply_command(ReducerCommand::AddToQueue("alpha".to_string()));
        assert_eq!(state.display_status("alpha"), "queued");
        assert!(
            !state
                .change_runtime
                .get("alpha")
                .expect("runtime entry")
                .dequeued,
            "an explicit requeue must release the reactivation guard"
        );
        assert!(state.is_ordinary_queue_eligible("alpha"));

        // ── The guard is process-local, never restart-routing evidence ───────
        let mut restarted = OrchestratorState::new(vec!["alpha".to_string()], 0);
        restarted.apply_execution_event(&stop_refresh_event("alpha"));
        assert_eq!(
            restarted.display_status("alpha"),
            "merge wait",
            "a restarted process must re-derive routing from workspace evidence alone"
        );
    }

    /// A refresh whose workspace observation would restore `merge wait`.
    fn stop_refresh_event(change_id: &str) -> crate::events::ExecutionEvent {
        crate::events::ExecutionEvent::ChangesRefreshed {
            changes: Vec::new(),
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::from([change_id.to_string()]),
        }
    }

    // =======================================================================
    // Dependency blocker projection
    //
    // Unit scope on purpose: the reducer is in-memory, so these exercise the
    // state contract itself with no Git, filesystem, process, or clock
    // involved. The scheduler-side reconciliation call and the deduplicated
    // diagnostics around it are covered by the parallel-executor regression.
    // =======================================================================

    /// An admitted change with an unresolved dependency is blocked from the
    /// first coherent classification, with the complete typed contract.
    #[test]
    fn dependency_blocker_projection_initial() {
        let mut state = OrchestratorState::new(vec!["beta".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("beta".to_string()));

        assert!(
            state.reconcile_dependency_blocker("beta", &["alpha".to_string()]),
            "the first classification must publish the wait"
        );

        let rt = state.change_runtime("beta").expect("runtime entry");
        assert_eq!(state.display_status("beta"), "blocked");
        assert_eq!(
            rt.queue_intent,
            QueueIntent::Queued,
            "a dependency wait excludes dispatch and never revokes admitted work"
        );
        assert_eq!(rt.wait_state, WaitState::DependencyBlocked);
        assert!(!rt.is_active(), "no execution episode has started");

        let view = state.blocker_view("beta").expect("structured blocker");
        assert_eq!(view.status, "blocked");
        assert_eq!(view.kind, BlockerKind::Dependency);
        assert_eq!(view.dependencies, vec!["alpha".to_string()]);
        assert_eq!(view.category.as_deref(), Some("dependency_blocked"));
        assert!(view
            .detail
            .as_deref()
            .is_some_and(|detail| detail.contains("alpha")));

        // The typed execution state is the retained admission, not a new
        // episode: the change is queued behind evidence, not running.
        assert_eq!(
            crate::orchestration::execution_facts::project_execution_state(
                rt,
                crate::orchestration::execution_facts::ExecutionPhase::None,
                false,
            ),
            crate::orchestration::execution_facts::ChangeExecutionState::Queued
        );
    }

    /// The bug this change repairs: diagnostic deduplication must not be able
    /// to erase or withhold the current blocker, and a reducer that never saw
    /// the original transition must still report it.
    #[test]
    fn dependency_blocker_projection_rebuild() {
        let mut state = OrchestratorState::new(vec!["beta".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("beta".to_string()));
        state.reconcile_dependency_blocker("beta", &["alpha".to_string()]);

        // Same unresolved set, reclassified: no state churn, and no revision of
        // the projection either.
        assert!(
            !state.reconcile_dependency_blocker("beta", &["alpha".to_string()]),
            "an unchanged fingerprint must settle as a no-op, not a rewrite"
        );
        assert_eq!(state.display_status("beta"), "blocked");

        // A changed blocker set is still reconciled, not suppressed.
        assert!(
            state.reconcile_dependency_blocker("beta", &["alpha".to_string(), "gamma".to_string()])
        );
        assert_eq!(
            state.blocker_view("beta").expect("blocker").dependencies,
            vec!["alpha".to_string(), "gamma".to_string()]
        );

        // Reducer replacement — a refresh, a restart of the projection, any
        // loss of the ephemeral diagnostic fingerprints. The next
        // classification republishes the same current blocker without needing a
        // changed fingerprint to justify itself.
        let mut rebuilt = OrchestratorState::new(vec!["beta".to_string()], 0);
        rebuilt.apply_command(ReducerCommand::AddToQueue("beta".to_string()));
        assert_eq!(
            rebuilt.display_status("beta"),
            "queued",
            "a fresh reducer starts from retained queue intent alone"
        );
        assert!(rebuilt.blocker_view("beta").is_none());

        rebuilt.reconcile_dependency_blocker("beta", &["alpha".to_string()]);
        assert_eq!(rebuilt.display_status("beta"), "blocked");
        assert_eq!(
            rebuilt.blocker_view("beta").expect("blocker").dependencies,
            vec!["alpha".to_string()]
        );
    }

    /// Resolution clears the wait exactly once and hands the row back to its
    /// retained queue intent — and nothing else's hold is cleared with it.
    #[test]
    fn dependency_blocker_projection_resolution() {
        let mut state = OrchestratorState::new(vec!["beta".to_string(), "held".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("beta".to_string()));
        state.apply_command(ReducerCommand::AddToQueue("held".to_string()));
        state.reconcile_dependency_blocker("beta", &["alpha".to_string()]);

        assert!(state.clear_dependency_blocker("beta"));
        assert_eq!(
            state.display_status("beta"),
            "queued",
            "the retained queue intent decides the display again"
        );
        assert!(state.blocker_view("beta").is_none());
        assert!(
            !state.clear_dependency_blocker("beta"),
            "clearing an already-cleared wait changes nothing"
        );

        // An unrelated hold belongs to another owner and survives dependency
        // evidence about a different change.
        state.runtime_entry("held").transition_to_stalled(
            "acceptance_finding",
            "no progress",
            "snapshot",
        );
        assert!(!state.clear_dependency_blocker("held"));
        assert_eq!(state.display_status("held"), "stalled");
    }

    /// A ready change waiting only for an execution slot keeps plain `queued`:
    /// the projection reports the blockers classification found, and fabricates
    /// none.
    #[test]
    fn dependency_blocker_projection_capacity_only() {
        let mut state = OrchestratorState::new(vec!["ready".to_string()], 0);
        state.apply_command(ReducerCommand::AddToQueue("ready".to_string()));

        // Classification found nothing unresolved, so the withdrawal path runs
        // and leaves the row exactly as it was.
        assert!(!state.clear_dependency_blocker("ready"));
        assert_eq!(state.display_status("ready"), "queued");
        assert!(
            state.blocker_view("ready").is_none(),
            "an occupied execution slot is not a blocker"
        );
        let rt = state.change_runtime("ready").expect("runtime entry");
        assert_eq!(rt.wait_state, WaitState::None);
        assert_eq!(rt.blocker_kind(), BlockerKind::None);
    }

    /// A terminal or active row is never decorated with a dependency wait: the
    /// reconciliation path applies exactly the guard the transition path does.
    #[test]
    fn dependency_blocker_projection_refuses_terminal_and_active_rows() {
        let mut state = OrchestratorState::new(vec!["done".to_string(), "running".to_string()], 0);
        state.runtime_entry("done").terminal = TerminalState::Merged;
        state.runtime_entry("running").activity = ActivityState::Applying;

        assert!(!state.reconcile_dependency_blocker("done", &["alpha".to_string()]));
        assert!(!state.reconcile_dependency_blocker("running", &["alpha".to_string()]));
        assert_eq!(state.display_status("done"), "merged");
        assert_eq!(state.display_status("running"), "applying");
    }
}