cflx 0.6.327

//! Queue state management and dispatch coordination for [`super::ParallelExecutor`].
//!
//! This module handles queue monitoring, debounce logic, dependency checking,
//! and the reanalysis/dispatch cycle that drives the parallel execution scheduler.

use std::collections::HashSet;
use std::sync::atomic::Ordering;
use std::sync::Arc;

use crate::analyzer::{AnalysisOutcome, AnalysisResult};
use crate::dependency_targets::DependencyTargetClass;
use crate::error::{OrchestratorError, Result};
use crate::events::{ExecutionEvent, LogEntry};
use crate::orchestration::state::{StaleResolveEvidence, StaleResolveSettlement, WaitState};
use crate::orchestration::{
    execute_rejection_flow, handle_blocked_from_rejecting, handle_resume_apply_from_rejecting,
    run_rejection_review, RejectionReviewVerdict,
};
use crate::parallel::analysis_signature::{
    build_analysis_input_signature, proposal_digest_from_path, AnalysisInputMaterials,
    AnalysisInputSignature, BoundedAnalysisRetry, BoundedRetryCause, CompletedAnalysisInput,
};
use crate::parallel::dedup::DiagnosticDeduplicationKey;
use tokio::sync::mpsc;
use tokio::task::JoinSet;
use tracing::{debug, error, info, warn};

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(super) struct QueueReconciliationOutcome {
    pub queued_added: usize,
    pub repair_added: usize,
    /// Scheduler-local candidates dropped because reducer intent revoked them.
    pub revoked_removed: usize,
    /// Queued intent explicitly reconciled away because a fresh
    /// repository-visible view proved the change is not an active candidate.
    ///
    /// Counted separately from [`Self::revoked_removed`]: that one records an
    /// operator revocation this pass merely observed, while this one records a
    /// transition the scheduler itself submitted to stop a queued projection
    /// that had no scheduler-local work, no wake edge, and no typed wait behind
    /// it.
    pub unavailable_reconciled: usize,
    /// Queued intent left exactly as it was because the archived-dirty repair
    /// probe produced no answer at all.
    ///
    /// This is the counterpart of [`Self::unavailable_reconciled`], and the two
    /// are mutually exclusive per change: a settle needs proven absence, and an
    /// undetermined repair precondition is not proof of anything. Counted so a
    /// deferral is observable rather than indistinguishable from a pass that
    /// simply found nothing to do.
    pub repair_evidence_deferred: usize,
}

/// What one scheduler pass's explicit-retry Step 0 found.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(super) struct RetryEdgeConsumption {
    /// Edges this pass took out of the dynamic queue.
    ///
    /// Committed operator intent this pass is the first to observe, which is what
    /// the persistent-idle episode rearms on. A latch retained from an earlier
    /// pass is deliberately not counted here: rearming on it again would republish
    /// the closing Ready edge on every later wake.
    pub newly_drained: usize,
    /// Whether analysis-bypass authority is armed for this pass, from a newly
    /// drained edge or from one an earlier pass could not spend.
    pub bypass_armed: bool,
}

fn analysis_attempt_id(
    iteration: u32,
    trigger: impl std::fmt::Display,
    queued: &[crate::openspec::Change],
) -> String {
    let mut queued_ids: Vec<&str> = queued.iter().map(|change| change.id.as_str()).collect();
    queued_ids.sort_unstable();
    format!(
        "iteration={};trigger={};queued={}",
        iteration,
        trigger,
        queued_ids.join(",")
    )
}

impl QueueReconciliationOutcome {
    #[cfg(test)]
    pub fn total_added(self) -> usize {
        self.queued_added + self.repair_added
    }

    pub fn has_queued_additions(self) -> bool {
        self.queued_added > 0
    }

    pub fn has_repair_additions(self) -> bool {
        self.repair_added > 0
    }
}

/// Outcome of validating one dynamic queue wake-up hint against reducer intent.
///
/// The refusal carries the diagnostic reason and operator-facing log line so the
/// ingestion loop keeps a single skip path and every refusal is observable.
#[derive(Debug)]
enum DynamicQueueAdmission {
    Admit,
    Refuse {
        reason: &'static str,
        log: String,
    },
    /// Disposition is unknown because reducer evidence is incomplete.
    ///
    /// This is not a refusal. A popped hint may be the only wake edge a queued
    /// change has, so the caller puts it back untouched instead of consuming it,
    /// and no scheduler-visible queue state is derived from the gap.
    Retain,
}

#[cfg(test)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum QueuedWorkClass {
    DispatchableApply,
    ManualMergeWait,
    SchedulerLaneWait,
    TerminalErrorRetryRequired,
    DependencyBlocked,
    CandidateUnavailable,
    AcceptanceStalled,
}

#[derive(Debug, Clone, Default)]
pub(super) struct BlockedOnlyQueueClassification {
    pub dispatchable: Vec<crate::openspec::Change>,
    pub manual_merge_wait: Vec<String>,
    pub scheduler_lane_wait: Vec<String>,
    pub terminal_error_retry_required: Vec<String>,
    pub dependency_blocked: Vec<String>,
    pub candidate_unavailable: Vec<String>,
    /// Changes held by a reconciled runtime acceptance stall. They are waiting
    /// work, never dispatchable, so ordinary reconciliation cannot re-submit
    /// them and produce a `Blocked -> Blocked` loop.
    pub acceptance_stalled: Vec<String>,
}

impl BlockedOnlyQueueClassification {
    pub fn has_dispatchable_apply(&self) -> bool {
        !self.dispatchable.is_empty()
    }

    pub fn has_blocked_or_waiting_work(&self) -> bool {
        !self.manual_merge_wait.is_empty()
            || !self.scheduler_lane_wait.is_empty()
            || !self.terminal_error_retry_required.is_empty()
            || !self.dependency_blocked.is_empty()
            || !self.candidate_unavailable.is_empty()
            || !self.acceptance_stalled.is_empty()
    }

    pub fn is_blocked_only(&self) -> bool {
        !self.has_dispatchable_apply() && self.has_blocked_or_waiting_work()
    }

    #[cfg(test)]
    pub fn class_for(&self, change_id: &str) -> Option<QueuedWorkClass> {
        if self
            .dispatchable
            .iter()
            .any(|change| change.id == change_id)
        {
            Some(QueuedWorkClass::DispatchableApply)
        } else if self.manual_merge_wait.iter().any(|id| id == change_id) {
            Some(QueuedWorkClass::ManualMergeWait)
        } else if self.scheduler_lane_wait.iter().any(|id| id == change_id) {
            Some(QueuedWorkClass::SchedulerLaneWait)
        } else if self
            .terminal_error_retry_required
            .iter()
            .any(|id| id == change_id)
        {
            Some(QueuedWorkClass::TerminalErrorRetryRequired)
        } else if self.dependency_blocked.iter().any(|id| id == change_id) {
            Some(QueuedWorkClass::DependencyBlocked)
        } else if self.acceptance_stalled.iter().any(|id| id == change_id) {
            Some(QueuedWorkClass::AcceptanceStalled)
        } else if self.candidate_unavailable.iter().any(|id| id == change_id) {
            Some(QueuedWorkClass::CandidateUnavailable)
        } else {
            None
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum QueueReconciliationDiagnosticLevel {
    Info,
    /// A fresh repository-visible view made a previously missing candidate
    /// loadable, and this pass admitted it.
    Admitted,
    /// An explicit reducer transition settled queued intent this pass proved
    /// could never produce scheduler-local work.
    Settled,
    Warn,
}

/// What one fresh repository-visible lookup proved about a queued candidate.
///
/// The three arms are deliberately distinct: "the catalog says no" and "the
/// catalog could not be read" are not the same fact, and only the first may
/// justify reconciling accepted queue intent away.
enum FreshCandidateLookup {
    /// The active-change catalog now contains the change.
    Loadable(Box<crate::openspec::Change>),
    /// The catalog was read successfully and does not contain the change.
    Absent,
    /// The catalog could not be read, so absence is unproven.
    Unreadable(String),
}

/// What one archived-dirty repair probe proved about a reducer-queued change
/// the pass's active-change catalog view could not load.
///
/// Mirrors [`FreshCandidateLookup`] on purpose: "no archived-dirty repair
/// candidate applies" and "the repair evidence could not be gathered" are
/// different facts, and only the first is a precondition the missing-candidate
/// verdict may be built on.
enum ArchivedDirtyRepairEvidence {
    /// The change has a repairable archived-dirty workspace revision.
    Candidate(Box<crate::openspec::Change>),
    /// The evidence was read and proves no archived-dirty repair applies.
    NotApplicable,
    /// The evidence could not be gathered, so repair availability is unproven.
    ///
    /// Carries the bounded reason the probe could not answer, for diagnostics.
    Undetermined(&'static str),
}

use super::cleanup::WorkspaceCleanupGuard;
use super::dependency::{DependencyContext, EffectiveDependencyBaseEvidence};
use super::dispatch::archived_dirty_repair_candidate_from_workspace;
use super::dynamic_queue::ReanalysisReason;
use super::events::send_event;
use super::lifecycle_slots::{merge_result_transition, workspace_completion_transition};
use super::manual_continuation::ManualContinuationAuthorization;
use super::merge::base_dirty_reason;
use super::work_snapshot::ReducerWorkSnapshot;
use super::{
    AlreadyReportedFailureKind, MergeResult, MergeResultDisposition, MergeResultOrigin,
    MergeTaskOutcome, ParallelEvent, ParallelExecutor, WorkspaceResult,
};

pub(super) struct ReanalysisDispatchContext<'a, F> {
    pub queued: &'a mut Vec<crate::openspec::Change>,
    pub in_flight: &'a mut HashSet<String>,
    /// The evaluation-scoped reducer view, when the caller already captured one.
    ///
    /// The scheduler loop captures exactly one view per pass and threads it
    /// here, so admission, reconciliation, classification, and the drain/idle
    /// decision all read the same reducer revision. `None` asks this evaluation
    /// to await its own equivalent view; direct-call tests use that form.
    pub work_snapshot: Option<&'a ReducerWorkSnapshot>,
    pub max_parallelism: usize,
    pub iteration: u32,
    pub reanalysis_reason: ReanalysisReason,
    pub analyzer: &'a F,
    pub join_set: &'a mut JoinSet<WorkspaceResult>,
    pub cleanup_guard: &'a mut WorkspaceCleanupGuard,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ReanalysisFlowDecision {
    Continue,
    Done { should_break: bool, iteration: u32 },
}

impl ReanalysisFlowDecision {
    fn done(should_break: bool, iteration: u32) -> Self {
        Self::Done {
            should_break,
            iteration,
        }
    }

    fn into_result(self) -> Option<(bool, u32)> {
        match self {
            Self::Continue => None,
            Self::Done {
                should_break,
                iteration,
            } => Some((should_break, iteration)),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct ReanalysisExecutionDecision {
    effective_reason: ReanalysisReason,
    /// Dispatch capacity recomputed for *this* evaluation.
    ///
    /// The analyzer gate and the trigger-lifetime rule both read this one value
    /// rather than recomputing it separately, so they can never disagree about
    /// whether the same pass had capacity.
    available_slots: usize,
}

enum DependencyAnalysisAttempt {
    Completed(AnalysisOutcome),
    EmptyOrder(ReanalysisFlowDecision),
}

/// Everything the pre-analysis pipeline evaluates before the analyzer runs.
///
/// Grouped rather than passed positionally so the whole pipeline is visibly
/// driven by one evaluation's inputs — including `work_snapshot`, the coherent
/// reducer view every gate in it must agree on.
struct DependencyAnalysisInputs<'a, F> {
    queued: &'a mut [crate::openspec::Change],
    in_flight: &'a HashSet<String>,
    max_parallelism: usize,
    iteration: u32,
    reanalysis_reason: ReanalysisReason,
    analyzer: &'a F,
    work_snapshot: &'a ReducerWorkSnapshot,
}

/// One pass through the pre-analysis pipeline.
///
/// `captured_signature` is the analysis-input snapshot taken immediately before the analyzer
/// was invoked. Recording that pre-analysis value (rather than recomputing it afterwards) is
/// what keeps a queue or repository change that happened *during* analysis visible to the
/// next probe instead of being hidden by the newer state.
struct DependencyAnalysisPass {
    attempt: DependencyAnalysisAttempt,
    captured_signature: Option<AnalysisInputSignature>,
}

impl DependencyAnalysisPass {
    fn terminal(decision: ReanalysisFlowDecision) -> Self {
        Self {
            attempt: DependencyAnalysisAttempt::EmptyOrder(decision),
            captured_signature: None,
        }
    }
}

/// Whether `reason` is an explicit scheduler edge that must analyze once per event.
///
/// These are exactly the reasons that already bypass the queue-coalescing debounce, so they
/// keep a single authoritative meaning: a real state-transition event gets one immediate
/// evaluation, matching signature or not.
///
/// `Completion` is deliberately absent. It is not consumed as a one-shot edge and does not
/// bypass debounce today, so treating it as an unconditional bypass would let a retained
/// reason replay analysis on every wake. It needs no bypass either: a task completion always
/// changes in-flight membership or capacity, which changes the signature on its own.
fn bypasses_unchanged_input_gate(reason: ReanalysisReason) -> bool {
    matches!(
        reason,
        ReanalysisReason::QueueNotification
            | ReanalysisReason::SlotRecovery
            | ReanalysisReason::ResolveCompletion
            | ReanalysisReason::RepairCandidate
    )
}

/// Whether the ordinary-timer suppression gate allows this pass to invoke the analyzer.
enum UnchangedInputGate {
    /// Invoke the analyzer. `captured` is `None` when no signature could be built, in which
    /// case the pass runs fail-open and records nothing.
    Analyze {
        captured: Option<AnalysisInputSignature>,
    },
    /// The current input equals an already completed input; skip the expensive analyzer.
    Suppress,
}

/// Result of one dispatch-selection pass.
///
/// `selected` is needed by the suppression decision: a pass that selects nothing while
/// capacity is positive and nothing is in flight must stay eligible for the next debounced
/// timer evaluation.
struct DispatchSelectionOutcome {
    iteration: u32,
    selected: usize,
}

struct DispatchSelectedCandidatesContext<'a> {
    queued: &'a mut Vec<crate::openspec::Change>,
    in_flight: &'a mut HashSet<String>,
    iteration: u32,
    analysis_result: &'a AnalysisResult,
    available_slots: usize,
    join_set: &'a mut JoinSet<WorkspaceResult>,
    cleanup_guard: &'a mut WorkspaceCleanupGuard,
}

fn on_merged_failure_message(change_id: &str, error: &OrchestratorError) -> String {
    format!(
        "on_merged hook failed for '{}'; merged transition blocked: {}",
        change_id, error
    )
}

impl ParallelExecutor {
    /// Check if debounce period has elapsed for queue changes.
    ///
    /// Returns `true` if:
    /// - No recent queue changes, OR
    /// - 10 seconds have passed since the last queue change
    ///
    /// This prevents immediate re-analysis when the queue changes, giving time for
    /// multiple changes to be queued before triggering expensive re-analysis.
    ///
    /// Note: This is now separated from slot availability check. Re-analysis can
    /// proceed even when available_slots == 0, and the next dispatch will happen
    /// when slots become available.
    pub async fn should_reanalyze(&self, bypass_debounce: bool) -> bool {
        super::dynamic_queue::should_reanalyze_queue(&self.last_queue_change_at, bypass_debounce)
            .await
    }

    pub(super) fn is_cancelled(&self) -> bool {
        self.cancel_token
            .as_ref()
            .is_some_and(|token| token.is_cancelled())
    }

    /// Capture one coherent reducer work view for this scheduler evaluation.
    ///
    /// The read is *awaited*, never attempted: a scheduler pass that collides
    /// with a short reducer write suspends here and is woken by Tokio the moment
    /// the writer releases, which continues the same evaluation without needing
    /// another queue mutation or wake notification. Turning that collision into
    /// queue state — a refused hint, an empty reconciliation, a
    /// `candidate_unavailable` classification, a drained finite run, or a
    /// persistent idle wait — is what stranded queued work.
    ///
    /// The guard is dropped before returning. Every repository probe, VCS call,
    /// dependency analysis, and dispatch downstream of a capture therefore runs
    /// with no reducer guard held, so reducer events and operator commands can
    /// still take the write lock while classification continues.
    ///
    /// Cancellation stays responsive: when a cancel token is wired, it races the
    /// acquisition and wins, yielding [`ReducerWorkSnapshot::incomplete`]. That
    /// view withholds every candidate and, because it is not complete, cannot
    /// authorize drain, finite termination, or idle either.
    pub(super) async fn capture_reducer_work_snapshot(&self) -> ReducerWorkSnapshot {
        let Some(shared) = &self.shared_orchestrator_state else {
            return ReducerWorkSnapshot::absent();
        };

        match &self.cancel_token {
            Some(token) => {
                tokio::select! {
                    biased;
                    _ = token.cancelled() => {
                        debug!("Abandoned reducer work snapshot acquisition after cancellation");
                        ReducerWorkSnapshot::incomplete()
                    }
                    guard = shared.read() => ReducerWorkSnapshot::from_state(&guard),
                }
            }
            None => {
                let guard = shared.read().await;
                ReducerWorkSnapshot::from_state(&guard)
            }
        }
    }

    /// Await one coherent reducer view and sync the scheduler-local lane waits.
    pub(super) async fn sync_resolve_wait_from_shared_state(&mut self) {
        let snapshot = self.capture_reducer_work_snapshot().await;
        self.sync_resolve_wait_from_snapshot(&snapshot);
    }

    /// Copy reducer-owned lane waits out of an already-captured view.
    ///
    /// An incomplete view carries no lane evidence, so the scheduler-local sets
    /// are left exactly as they were rather than being cleared into a false
    /// "no lane is waiting".
    pub(super) fn sync_resolve_wait_from_snapshot(&mut self, snapshot: &ReducerWorkSnapshot) {
        if !snapshot.reducer_present() || !snapshot.is_complete() {
            return;
        }
        self.resolve_wait_changes = snapshot.resolve_wait_ids().clone();
        self.reject_wait_changes = snapshot.reject_wait_ids().clone();
    }

    pub(super) async fn clear_resolve_wait_intent_for_outcome(&mut self, change_id: &str) {
        self.resolve_wait_changes.remove(change_id);
        self.last_dispatched_resolve_wait_changes.remove(change_id);
        if let Some(shared) = &self.shared_orchestrator_state {
            let mut guard = shared.write().await;
            guard.clear_resolve_wait_intent(change_id);
        }
    }

    /// Settle a stale deferred resolve retry, then release scheduler-local ownership.
    ///
    /// Ordering is the point: the shared reducer records the typed outcome first,
    /// and scheduler-local retry membership plus base-lane occupancy are released
    /// only afterwards. A refresh that lands mid-settlement therefore sees either
    /// the accepted retry intent or the settled outcome, never an idle
    /// `not queued` row in between.
    ///
    /// Repository content is never touched here: this is bookkeeping over
    /// evidence that was already read.
    pub(super) async fn settle_stale_resolve_retry_for(
        &mut self,
        change_id: &str,
        evidence: StaleResolveEvidence,
    ) -> StaleResolveSettlement {
        let settlement = match &self.shared_orchestrator_state {
            Some(shared) => {
                let mut guard = shared.write().await;
                guard.settle_stale_resolve_retry(change_id, evidence)
            }
            // No reducer is wired: the scheduler still classifies the retry the
            // same way so its own bookkeeping stays truthful.
            None => match evidence {
                StaleResolveEvidence::Proven => StaleResolveSettlement::Merged,
                StaleResolveEvidence::Absent | StaleResolveEvidence::Unknown => {
                    StaleResolveSettlement::MergeWaitRetained
                }
            },
        };

        self.resolve_wait_changes.remove(change_id);
        self.last_dispatched_resolve_wait_changes.remove(change_id);
        match settlement {
            StaleResolveSettlement::MergeWaitRetained => {
                self.merge_wait_changes.insert(change_id.to_string());
            }
            StaleResolveSettlement::Merged | StaleResolveSettlement::AlreadySettled => {
                self.merge_wait_changes.remove(change_id);
            }
        }
        self.abandon_base_mutating_lane_occupant_for_give_up(change_id)
            .await;
        settlement
    }

    pub(super) async fn clear_reject_wait_intent_for_success(&mut self, change_id: &str) {
        self.reject_wait_changes.remove(change_id);
        self.last_dispatched_reject_wait_changes.remove(change_id);
        if let Some(shared) = &self.shared_orchestrator_state {
            let mut guard = shared.write().await;
            guard.clear_reject_wait_intent(change_id);
        }
    }

    async fn abandon_base_mutating_lane_occupant_for_give_up(&mut self, change_id: &str) {
        if let Some(shared) = &self.shared_orchestrator_state {
            let mut guard = shared.write().await;
            let released = guard.abandon_base_mutating_lane_occupant(change_id);
            if released {
                info!(
                    change_id = %change_id,
                    "Released base-mutating lane after spawned retry give-up"
                );
            }
        }
    }

    /// Pre-apply a *successful* rejection-review verdict to the shared reducer.
    ///
    /// Deliberately limited to the success variants. A `RejectionReviewFailed`
    /// is a mark-revoking transition, and the authoritative dispatcher can only
    /// see that edge if it is the one that applies the event: reducing it here
    /// first would leave the dispatched copy with nothing to compare and the
    /// stale execution mark in place.
    async fn apply_rejection_review_event_in_shared_state(&mut self, event: &ExecutionEvent) {
        if let Some(shared) = &self.shared_orchestrator_state {
            let mut guard = shared.write().await;
            guard.apply_execution_event(event);
        }
    }

    pub(super) async fn mark_deferred_merge_completed_in_shared_state(
        &mut self,
        change_id: &str,
        revision: &str,
    ) {
        if let Some(shared) = &self.shared_orchestrator_state {
            let mut guard = shared.write().await;
            guard.apply_execution_event(&ExecutionEvent::MergeCompleted {
                change_id: change_id.to_string(),
                revision: revision.to_string(),
            });
        }
    }

    pub(super) fn trigger_resolve_wait_retry_dispatch(&mut self) {
        self.resolve_wait_retry_triggered = true;
    }

    pub(super) fn should_dispatch_resolve_wait_retry(&self) -> bool {
        if self.resolve_wait_changes.is_empty() && self.reject_wait_changes.is_empty() {
            return false;
        }

        self.resolve_wait_retry_triggered
            || self.last_dispatched_resolve_wait_changes != self.resolve_wait_changes
            || self.last_dispatched_reject_wait_changes != self.reject_wait_changes
    }

    #[allow(dead_code)]
    pub(super) async fn maybe_dispatch_resolve_wait_retry(&mut self) {
        let base_dirty_changed_to_clean = self
            .resolve_wait_base_dirty_changed_to_clean()
            .await
            .unwrap_or(false);

        if !self.should_dispatch_resolve_wait_retry() && !base_dirty_changed_to_clean {
            return;
        }

        self.retry_deferred_base_lane_waiters().await;
        self.last_dispatched_resolve_wait_changes = self.resolve_wait_changes.clone();
        self.last_dispatched_reject_wait_changes = self.reject_wait_changes.clone();
        self.resolve_wait_retry_triggered = false;
    }

    pub(super) async fn maybe_dispatch_resolve_wait_retry_with_tx(
        &mut self,
        merge_result_tx: &mpsc::Sender<MergeResult>,
    ) {
        let base_dirty_changed_to_clean = self
            .resolve_wait_base_dirty_changed_to_clean()
            .await
            .unwrap_or(false);

        if !self.should_dispatch_resolve_wait_retry() && !base_dirty_changed_to_clean {
            return;
        }

        self.dispatch_deferred_base_lane_waiter(merge_result_tx.clone())
            .await;
    }

    async fn resolve_wait_base_dirty_changed_to_clean(&mut self) -> Result<bool> {
        if self.resolve_wait_changes.is_empty() && self.reject_wait_changes.is_empty() {
            self.last_resolve_wait_base_dirty = None;
            return Ok(false);
        }

        let base_dirty = base_dirty_reason(&self.repo_root).await?.is_some();
        let changed_to_clean =
            matches!(self.last_resolve_wait_base_dirty, Some(true)) && !base_dirty;
        if changed_to_clean {
            info!(
                repo_root = %self.repo_root.display(),
                resolve_wait_count = self.resolve_wait_changes.len(),
                reject_wait_count = self.reject_wait_changes.len(),
                "Base repository transitioned from dirty to clean while base-lane waiters exist; waking retry dispatch"
            );
        }
        self.last_resolve_wait_base_dirty = Some(base_dirty);
        Ok(changed_to_clean)
    }

    /// Whether reducer-owned or scheduler-local base-lane retry intent exists.
    ///
    /// Awaited rather than attempted: a caller uses this to decide whether a run
    /// has anything to do at all, and reporting "no lane wait" because a reducer
    /// write was in progress would turn real retry intent into a skipped run.
    pub(crate) async fn has_resolve_wait(&self) -> bool {
        let snapshot = self.capture_reducer_work_snapshot().await;
        if snapshot.reducer_present() && snapshot.is_complete() {
            return !snapshot.resolve_wait_ids().is_empty()
                || !snapshot.reject_wait_ids().is_empty();
        }
        !self.resolve_wait_changes.is_empty() || !self.reject_wait_changes.is_empty()
    }

    #[allow(dead_code)]
    pub(super) fn skip_reason_for_change(&self, change_id: &str) -> Option<String> {
        // Only skip changes with failed dependencies (not merge-wait dependencies).
        // Merge-wait dependencies are handled as blocked/queued status via dependency resolution.
        if let Some(failed_dep) = self.failed_tracker.should_skip(change_id) {
            return Some(format!("Dependency '{}' failed", failed_dep));
        }
        None
    }

    fn dependency_blocker_fingerprint(
        change_id: &str,
        blockers: &[(String, DependencyTargetClass)],
    ) -> super::DependencyBlockerFingerprint {
        DependencyContext::blocker_fingerprint(change_id, blockers)
    }

    fn should_emit_dependency_blocked_transition(
        &mut self,
        change_id: &str,
        blockers: &[(String, DependencyTargetClass)],
    ) -> bool {
        let fingerprint = Self::dependency_blocker_fingerprint(change_id, blockers);
        let changed = self
            .dependency_blocker_fingerprints
            .get(change_id)
            .is_none_or(|previous| previous != &fingerprint);

        if changed {
            self.dependency_blocker_fingerprints
                .insert(change_id.to_string(), fingerprint.clone());
            self.diagnostic_dedup
                .should_emit(DiagnosticDeduplicationKey::DependencyBlocker {
                    change_id: change_id.to_string(),
                    fingerprint,
                });
            return true;
        }

        false
    }

    /// Publish the current unresolved dependency set into reducer projection.
    ///
    /// Called on *every* coherent classification, never behind
    /// [`Self::should_emit_dependency_blocked_transition`]. The fingerprint
    /// store decides whether another operator diagnostic is worth emitting; it
    /// is not current-state authority, and using it as one is what let a
    /// refreshed or replaced reducer report an unresolved dependency wait as
    /// plain `queued` with no blocker.
    ///
    /// Nothing durable is written: this is in-memory reducer state for one
    /// process lifetime, reconstructed from repository-visible classification
    /// on the next pass.
    async fn reconcile_dependency_blocker_projection(
        &self,
        change_id: &str,
        dependency_ids: &[String],
    ) {
        let Some(shared) = &self.shared_orchestrator_state else {
            return;
        };
        let changed = {
            let mut guard = shared.write().await;
            guard.reconcile_dependency_blocker(change_id, dependency_ids)
        };
        if changed {
            debug!(
                change_id,
                dependency_ids = ?dependency_ids,
                "Reconciled dependency blocker into reducer projection"
            );
        }
    }

    /// Withdraw a dependency wait once classification found nothing unresolved.
    ///
    /// Unconditional for the same reason the publish side is: the retained
    /// queue intent must decide the display again as soon as the evidence says
    /// so, whether or not a `DependencyResolved` diagnostic edge was emitted.
    async fn clear_dependency_blocker_projection(&self, change_id: &str) {
        let Some(shared) = &self.shared_orchestrator_state else {
            return;
        };
        let cleared = {
            let mut guard = shared.write().await;
            guard.clear_dependency_blocker(change_id)
        };
        if cleared {
            debug!(
                change_id,
                "Cleared dependency blocker projection after repository-visible resolution"
            );
        }
    }

    async fn emit_dependency_blocker_diagnostic(
        &self,
        change_id: &str,
        blockers: &[(String, DependencyTargetClass)],
    ) {
        for (dep_id, class) in blockers {
            let message = if matches!(class, DependencyTargetClass::Resolving) {
                format!(
                    "Change '{}' blocked by resolving dependency '{}' and will remain queued",
                    change_id, dep_id
                )
            } else if matches!(class, DependencyTargetClass::Archived) {
                match self.effective_dependency_base().await {
                    Ok(effective_base) => format!(
                        "Change '{}' blocked by archived-but-not-merged dependency '{}' on effective dependency base '{}' and will remain queued",
                        change_id, dep_id, effective_base
                    ),
                    Err(err) => format!(
                        "Change '{}' blocked by archived dependency '{}' while effective dependency base could not be determined ({}) and will remain queued",
                        change_id, dep_id, err
                    ),
                }
            } else {
                format!(
                    "Change '{}' blocked by {} dependency '{}' and will remain queued",
                    change_id,
                    class.as_str(),
                    dep_id
                )
            };
            match class {
                DependencyTargetClass::Missing
                | DependencyTargetClass::Rejected
                | DependencyTargetClass::Error => warn!("{}", message),
                DependencyTargetClass::Queued
                | DependencyTargetClass::InFlight
                | DependencyTargetClass::Resolving
                | DependencyTargetClass::ActiveButNotQueued => info!("{}", message),
                DependencyTargetClass::Archived => info!("{}", message),
            }
            if matches!(
                class,
                DependencyTargetClass::Missing
                    | DependencyTargetClass::Rejected
                    | DependencyTargetClass::Error
            ) {
                send_event(&self.event_tx, ParallelEvent::Error { message }).await;
            }
        }
    }

    /// Select the repository-visible base used for archived dependency merge checks.
    ///
    /// Ordinary runs use the original branch captured at scheduler startup. Stacked
    /// orchestration can advance a separate integration branch after startup; when the
    /// repository is currently on such a branch, that current branch is the effective
    /// dependency base for dispatch decisions.
    async fn effective_dependency_base(&self) -> Result<String> {
        let mut dependency_context =
            DependencyContext::from_executor(self, std::iter::empty::<&str>(), &HashSet::new())
                .await;
        dependency_context
            .effective_dependency_base(self.workspace_manager.as_ref())
            .await
            .map(str::to_string)
    }

    /// Repository-visible evidence describing the effective dependency base.
    ///
    /// Signature probing and dependency merge checks share this one resolution path, so the
    /// analysis-input signature can never be invalidated by a different ref than the one
    /// dispatch eligibility is decided from.
    pub(super) async fn effective_dependency_base_evidence(
        &self,
    ) -> Result<EffectiveDependencyBaseEvidence> {
        let mut dependency_context =
            DependencyContext::from_executor(self, std::iter::empty::<&str>(), &HashSet::new())
                .await;
        dependency_context
            .effective_dependency_base_evidence(self.workspace_manager.as_ref())
            .await
    }

    /// Check if a dependency is resolved (merged to the effective dependency base).
    ///
    /// A dependency is considered resolved if its archive commit is present in the
    /// effective base branch/tree and its active change directory is absent there.
    /// This preserves the archive-only safety guard while allowing stacked runs to
    /// unblock dependents once the dependency is merged into the integration branch.
    #[allow(dead_code)]
    pub(super) async fn is_dependency_resolved(&self, dep_id: &str) -> Result<bool> {
        let (is_resolved, _) = self.is_dependency_resolved_with_base(dep_id).await?;
        Ok(is_resolved)
    }

    async fn is_dependency_resolved_with_base(&self, dep_id: &str) -> Result<(bool, String)> {
        let mut dependency_context =
            DependencyContext::from_executor(self, std::iter::empty::<&str>(), &HashSet::new())
                .await;
        dependency_context
            .is_dependency_resolved_with_base(dep_id, self.workspace_manager.as_ref())
            .await
    }

    /// Number of manual resolve operations a frontend reports as running.
    ///
    /// Observability and drain evidence only. It is deliberately *not* a
    /// capacity input: see [`Self::calculate_available_slots`].
    pub(super) fn manual_resolve_active(&self) -> usize {
        self.manual_resolve_count
            .as_ref()
            .map(|counter| counter.load(std::sync::atomic::Ordering::Relaxed))
            .unwrap_or(0)
    }

    /// Calculate available execution slots from unique lifecycle occupancy.
    ///
    /// Occupancy is the union of the caller's in-flight set and the lifecycle
    /// membership, deduplicated by change ID, so one admitted change counts
    /// exactly once no matter how many of its phases are observable at the same
    /// instant. That union — not the two sets separately — is why apply,
    /// acceptance, archive, background merge, automatic resolve, manual resolve,
    /// and a retained `merge wait` cannot add up to more than
    /// `max_concurrent_workspaces`.
    ///
    /// The resolve counters are *not* subtracted any more. A resolve runs inside
    /// the lifecycle slot its change already owns, so subtracting it as well
    /// double-counted one admitted change and, worse, reserved nothing at all
    /// for the conflict-free background merge that owns most of the interval.
    ///
    /// Retained occupancy the scheduler could not reserve a permit for fails
    /// closed: capacity is reported as zero until the ambiguity is resolved,
    /// because admitting on unproven occupancy is exactly the overrun this
    /// accounting exists to prevent.
    pub(super) fn calculate_available_slots(
        &self,
        max_parallelism: usize,
        in_flight: &HashSet<String>,
    ) -> usize {
        self.lifecycle_slots.available(max_parallelism, in_flight)
    }

    /// Reconstruct and settle lifecycle occupancy for reducer-visible waits.
    ///
    /// Two directions, both driven by evidence rather than by a durable lease:
    ///
    /// - a change the reducer reports in `merge wait`, `resolve pending`, or
    ///   `reject pending` that owns no slot takes one now. This is what makes a
    ///   process restart reconstruct retained occupancy from workspace, Git, and
    ///   reducer evidence, and what stops a wait this process never dispatched
    ///   from being admitted around;
    /// - a retained slot whose change reached terminal settlement, or which has
    ///   demonstrably left the wait it was confirmed in, is released once.
    ///
    /// Only retained-phase slots are released here. A workspace task and
    /// a background merge each own their own release edge and always reach it,
    /// so reconciliation must never race them for a change that is still
    /// executing.
    ///
    /// An incomplete reducer view is no evidence at all and changes nothing.
    pub(super) fn reconcile_retained_lifecycle_slots(&mut self, snapshot: &ReducerWorkSnapshot) {
        if !snapshot.reducer_present() || !snapshot.is_complete() {
            return;
        }

        let retained: HashSet<String> = snapshot
            .merge_wait_ids()
            .iter()
            .chain(snapshot.resolve_wait_ids())
            .chain(snapshot.reject_wait_ids())
            .cloned()
            .collect();

        let report = self.lifecycle_slots.reconcile_retained(
            &retained,
            snapshot.active_ids(),
            snapshot.settled_ids(),
        );

        if !report.reserved.is_empty() {
            info!(
                changes = ?report.reserved,
                "Reconstructed lifecycle slot occupancy for retained base-lane waits"
            );
        }
        if !report.unavailable.is_empty() {
            warn!(
                changes = ?report.unavailable,
                "Retained base-lane waits could not reserve a lifecycle slot; \
                 suppressing new dispatch until occupancy is provable"
            );
        }
        if !report.released.is_empty() {
            info!(
                changes = ?report.released,
                "Released the lifecycle slots of settled retained changes"
            );
        }
    }

    /// Classify queued candidates against the ephemeral failed-change tracker.
    ///
    /// Returns `(change_id, sorted failed blockers)` for every candidate that is
    /// currently held back by a failed dependency, in candidate order.
    ///
    /// Nothing is removed from the caller's candidate list: reducer-admitted
    /// queue intent stays authoritative, so a failed-dependent change remains
    /// locally represented as blocked queued work. Removing it here and letting
    /// reconciliation restore it from reducer intent is exactly the churn that
    /// produced an unbounded skip/re-add/re-analysis cycle.
    pub(super) fn failed_dependency_blocked_candidates(
        &self,
        queued: &[crate::openspec::Change],
    ) -> Vec<(String, Vec<String>)> {
        queued
            .iter()
            .filter_map(|change| {
                let blockers = self.failed_tracker.failed_blockers(&change.id);
                if blockers.is_empty() {
                    None
                } else {
                    Some((change.id.clone(), blockers))
                }
            })
            .collect()
    }

    /// Select changes to dispatch based on order, available slots, and dependency resolution.
    ///
    /// # Arguments
    /// * `analysis_result` - Result from dependency analysis
    /// * `available_slots` - Number of available execution slots
    ///
    /// # Returns
    /// Vector of selected change IDs ready for dispatch
    pub(super) async fn select_changes_for_dispatch(
        &mut self,
        analysis_result: &crate::analyzer::AnalysisResult,
        available_slots: usize,
        in_flight: &HashSet<String>,
    ) -> Vec<String> {
        let mut selected_changes: Vec<String> = Vec::new();
        let mut dependency_context = DependencyContext::from_executor(
            self,
            analysis_result.order.iter().map(String::as_str),
            in_flight,
        )
        .await;

        for change_id in &analysis_result.order {
            if dependency_context.is_terminal_error_change(change_id) {
                info!(
                    change_id = %change_id,
                    "Skipping ordinary apply dispatch because terminal error requires explicit retry"
                );
                continue;
            }

            // The analysis order is computed from the scheduler-local candidate
            // list, so a candidate whose queue intent was revoked after it was
            // added can still appear here whenever some *other* candidate keeps
            // analysis running. Current reducer intent decides, exactly as it
            // does in `classify_queued_work`.
            if dependency_context.withholds_ordinary_queue_intent(change_id) {
                info!(
                    change_id = %change_id,
                    "Skipping ordinary apply dispatch because current reducer intent does not admit ordinary work"
                );
                continue;
            }

            // A failed-dependent candidate stays in the local queue now, so it
            // reaches the analysis order and this gate is what keeps it out of
            // dispatch. Observation is owned by `observe_failed_dependency_blocks`,
            // so nothing is emitted here.
            if let Some(failed_dep) = self.failed_tracker.should_skip(change_id) {
                debug!(
                    change_id = %change_id,
                    dependency = %failed_dep,
                    "Skipping ordinary apply dispatch because a dependency failed in this run"
                );
                continue;
            }

            // Check if change has unresolved dependencies
            let proposal_path = self
                .repo_root
                .join("openspec/changes")
                .join(change_id)
                .join("proposal.md");
            let metadata_dependencies = if self.repo_root.join("openspec/changes").exists() {
                match crate::openspec::parse_proposal_dependencies_strict_from_file(&proposal_path)
                {
                    Ok(dependencies) => dependencies,
                    Err(error) => {
                        let message = format!(
                            "Change '{}' blocked because dependency metadata could not be read from '{}': {}",
                            change_id,
                            proposal_path.display(),
                            error
                        );
                        warn!("{}", message);
                        // Unreadable dependency metadata is still a current
                        // dependency wait: project it before the diagnostic
                        // gate, so a suppressed duplicate cannot leave the row
                        // looking plainly `queued`.
                        self.reconcile_dependency_blocker_projection(
                            change_id,
                            std::slice::from_ref(change_id),
                        )
                        .await;
                        if self.should_emit_dependency_blocked_transition(
                            change_id,
                            &[(change_id.clone(), DependencyTargetClass::Error)],
                        ) {
                            send_event(&self.event_tx, ParallelEvent::Error { message }).await;
                            send_event(
                                &self.event_tx,
                                ParallelEvent::DependencyBlocked {
                                    change_id: change_id.clone(),
                                    dependency_ids: vec![change_id.clone()],
                                },
                            )
                            .await;
                        }
                        continue;
                    }
                }
            } else {
                Vec::new()
            };
            let mut dependencies = analysis_result
                .dependencies
                .get(change_id)
                .cloned()
                .unwrap_or_default();
            for dep_id in metadata_dependencies {
                if dep_id != *change_id && !dependencies.contains(&dep_id) {
                    dependencies.push(dep_id);
                }
            }

            if !dependencies.is_empty() {
                let mut unresolved_deps = Vec::new();
                let mut blockers = Vec::new();
                for dep_id in &dependencies {
                    let class = dependency_context.classify(dep_id);
                    match class {
                        DependencyTargetClass::Missing | DependencyTargetClass::Rejected => {
                            unresolved_deps.push(dep_id.clone());
                            blockers.push((dep_id.clone(), class));
                            continue;
                        }
                        DependencyTargetClass::Error => {
                            warn!(
                                change_id = %change_id,
                                dependency = %dep_id,
                                "Blocking dispatch because dependency is in terminal error and requires explicit retry"
                            );
                            unresolved_deps.push(dep_id.clone());
                            blockers.push((dep_id.clone(), class));
                            continue;
                        }
                        DependencyTargetClass::Archived => {
                            DependencyContext::log_archived_dependency_check(change_id, dep_id);
                        }
                        DependencyTargetClass::Resolving => {
                            unresolved_deps.push(dep_id.clone());
                            blockers.push((dep_id.clone(), class));
                            continue;
                        }
                        DependencyTargetClass::Queued
                        | DependencyTargetClass::InFlight
                        | DependencyTargetClass::ActiveButNotQueued => {}
                    }

                    match dependency_context
                        .is_dependency_resolved_with_base(dep_id, self.workspace_manager.as_ref())
                        .await
                    {
                        Ok((true, effective_base)) => {
                            DependencyContext::log_dependency_resolved(
                                change_id,
                                dep_id,
                                class,
                                &effective_base,
                            );
                        }
                        Ok((false, effective_base)) => {
                            DependencyContext::log_dependency_unresolved(
                                change_id,
                                dep_id,
                                class,
                                &effective_base,
                            );
                            unresolved_deps.push(dep_id.clone());
                            blockers.push((dep_id.clone(), class));
                        }
                        Err(e) => {
                            error!(
                                "Failed to evaluate dependency resolution for '{}' (dependency '{}'): {}",
                                change_id, dep_id, e
                            );
                            send_event(
                                &self.event_tx,
                                ParallelEvent::Error {
                                    message: format!(
                                        "Failed to evaluate dependency resolution for '{}' (dependency '{}'): {}",
                                        change_id, dep_id, e
                                    ),
                                },
                            )
                            .await;
                            unresolved_deps.push(dep_id.clone());
                            blockers.push((dep_id.clone(), class));
                        }
                    }
                }

                if !unresolved_deps.is_empty() {
                    // State first, diagnostics second. The projection describes
                    // what is true right now and is rebuilt on every coherent
                    // classification; the events below describe a *transition*
                    // and stay deduplicated per blocker fingerprint.
                    self.reconcile_dependency_blocker_projection(change_id, &unresolved_deps)
                        .await;
                    if self.should_emit_dependency_blocked_transition(change_id, &blockers) {
                        info!(
                            "Change '{}' blocked: waiting for dependencies {:?}",
                            change_id, unresolved_deps
                        );
                        self.emit_dependency_blocker_diagnostic(change_id, &blockers)
                            .await;
                        send_event(
                            &self.event_tx,
                            ParallelEvent::DependencyBlocked {
                                change_id: change_id.clone(),
                                dependency_ids: unresolved_deps,
                            },
                        )
                        .await;
                    } else {
                        debug!(
                            change_id,
                            blockers = ?blockers,
                            "Suppressing repeated dependency blocked transition while the current \
                             blocker projection stays reconciled"
                        );
                    }
                    continue;
                }
            }

            // Nothing is unresolved for this change any more, so the dependency
            // wait is withdrawn from the projection unconditionally — the
            // retained queue intent decides the display again from here.
            self.clear_dependency_blocker_projection(change_id).await;

            // Check if this change was previously blocked and is now resolved.
            // Clearing the in-memory fingerprint makes the next blocked observation a new transition.
            if self
                .dependency_blocker_fingerprints
                .remove(change_id)
                .is_some()
            {
                info!(
                    "Change '{}' dependencies resolved, forcing fresh workspace recreation",
                    change_id
                );
                self.force_recreate_worktree.insert(change_id.clone());
                send_event(
                    &self.event_tx,
                    ParallelEvent::DependencyResolved {
                        change_id: change_id.clone(),
                    },
                )
                .await;
            }

            if selected_changes.len() < available_slots {
                selected_changes.push(change_id.clone());
            }
        }

        selected_changes
    }

    /// Handle completion of a workspace task (apply+archive).
    ///
    /// This processes success/failure, attempts merge if archived, and triggers cleanup.
    ///
    /// # Lifecycle slot
    ///
    /// The returning task does not end the change's admitted lifecycle, so its
    /// slot is *transferred* here rather than released: a successful archive
    /// hands it to background merge handling, and only a terminal error, a
    /// rejection, or a completion with nothing left to merge releases it. The
    /// transfer happens before `in_flight` membership is dropped and before the
    /// merge task is spawned, so no interleaving leaves the change owning zero
    /// slots.
    ///
    /// # Arguments
    /// * `workspace_result` - Result from completed workspace task
    /// * `max_parallelism` - Maximum parallelism for logging
    /// * `in_flight` - Set to update (remove completed change)
    /// * `cleanup_guard` - Guard for workspace cleanup tracking
    pub(super) async fn handle_workspace_completion(
        &mut self,
        workspace_result: WorkspaceResult,
        max_parallelism: usize,
        in_flight: &mut HashSet<String>,
        merge_result_tx: &mpsc::Sender<MergeResult>,
    ) {
        // Decide the slot's next owner *before* the change leaves `in_flight`,
        // so capacity is never computed from a moment where neither the
        // in-flight set nor the lifecycle phase accounts for this change.
        self.lifecycle_slots.apply_transition(
            &workspace_result.change_id,
            workspace_completion_transition(
                workspace_result.error.is_some(),
                workspace_result.rejected.is_some(),
                workspace_result.final_revision.is_some(),
            ),
        );

        // Remove from in-flight
        in_flight.remove(&workspace_result.change_id);

        // Every ordinary workspace-task return funnels through here, including
        // the ones that never emitted an operation-started event (an already
        // merged resume, a deferred rejection review, a pre-operation stop). One
        // clearing event here is what keeps `preparing` from outliving the
        // dispatch that announced it; it is a no-op once a real transition has
        // already taken over.
        send_event(
            &self.event_tx,
            ParallelEvent::WorkspacePreparationEnded {
                change_id: workspace_result.change_id.clone(),
            },
        )
        .await;

        // Clean up kill token registry
        if let Some(ref queue) = self.dynamic_queue {
            queue
                .unregister_kill_token(&workspace_result.change_id)
                .await;
        }

        info!(
            "Task completed: change='{}', in_flight={}, available_slots={}, lifecycle_slots={}, error={:?}",
            workspace_result.change_id,
            in_flight.len(),
            self.calculate_available_slots(max_parallelism, in_flight),
            self.lifecycle_slots.len(),
            workspace_result.error
        );

        // Handle result (failure, rejection, or success)
        if let Some(error) = &workspace_result.error {
            error!("Change '{}' failed: {}", workspace_result.change_id, error);
            self.failed_tracker.mark_failed(&workspace_result.change_id);
            send_event(
                &self.event_tx,
                ParallelEvent::ProcessingError {
                    id: workspace_result.change_id.clone(),
                    error: error.clone(),
                },
            )
            .await;

            // Rejection review failure releases the rejecting lane. If any archived rows are
            // waiting in ResolveWait due to that blocker, retry them without blocking the scheduler.
            self.dispatch_deferred_base_lane_waiter(merge_result_tx.clone())
                .await;
        } else if let Some(reason) = &workspace_result.rejected {
            info!(
                "Change '{}' rejected after acceptance blocker: {}",
                workspace_result.change_id, reason
            );
            send_event(
                &self.event_tx,
                ParallelEvent::ChangeRejected {
                    change_id: workspace_result.change_id.clone(),
                    reason: reason.clone(),
                },
            )
            .await;
            // Rejected flow is terminal after base-side REJECTED marker commit
            // and should not proceed to merge. Ensure preserved workspace is cleaned up.
            if let Err(e) = self
                .workspace_manager
                .cleanup_workspace(&workspace_result.workspace_name)
                .await
            {
                error!(
                    "Failed to cleanup rejected workspace '{}' for change '{}': {}",
                    workspace_result.workspace_name, workspace_result.change_id, e
                );
            }

            // Rejection review completion (confirm) also releases the rejecting lane;
            // retry deferred merges so ResolveWait rows are not stranded.
            self.dispatch_deferred_base_lane_waiter(merge_result_tx.clone())
                .await;
        } else {
            info!(
                "Change '{}' completed successfully",
                workspace_result.change_id
            );

            // Run merge+cleanup in background and report result back to scheduler loop.
            if workspace_result.final_revision.is_some() {
                self.spawn_merge_task(workspace_result, merge_result_tx.clone());
            }
        }
    }

    fn spawn_merge_task(
        &self,
        workspace_result: WorkspaceResult,
        merge_result_tx: mpsc::Sender<MergeResult>,
    ) {
        let mut merge_executor = ParallelExecutor::new(
            self.repo_root.clone(),
            self.config.clone(),
            self.event_tx.clone(),
        );
        merge_executor.max_conflict_retries = self.max_conflict_retries;
        merge_executor.shared_stagger_state = self.shared_stagger_state.clone();
        merge_executor.auto_resolve_count = self.auto_resolve_count.clone();
        merge_executor.pending_merge_count = self.pending_merge_count.clone();
        merge_executor.cancel_token = self.cancel_token.clone();
        merge_executor.manual_resolve_count = self.manual_resolve_count.clone();
        merge_executor.hooks = self.hooks.clone();

        self.pending_merge_count.fetch_add(1, Ordering::Relaxed);

        tokio::spawn(async move {
            let change_id = workspace_result.change_id.clone();
            let workspace_name = workspace_result.workspace_name.clone();
            let Some(_active_merge_guard) =
                super::merge::ActivePostArchiveMergeGuard::acquire(change_id.clone())
            else {
                info!(
                    change_id = %change_id,
                    workspace = %workspace_name,
                    "Skipping duplicate post-archive merge task because the same change is already active"
                );
                if let Err(send_error) = merge_result_tx
                    .send(MergeResult {
                        change_id,
                        workspace_name,
                        origin: MergeResultOrigin::PostArchiveMerge,
                        outcome: MergeTaskOutcome::deferred(
                            "duplicate post-archive merge task suppressed because the same change is already active",
                            true,
                        ),
                    })
                    .await
                {
                    warn!(
                        "Failed to send duplicate merge suppression result to scheduler loop: {}",
                        send_error
                    );
                }
                return;
            };
            let outcome = merge_executor
                .handle_merge_and_cleanup(workspace_result)
                .await;

            if let Err(send_error) = merge_result_tx
                .send(MergeResult {
                    change_id,
                    workspace_name,
                    origin: MergeResultOrigin::PostArchiveMerge,
                    outcome,
                })
                .await
            {
                warn!(
                    "Failed to send merge result to scheduler loop: {}",
                    send_error
                );
            }
        });
    }

    async fn release_retry_lane_after_non_terminal_outcome(
        &mut self,
        change_id: &str,
        origin: MergeResultOrigin,
    ) {
        let wait_state = match origin {
            MergeResultOrigin::ResolveWaitRetry => WaitState::ResolveWait,
            MergeResultOrigin::RejectWaitRetry => WaitState::RejectWait,
            MergeResultOrigin::PostArchiveMerge => return,
        };

        if let Some(shared) = &self.shared_orchestrator_state {
            let released = {
                let mut guard = shared.write().await;
                guard.release_base_mutating_lane_after_retry(change_id, wait_state)
            };
            if released {
                info!(
                    change_id = %change_id,
                    origin = ?origin,
                    "Released base-mutating lane after non-terminal spawned retry outcome"
                );
            } else {
                debug!(
                    change_id = %change_id,
                    origin = ?origin,
                    "No base-mutating lane release needed after non-terminal spawned retry outcome"
                );
            }
        }
        self.sync_resolve_wait_from_shared_state().await;
    }

    #[allow(dead_code)]
    pub(super) async fn handle_merge_result(
        &mut self,
        merge_result: MergeResult,
    ) -> MergeResultDisposition {
        let (merge_result_tx, _merge_result_rx) = mpsc::channel(1);
        self.handle_merge_result_with_tx(merge_result, &merge_result_tx)
            .await
    }

    /// Apply one typed background base-lane result and return its scheduler disposition.
    ///
    /// Ownership release comes first and is independent of severity: the pending
    /// counter and, for a spawned retry, the base-mutating lane are returned to
    /// the scheduler before the disposition is decided, so a failure can never
    /// strand the lane it borrowed.
    ///
    /// This is also the single global-Error owner for the background base-lane
    /// boundary. Only [`MergeTaskOutcome::RunFatal`] emits one, exactly once,
    /// and it always comes with `AbortRun`; every change-local outcome already
    /// has an authoritative change-scoped owner upstream and must not be
    /// promoted here.
    pub(super) async fn handle_merge_result_with_tx(
        &mut self,
        merge_result: MergeResult,
        merge_result_tx: &mpsc::Sender<MergeResult>,
    ) -> MergeResultDisposition {
        self.pending_merge_count.fetch_sub(1, Ordering::Relaxed);

        let disposition = merge_result.outcome.disposition();

        // Lifecycle-slot settlement, decided by the same outcome the disposition
        // is: a completed merge ends the admitted lifecycle and releases the
        // slot once, while every non-merged outcome leaves the change admitted —
        // deferred for scheduler-owned retry, exhausted into `merge wait`, or
        // waiting for an operator. Retaining there is the backpressure that
        // stops a newer, more divergent worktree from taking the slot of work
        // that has not settled.
        //
        // A run-fatal outcome is the exception: the loop is about to drain and
        // fail, and its exit clears every slot this run owned.
        if self.lifecycle_slots.apply_transition(
            &merge_result.change_id,
            merge_result_transition(
                disposition.is_merged(),
                matches!(merge_result.outcome, MergeTaskOutcome::RunFatal { .. }),
            ),
        ) {
            info!(
                change_id = %merge_result.change_id,
                "Released the lifecycle slot after merge settlement"
            );
        }

        // Lane release for spawned retries happens for every non-merged outcome,
        // ahead of the disposition-specific handling below. The merged path
        // instead promotes the next waiter, which is its own release path.
        if !disposition.is_merged() {
            let releases_retry_lane = match &merge_result.outcome {
                MergeTaskOutcome::Deferred { auto_resumable, .. } => *auto_resumable,
                _ => true,
            };
            if releases_retry_lane {
                self.release_retry_lane_after_non_terminal_outcome(
                    &merge_result.change_id,
                    merge_result.origin,
                )
                .await;
            }
        }

        match &merge_result.outcome {
            MergeTaskOutcome::Merged => {
                info!(
                    origin = ?merge_result.origin,
                    "Background base-lane task completed successfully for '{}'",
                    merge_result.change_id
                );
                self.dispatch_deferred_base_lane_waiter(merge_result_tx.clone())
                    .await;
            }

            MergeTaskOutcome::Deferred {
                reason,
                auto_resumable,
            } => {
                info!(
                    "Background merge task deferred for '{}' (workspace '{}', auto_resumable={}): {}",
                    merge_result.change_id, merge_result.workspace_name, auto_resumable, reason
                );
            }

            MergeTaskOutcome::ResolveExhausted {
                change_id,
                attempts,
                classification,
                detail,
            } => {
                // `ResolveFailed` already carried this to the reducer with the
                // change ID attached, and the worktree is preserved for explicit
                // retry. Wrapping it in a global Error would erase both facts.
                warn!(
                    change_id = %change_id,
                    workspace = %merge_result.workspace_name,
                    origin = ?merge_result.origin,
                    attempts = attempts,
                    classification = classification.token(),
                    "Bounded post-archive resolve exhausted; change remains in merge wait for explicit retry: {}",
                    detail
                );
            }

            MergeTaskOutcome::RecoverableAlreadyReported {
                change_id,
                kind,
                detail,
            } => {
                warn!(
                    change_id = %change_id,
                    workspace = %merge_result.workspace_name,
                    origin = ?merge_result.origin,
                    owner = kind.token(),
                    "Background base-lane failure already reported by its typed owner: {}",
                    detail
                );
            }

            MergeTaskOutcome::RunFatal { detail } => {
                error!(
                    change_id = %merge_result.change_id,
                    workspace = %merge_result.workspace_name,
                    origin = ?merge_result.origin,
                    "Background base-lane task failed fatally: {}",
                    detail
                );
                let message = format!(
                    "Background merge failed for '{}' (workspace '{}'): {}",
                    merge_result.change_id, merge_result.workspace_name, detail
                );
                // One global Error per aborting run: a later fatal result that
                // arrives during the bounded drain must not add a second.
                if self.run_fatal_abort.is_none() {
                    self.run_fatal_abort = Some(message.clone());
                    send_event(&self.event_tx, ParallelEvent::Error { message }).await;
                }
            }
        }

        if disposition == MergeResultDisposition::ContinueWithErrors {
            let failed_change = merge_result
                .outcome
                .scoped_change_id()
                .unwrap_or(merge_result.change_id.as_str())
                .to_string();
            self.change_failures_this_run.insert(failed_change);
        }

        disposition
    }

    async fn dispatch_deferred_base_lane_waiter(
        &mut self,
        merge_result_tx: mpsc::Sender<MergeResult>,
    ) {
        let Some(shared) = &self.shared_orchestrator_state else {
            // Without reducer-owned lane state there is no safe single-flight signal for a
            // detached retry. Legacy callers can still use retry_deferred_base_lane_waiters
            // in tests that do not wire the shared scheduler state.
            self.retry_deferred_merges().await;
            self.last_dispatched_resolve_wait_changes = self.resolve_wait_changes.clone();
            self.last_dispatched_reject_wait_changes = self.reject_wait_changes.clone();
            self.resolve_wait_retry_triggered = false;
            return;
        };

        let promoted = {
            let mut guard = shared.write().await;
            guard.promote_next_base_mutating_lane_waiter()
        };

        let Some((change_id, wait_state)) = promoted else {
            self.sync_resolve_wait_from_shared_state().await;
            self.last_dispatched_resolve_wait_changes = self.resolve_wait_changes.clone();
            self.last_dispatched_reject_wait_changes = self.reject_wait_changes.clone();
            self.resolve_wait_retry_triggered = false;
            return;
        };

        match wait_state {
            crate::orchestration::state::WaitState::ResolveWait => {
                self.resolve_wait_changes.insert(change_id.clone());
                let authorization = self.manual_continuation_for(&change_id).await;
                self.spawn_base_lane_retry_task(
                    change_id,
                    MergeResultOrigin::ResolveWaitRetry,
                    merge_result_tx,
                    authorization,
                );
            }
            crate::orchestration::state::WaitState::RejectWait => {
                self.reject_wait_changes.insert(change_id.clone());
                self.spawn_base_lane_retry_task(
                    change_id,
                    MergeResultOrigin::RejectWaitRetry,
                    merge_result_tx,
                    None,
                );
            }
            other => {
                warn!(
                    "Ignoring unsupported base-mutating lane promotion for '{}' with wait state {:?}",
                    change_id, other
                );
            }
        }

        self.last_dispatched_resolve_wait_changes = self.resolve_wait_changes.clone();
        self.last_dispatched_reject_wait_changes = self.reject_wait_changes.clone();
        self.resolve_wait_retry_triggered = false;
    }

    /// Build the one-dispatch continuation authorization for `change_id`, when
    /// the reducer records an explicit operator resolve intent for it.
    ///
    /// The reducer permission is read, not taken: only the dispatch that owns
    /// the base lane may consume it.
    async fn manual_continuation_for(
        &self,
        change_id: &str,
    ) -> Option<Arc<ManualContinuationAuthorization>> {
        let shared = self.shared_orchestrator_state.as_ref()?;
        shared
            .read()
            .await
            .has_manual_resolve_retry(change_id)
            .then(|| Arc::new(ManualContinuationAuthorization::new(change_id)))
    }

    fn spawn_base_lane_retry_task(
        &self,
        change_id: String,
        origin: MergeResultOrigin,
        merge_result_tx: mpsc::Sender<MergeResult>,
        manual_continuation: Option<Arc<ManualContinuationAuthorization>>,
    ) {
        let mut retry_executor = ParallelExecutor::new(
            self.repo_root.clone(),
            self.config.clone(),
            self.event_tx.clone(),
        );
        retry_executor.max_conflict_retries = self.max_conflict_retries;
        retry_executor.shared_stagger_state = self.shared_stagger_state.clone();
        retry_executor.auto_resolve_count = self.auto_resolve_count.clone();
        retry_executor.pending_merge_count = self.pending_merge_count.clone();
        retry_executor.cancel_token = self.cancel_token.clone();
        retry_executor.manual_resolve_count = self.manual_resolve_count.clone();
        retry_executor.hooks = self.hooks.clone();
        retry_executor.shared_orchestrator_state = self.shared_orchestrator_state.clone();

        // Scheduler-owned base-lane work is admitted work too: a resolve or
        // rejection review starting here ends the persistent-idle episode, so
        // the next park emits a fresh idle edge.
        self.rearm_persistent_idle();

        self.pending_merge_count.fetch_add(1, Ordering::Relaxed);
        tokio::spawn(async move {
            let mut workspace_name = change_id.clone();
            let outcome = async {
                match origin {
                    MergeResultOrigin::ResolveWaitRetry => {
                        if let Ok(Some(workspace_info)) = retry_executor
                            .workspace_manager
                            .find_existing_workspace(&change_id)
                            .await
                        {
                            workspace_name = workspace_info.workspace_name;
                        }
                        retry_executor
                            .retry_deferred_merges_for(vec![change_id.clone()], manual_continuation)
                            .await
                    }
                    MergeResultOrigin::RejectWaitRetry => {
                        if let Ok(Some(workspace_info)) = retry_executor
                            .workspace_manager
                            .find_existing_workspace(&change_id)
                            .await
                        {
                            workspace_name = workspace_info.workspace_name;
                        }
                        retry_executor
                            .retry_deferred_rejection_review_for(change_id.clone())
                            .await
                    }
                    MergeResultOrigin::PostArchiveMerge => unreachable!(
                        "post-archive merge is dispatched by spawn_merge_task, not retry dispatcher"
                    ),
                }
            }
            .await;

            if let Err(send_error) = merge_result_tx
                .send(MergeResult {
                    change_id,
                    workspace_name,
                    origin,
                    outcome,
                })
                .await
            {
                warn!(
                    "Failed to send base-lane retry result to scheduler loop: {}",
                    send_error
                );
            }
        });
    }

    /// Retry one pending base-mutating lane operation according to reducer-owned ordering.
    #[allow(dead_code)]
    pub(super) async fn retry_deferred_base_lane_waiters(&mut self) {
        let Some(shared) = &self.shared_orchestrator_state else {
            self.retry_deferred_merges().await;
            return;
        };

        let promoted = {
            let mut guard = shared.write().await;
            guard.promote_next_base_mutating_lane_waiter()
        };

        match promoted {
            Some((change_id, crate::orchestration::state::WaitState::ResolveWait)) => {
                self.resolve_wait_changes.insert(change_id.clone());
                let authorization = self.manual_continuation_for(&change_id).await;
                let _ = self
                    .retry_deferred_merges_for(vec![change_id], authorization)
                    .await;
            }
            Some((change_id, crate::orchestration::state::WaitState::RejectWait)) => {
                self.reject_wait_changes.insert(change_id.clone());
                let _ = self.retry_deferred_rejection_review_for(change_id).await;
            }
            Some((change_id, wait_state)) => {
                warn!(
                    "Ignoring unsupported base-mutating lane promotion for '{}' with wait state {:?}",
                    change_id, wait_state
                );
            }
            None => {}
        }

        if let Some(shared) = &self.shared_orchestrator_state {
            let guard = shared.read().await;
            self.resolve_wait_changes = guard.resolve_wait_change_ids().into_iter().collect();
            self.reject_wait_changes = guard.reject_wait_change_ids().into_iter().collect();
        }
    }

    /// Retry merge for all ResolveWait changes (auto-resumable deferred merges).
    ///
    /// Called after a merge or resolve succeeds, since the previously blocking
    /// condition (dirty base, another merge in progress) may now be resolved.
    /// For each change in `resolve_wait_changes`:
    /// - If merge now succeeds → send `MergeCompleted`, run hook, cleanup workspace.
    /// - If still deferred and auto-resumable → keep in ResolveWait.
    /// - If still deferred and manual action required → move to MergeWait.
    /// - On error → log and keep in ResolveWait for the next retry opportunity.
    pub(super) async fn retry_deferred_merges(&mut self) {
        if let Some(shared) = &self.shared_orchestrator_state {
            let guard = shared.read().await;
            self.resolve_wait_changes = guard.resolve_wait_change_ids().into_iter().collect();
        }

        if self.resolve_wait_changes.is_empty() {
            return;
        }

        let deferred: Vec<String> = self.resolve_wait_changes.iter().cloned().collect();
        let _ = self.retry_deferred_merges_for(deferred, None).await;
    }

    pub(super) fn stale_retry_reason(workspace_info: &crate::vcs::WorkspaceInfo) -> Option<String> {
        if workspace_info.path.exists() {
            return None;
        }

        Some(format!(
            "workspace path '{}' no longer exists",
            workspace_info.path.display()
        ))
    }

    pub(super) async fn retry_deferred_merges_for(
        &mut self,
        deferred: Vec<String>,
        manual_continuation: Option<Arc<ManualContinuationAuthorization>>,
    ) -> MergeTaskOutcome {
        let mut outcome = MergeTaskOutcome::Merged;
        for change_id in deferred.into_iter().take(1) {
            // A change with durable publication-required evidence is already in
            // cumulative base but is *not* done: retry must resume publication
            // rather than be discarded as a stale already-merged retry.
            let owes_publication = self.has_pending_publication_for(&change_id).await;
            // Read base-tree evidence once per retry: every stale branch below
            // settles from this same observation.
            let integration_evidence = if owes_publication {
                // Durable publication is still owed, so this change is in
                // cumulative base but not done. Base integration is not the
                // question here, and it must never settle as merged.
                StaleResolveEvidence::Absent
            } else {
                self.classify_base_integration_evidence(&change_id).await
            };
            // Only proof settles the retry as already done. Absent and unreadable
            // evidence both stay retryable: the retry proceeds, and if it cannot
            // proceed the stale branches below settle it to manual `merge wait`
            // carrying this same unproven evidence.
            if matches!(integration_evidence, StaleResolveEvidence::Proven) {
                info!(
                    "Skipping stale deferred merge retry for '{}' because it is already merged to base",
                    change_id
                );
                self.settle_stale_resolve_retry_for(&change_id, integration_evidence)
                    .await;
                continue;
            }

            send_event(
                &self.event_tx,
                ParallelEvent::Log(LogEntry::info(format!(
                    "ResolveWait retry dispatch started for '{}'",
                    change_id
                ))),
            )
            .await;

            // Locate the preserved workspace for this change.
            let workspace_info = match self
                .workspace_manager
                .find_existing_workspace(&change_id)
                .await
            {
                Ok(Some(ws)) => ws,
                Ok(None) => {
                    // Stale bookkeeping, not a failure: there is no workspace
                    // left to retry, so the intent is cleared and the run
                    // continues. A global Error here would claim the run is
                    // invalid when nothing is.
                    let message = format!(
                        "No workspace found for ResolveWait retry '{}', clearing stale retry intent",
                        change_id
                    );
                    warn!("{}", message);
                    send_event(
                        &self.event_tx,
                        ParallelEvent::Log(LogEntry::warn(&message).with_change_id(&change_id)),
                    )
                    .await;
                    // The consumed reservation goes away because there is no
                    // workspace left to retry, but base integration was not
                    // proven: the row stays retryable `merge wait` rather than
                    // dropping to idle `not queued`.
                    self.settle_stale_resolve_retry_for(&change_id, integration_evidence)
                        .await;
                    outcome = MergeTaskOutcome::Merged;
                    continue;
                }
                Err(e) => {
                    // A repository query failed before any change-scoped
                    // transition could be established, so scope is unknown and
                    // the outcome fails closed to the single global owner.
                    let message = format!(
                        "Failed to find workspace for ResolveWait retry '{}': {}",
                        change_id, e
                    );
                    warn!("{}", message);
                    outcome = MergeTaskOutcome::run_fatal(message);
                    continue;
                }
            };

            if let Some(stale_reason) = Self::stale_retry_reason(&workspace_info) {
                warn!(
                    change_id = %change_id,
                    workspace = %workspace_info.workspace_name,
                    workspace_path = %workspace_info.path.display(),
                    stale_reason = %stale_reason,
                    "Deferred merge retry workspace path is stale; keeping manual merge wait"
                );
                self.settle_stale_resolve_retry_for(&change_id, integration_evidence)
                    .await;
                outcome = MergeTaskOutcome::Merged;
                continue;
            }

            info!(
                "Retrying deferred merge for '{}' (workspace: {})",
                change_id, workspace_info.workspace_name
            );

            let revisions = vec![workspace_info.workspace_name.clone()];
            let change_ids = vec![change_id.clone()];
            let archive_paths = vec![workspace_info.path.clone()];

            let attempt = self
                .attempt_merge(
                    &revisions,
                    &change_ids,
                    &archive_paths,
                    manual_continuation.as_deref(),
                )
                .await;

            // Clearing follows the dispatch that actually consumed the
            // authorization, never the promotion that merely carried it: a
            // retry deferred by base-lane occupancy consumed nothing and must
            // keep its permission for the auto-resumed attempt.
            if manual_continuation
                .as_ref()
                .is_some_and(|authorization| authorization.is_consumed())
            {
                if let Some(shared) = &self.shared_orchestrator_state {
                    shared
                        .write()
                        .await
                        .consume_manual_resolve_retry(&change_id);
                }
            }

            match attempt {
                Ok(super::merge::MergeAttempt::Merged { revision }) => {
                    info!("Deferred merge succeeded for '{}' on retry", change_id);

                    // With upstream publication enabled the base-lane sequence
                    // already ran `on_merged` and emitted the change-scoped
                    // publication events; local merge is not terminal there.
                    let publication_owned_completion = self.upstream_enabled();
                    let retry_hooks = if publication_owned_completion {
                        None
                    } else {
                        self.hooks.as_ref()
                    };

                    // Run on_merged hook before merged status transition (MergeCompleted event).
                    if let Some(hooks) = retry_hooks {
                        let (completed_tasks, total_tasks) =
                            match crate::openspec::list_changes_native() {
                                Ok(changes) => changes
                                    .iter()
                                    .find(|c| c.id == change_id)
                                    .map(|c| (c.completed_tasks, c.total_tasks))
                                    .unwrap_or((0, 0)),
                                Err(e) => {
                                    warn!("Failed to fetch task counts for on_merged hook: {}", e);
                                    (0, 0)
                                }
                            };
                        let ws_path = workspace_info.path.to_string_lossy().to_string();
                        let hook_ctx = crate::hooks::HookContext::new(0, 0, 0, false)
                            .with_change(&change_id, completed_tasks, total_tasks)
                            .with_apply_count(0)
                            .with_parallel_context(&ws_path, None);
                        if let Err(e) = hooks
                            .run_hook(crate::hooks::HookType::OnMerged, &hook_ctx)
                            .await
                        {
                            let message = on_merged_failure_message(&change_id, &e);
                            error!("{}", message);
                            self.clear_resolve_wait_intent_for_outcome(&change_id).await;
                            // No producer-side pre-application: the authoritative
                            // dispatcher applies this `HookFailed` itself, and only
                            // there can it observe the merge-wait recovery *edge*
                            // that revokes the change's execution mark. Reducing it
                            // here first would make the dispatched copy a no-op and
                            // leave the stale mark behind.
                            send_event(
                                &self.event_tx,
                                ParallelEvent::HookFailed {
                                    change_id: change_id.clone(),
                                    hook_type: crate::hooks::HookType::OnMerged.to_string(),
                                    error: e.to_string(),
                                },
                            )
                            .await;
                            send_event(
                                &self.event_tx,
                                ParallelEvent::ResolveFailed {
                                    change_id: change_id.clone(),
                                    error: message.clone(),
                                },
                            )
                            .await;
                            // `HookFailed` owns this transition; the queue must
                            // not promote it into a run-fatal outcome.
                            outcome = MergeTaskOutcome::already_reported(
                                &change_id,
                                AlreadyReportedFailureKind::Hook,
                                message,
                            );
                            continue;
                        }
                    }

                    self.clear_resolve_wait_intent_for_outcome(&change_id).await;
                    if !publication_owned_completion {
                        self.mark_deferred_merge_completed_in_shared_state(&change_id, &revision)
                            .await;

                        // Send MergeCompleted after on_merged hook (triggers merged status transition)
                        send_event(
                            &self.event_tx,
                            ParallelEvent::MergeCompleted {
                                change_id: change_id.clone(),
                                revision: revision.clone(),
                            },
                        )
                        .await;
                    }

                    // Clean up workspace.
                    send_event(
                        &self.event_tx,
                        ParallelEvent::CleanupStarted {
                            workspace: workspace_info.workspace_name.clone(),
                        },
                    )
                    .await;
                    if let Err(e) = self
                        .workspace_manager
                        .cleanup_workspace(&workspace_info.workspace_name)
                        .await
                    {
                        warn!(
                            "Failed to cleanup workspace '{}' after deferred merge retry: {}",
                            workspace_info.workspace_name, e
                        );
                    } else {
                        send_event(
                            &self.event_tx,
                            ParallelEvent::CleanupCompleted {
                                workspace: workspace_info.workspace_name.clone(),
                            },
                        )
                        .await;
                    }
                    outcome = MergeTaskOutcome::Merged;
                }
                Ok(super::merge::MergeAttempt::Deferred(deferred)) => {
                    info!(
                        "Deferred merge still blocked for '{}': {} (auto_resumable={})",
                        change_id, deferred.reason, deferred.auto_resumable
                    );
                    if deferred.auto_resumable {
                        self.merge_wait_changes.remove(&change_id);
                    } else {
                        self.resolve_wait_changes.remove(&change_id);
                        self.merge_wait_changes.insert(change_id.clone());
                    }
                    let reason = deferred.reason;
                    let auto_resumable = deferred.auto_resumable;
                    send_event(
                        &self.event_tx,
                        ParallelEvent::MergeDeferred {
                            change_id: change_id.clone(),
                            reason: reason.clone(),
                            auto_resumable,
                        },
                    )
                    .await;
                    outcome = MergeTaskOutcome::deferred(reason, auto_resumable);
                }
                Err(failure) => {
                    error!(
                        change_id = %change_id,
                        outcome = ?failure,
                        "Deferred merge retry failed"
                    );
                    // Keep in deferred set; another merge/resolve completion will
                    // trigger again. The typed failure already knows its scope.
                    outcome = failure.into_outcome(&change_id);
                }
            }
        }
        outcome
    }

    pub(super) async fn retry_deferred_rejection_review_for(
        &mut self,
        change_id: String,
    ) -> MergeTaskOutcome {
        let mut outcome = MergeTaskOutcome::Merged;
        send_event(
            &self.event_tx,
            ParallelEvent::Log(LogEntry::info(format!(
                "RejectWait retry dispatch started for '{}'",
                change_id
            ))),
        )
        .await;

        let workspace_info = match self
            .workspace_manager
            .find_existing_workspace(&change_id)
            .await
        {
            Ok(Some(ws)) => ws,
            Ok(None) => {
                // Stale bookkeeping, not a failure: nothing is left to review.
                let message = format!(
                    "No workspace found for RejectWait retry '{}', clearing reject wait",
                    change_id
                );
                warn!("{}", message);
                send_event(
                    &self.event_tx,
                    ParallelEvent::Log(LogEntry::warn(&message).with_change_id(&change_id)),
                )
                .await;
                self.clear_reject_wait_intent_for_success(&change_id).await;
                self.abandon_base_mutating_lane_occupant_for_give_up(&change_id)
                    .await;
                return MergeTaskOutcome::Merged;
            }
            Err(e) => {
                // Repository query failure before any change-scoped transition:
                // scope is unknown, so it fails closed.
                let message = format!(
                    "Failed to find workspace for RejectWait retry '{}': {}",
                    change_id, e
                );
                warn!("{}", message);
                return MergeTaskOutcome::run_fatal(message);
            }
        };

        send_event(
            &self.event_tx,
            ParallelEvent::WorkspaceStatusUpdated {
                change_id: change_id.clone(),
                workspace_name: workspace_info.workspace_name.clone(),
                status: crate::vcs::WorkspaceStatus::Rejecting,
            },
        )
        .await;
        self.clear_reject_wait_intent_for_success(&change_id).await;

        match run_rejection_review(
            &change_id,
            &workspace_info.path,
            &self.config,
            &self.ai_runner,
        )
        .await
        {
            Ok(RejectionReviewVerdict::Confirm) => {
                let rejected_path = workspace_info
                    .path
                    .join("openspec")
                    .join("changes")
                    .join(&change_id)
                    .join("REJECTED.md");
                let reason = format!(
                    "Rejecting review confirmed rejection (proposal: {})",
                    rejected_path.display()
                );
                // Base identity is repository truth, not a guess. The rejection
                // flow checks out this branch and commits REJECTED.md on it, so
                // substituting a literal default here would mutate whatever
                // branch happens to carry that name. Losing base identity fails
                // closed and the queue boundary owns the single global Error.
                let base_branch = match self
                    .workspace_manager
                    .ensure_original_branch_initialized()
                    .await
                {
                    Ok(branch) => branch,
                    Err(error) => {
                        let message = format!(
                            "Failed to resolve base branch while confirming deferred rejection review for '{}': {}",
                            change_id, error
                        );
                        error!("{}", message);
                        return MergeTaskOutcome::run_fatal(message);
                    }
                };

                match execute_rejection_flow(
                    &change_id,
                    &reason,
                    &workspace_info.path,
                    &base_branch,
                    &self.repo_root,
                )
                .await
                {
                    Ok(()) => {
                        let completed_event = ParallelEvent::RejectionReviewCompleted {
                            change_id: change_id.clone(),
                            outcome: crate::events::RejectionOutcome::Confirm,
                        };
                        self.apply_rejection_review_event_in_shared_state(&completed_event)
                            .await;
                        send_event(&self.event_tx, completed_event).await;
                        send_event(
                            &self.event_tx,
                            ParallelEvent::ChangeRejected {
                                change_id: change_id.clone(),
                                reason,
                            },
                        )
                        .await;
                        send_event(
                            &self.event_tx,
                            ParallelEvent::ChangeDequeued {
                                change_id: change_id.clone(),
                            },
                        )
                        .await;
                    }
                    Err(error) => {
                        error!(
                            "Deferred rejection review confirm flow failed for '{}': {}",
                            change_id, error
                        );
                        let failed_event = ParallelEvent::RejectionReviewFailed {
                            change_id: change_id.clone(),
                            error: format!(
                                "Rejected flow failed after deferred rejecting CONFIRM verdict: {}",
                                error
                            ),
                        };
                        // `RejectionReviewFailed` is the typed change-scoped
                        // owner for this failure, so the outcome must cross the
                        // shared base-lane boundary as already-reported. Leaving
                        // it as `Merged` would report a failed rejection as a
                        // completed merge and arm success-only follow-up.
                        outcome = MergeTaskOutcome::already_reported(
                            &change_id,
                            AlreadyReportedFailureKind::RejectionReview,
                            error.to_string(),
                        );
                        send_event(&self.event_tx, failed_event).await;
                    }
                }
            }
            Ok(RejectionReviewVerdict::Resume) => {
                match handle_resume_apply_from_rejecting(&change_id, &workspace_info.path).await {
                    Ok(()) => {
                        let completed_event = ParallelEvent::RejectionReviewCompleted {
                            change_id: change_id.clone(),
                            outcome: crate::events::RejectionOutcome::Resume,
                        };
                        self.apply_rejection_review_event_in_shared_state(&completed_event)
                            .await;
                        send_event(&self.event_tx, completed_event).await;
                        send_event(
                            &self.event_tx,
                            ParallelEvent::Log(
                                LogEntry::warn(
                                    "Deferred rejecting review returned RESUME; workspace is ready for apply resume",
                                )
                                .with_change_id(&change_id)
                                .with_operation("rejecting"),
                            ),
                        )
                        .await;
                    }
                    Err(error) => {
                        let failed_event = ParallelEvent::RejectionReviewFailed {
                            change_id: change_id.clone(),
                            error: error.to_string(),
                        };
                        outcome = MergeTaskOutcome::already_reported(
                            &change_id,
                            AlreadyReportedFailureKind::RejectionReview,
                            error.to_string(),
                        );
                        send_event(&self.event_tx, failed_event).await;
                    }
                }
            }
            Ok(RejectionReviewVerdict::Block) => {
                match handle_blocked_from_rejecting(&change_id, &workspace_info.path).await {
                    Ok(()) => {
                        let completed_event = ParallelEvent::RejectionReviewCompleted {
                            change_id: change_id.clone(),
                            outcome: crate::events::RejectionOutcome::Block,
                        };
                        self.apply_rejection_review_event_in_shared_state(&completed_event)
                            .await;
                        send_event(&self.event_tx, completed_event).await;
                        send_event(
                            &self.event_tx,
                            ParallelEvent::WorkspaceStatusUpdated {
                                change_id: change_id.clone(),
                                workspace_name: workspace_info.workspace_name.clone(),
                                status: crate::vcs::WorkspaceStatus::Blocked,
                            },
                        )
                        .await;
                    }
                    Err(error) => {
                        let failed_event = ParallelEvent::RejectionReviewFailed {
                            change_id: change_id.clone(),
                            error: error.to_string(),
                        };
                        outcome = MergeTaskOutcome::already_reported(
                            &change_id,
                            AlreadyReportedFailureKind::RejectionReview,
                            error.to_string(),
                        );
                        send_event(&self.event_tx, failed_event).await;
                    }
                }
            }
            Err(error) => {
                error!(
                    "Deferred rejection review failed for '{}': {}",
                    change_id, error
                );
                let failed_event = ParallelEvent::RejectionReviewFailed {
                    change_id: change_id.clone(),
                    error: format!("Rejecting review failed after deferred handoff: {}", error),
                };
                outcome = MergeTaskOutcome::already_reported(
                    &change_id,
                    AlreadyReportedFailureKind::RejectionReview,
                    error.to_string(),
                );
                send_event(&self.event_tx, failed_event).await;
            }
        }
        outcome
    }

    /// Retry rejection review for all RejectWait changes once the base-mutating lane is free.
    #[allow(dead_code)]
    pub(super) async fn retry_deferred_rejection_reviews(&mut self) {
        let Some(shared) = &self.shared_orchestrator_state else {
            return;
        };

        let (lane_occupied, reject_wait_ids) = {
            let guard = shared.read().await;
            (
                guard.is_base_mutating_lane_occupied(),
                guard.reject_wait_change_ids(),
            )
        };
        self.reject_wait_changes = reject_wait_ids.into_iter().collect();

        if lane_occupied || self.reject_wait_changes.is_empty() {
            return;
        }

        let Some(change_id) = self.reject_wait_changes.iter().min().cloned() else {
            return;
        };

        let _ = self.retry_deferred_rejection_review_for(change_id).await;
    }

    /// Decide whether one dynamic queue hint may become scheduler-local work.
    ///
    /// A dynamic entry is a wake-up hint, never eligibility truth. Every accepted
    /// path into ordinary work records reducer intent *before* the hint is
    /// published (`add_to_queue` applies `AddToQueue`/`RetryError` first, start
    /// queues its resolved targets first), so the reducer is the only authority
    /// here and this check must fail closed:
    ///
    /// - no reducer wired means eligibility is unproven, so the hint is refused
    ///   rather than admitted from the catalog;
    /// - a reducer-unknown ID carries no intent at all;
    /// - `DequeueChange` does not drain the queue, so a stop-and-dequeue can
    ///   leave a revoked ID sitting in it.
    ///
    /// Refusing costs at most one wake-up: `reconcile_queued_candidates_from_shared_state`
    /// runs on the same scheduler pass and re-adds every genuinely reducer-queued
    /// ID from `queued_change_ids()`, so intent that really exists is never lost.
    ///
    /// Contention is deliberately *not* a refusal. Admission reads the
    /// evaluation's awaited [`ReducerWorkSnapshot`], so a reducer writer can
    /// only delay this decision. If the view is incomplete the hint is retained
    /// rather than consumed, because a refused hint plus an equally
    /// evidence-starved reconciliation is exactly how the only wake edge for
    /// queued work went missing.
    fn admit_dynamic_queue_hint(
        &self,
        snapshot: &ReducerWorkSnapshot,
        dynamic_id: &str,
    ) -> DynamicQueueAdmission {
        if !snapshot.reducer_present() {
            return DynamicQueueAdmission::Refuse {
                reason: "reducer_state_absent",
                log: format!(
                    "Ignoring queue entry with no reducer to authorize it: {}",
                    dynamic_id
                ),
            };
        }
        if !snapshot.is_complete() {
            return DynamicQueueAdmission::Retain;
        }
        if snapshot.is_final_terminal_dispatch_stop(dynamic_id) {
            return DynamicQueueAdmission::Refuse {
                reason: "final_terminal",
                log: format!(
                    "Ignoring stale queue entry for final terminal change: {}",
                    dynamic_id
                ),
            };
        }
        if !snapshot.is_ordinary_queue_eligible(dynamic_id) {
            return DynamicQueueAdmission::Refuse {
                reason: "no_current_queue_intent",
                log: format!(
                    "Ignoring stale queue entry without current queue intent: {}",
                    dynamic_id
                ),
            };
        }
        DynamicQueueAdmission::Admit
    }

    /// Check dynamic queue for newly added changes and update queued list.
    ///
    /// # Arguments
    /// * `queued` - Mutable reference to queued changes list
    /// * `in_flight` - Set of currently executing changes
    /// * `reanalysis_reason` - Mutable reference to reanalysis reason tracking
    ///
    /// # Returns
    /// `true` if queue changed and reanalysis should be triggered
    ///
    /// Test-only convenience: the scheduler loop threads its evaluation-scoped
    /// view into [`Self::check_dynamic_queue_and_add_changes_with_snapshot`],
    /// so this awaits an equivalent one for direct callers.
    #[cfg(test)]
    pub(super) async fn check_dynamic_queue_and_add_changes(
        &mut self,
        queued: &mut Vec<crate::openspec::Change>,
        in_flight: &HashSet<String>,
        reanalysis_reason: &mut ReanalysisReason,
    ) -> bool {
        let snapshot = self.capture_reducer_work_snapshot().await;
        self.check_dynamic_queue_and_add_changes_with_snapshot(
            queued,
            in_flight,
            reanalysis_reason,
            &snapshot,
        )
        .await
    }

    /// Ingest dynamic queue hints against an already-captured reducer view.
    pub(super) async fn check_dynamic_queue_and_add_changes_with_snapshot(
        &mut self,
        queued: &mut Vec<crate::openspec::Change>,
        in_flight: &HashSet<String>,
        reanalysis_reason: &mut ReanalysisReason,
        snapshot: &ReducerWorkSnapshot,
    ) -> bool {
        if let Some(queue) = &self.dynamic_queue {
            let queue = queue.clone();
            let mut queue_changed = false;
            while let Some(dynamic_id) = queue.pop().await {
                if !queued.iter().any(|c| c.id == dynamic_id) && !in_flight.contains(&dynamic_id) {
                    // Reducer-owned intent is validated for every hint, before
                    // the catalog is consulted at all.
                    match self.admit_dynamic_queue_hint(snapshot, &dynamic_id) {
                        DynamicQueueAdmission::Admit => {}
                        DynamicQueueAdmission::Refuse { reason, log } => {
                            info!(
                                change_id = %dynamic_id,
                                reason,
                                "Skipping dynamic queue ingestion because reducer-owned queue intent does not admit it"
                            );
                            send_event(&self.event_tx, ParallelEvent::Log(LogEntry::info(log)))
                                .await;
                            continue;
                        }
                        // Incomplete evidence disposes of nothing. The hint goes
                        // back at the head of the queue with its wake edge
                        // intact, and ingestion stops so the rest of the queue
                        // is left for an evaluation that can actually judge it.
                        DynamicQueueAdmission::Retain => {
                            debug!(
                                change_id = %dynamic_id,
                                "Retaining dynamic queue hint because reducer evidence is incomplete"
                            );
                            queue.requeue_front(dynamic_id).await;
                            break;
                        }
                    }

                    match crate::openspec::list_changes_native_from(&self.repo_root) {
                        Ok(all_changes) => {
                            if let Some(new_change) =
                                all_changes.into_iter().find(|c| c.id == dynamic_id)
                            {
                                info!("Dynamically adding change to execution: {}", dynamic_id);
                                send_event(
                                    &self.event_tx,
                                    ParallelEvent::Log(LogEntry::info(format!(
                                        "Dynamically added to parallel execution: {}",
                                        dynamic_id
                                    ))),
                                )
                                .await;
                                queued.push(new_change);
                                queue_changed = true;
                            } else {
                                // A hint is a wake-up edge, and this lookup is
                                // only its *first* observation. The owner and
                                // the repository update can happen in either
                                // order, so consuming the hint here is not the
                                // verdict: reconciliation runs later in this
                                // same pass on the same reducer view and either
                                // re-reads a fresh catalog that now loads the
                                // change or explicitly settles the queued
                                // intent. Nothing is left queued with no work,
                                // no edge, and no typed wait behind it.
                                warn!(
                                    "Dynamically added change '{}' not found in openspec",
                                    dynamic_id
                                );
                                self.emit_queue_reconciliation_diagnostic(
                                    QueueReconciliationDiagnosticLevel::Warn,
                                    &dynamic_id,
                                    "candidate_not_found",
                                )
                                .await;
                            }
                        }
                        Err(e) => {
                            // Unreadable is not absent. The catalog produced no
                            // answer at all, so this hint keeps its wake edge
                            // instead of being spent on a lookup that never
                            // happened, and ingestion stops so the rest of the
                            // queue is judged by a pass that can read the
                            // repository.
                            warn!(
                                "Failed to load dynamically added change '{}': {}",
                                dynamic_id, e
                            );
                            if self.should_emit_queue_reconciliation_diagnostic(
                                &dynamic_id,
                                "candidate_load_failed",
                            ) {
                                send_event(
                                    &self.event_tx,
                                    ParallelEvent::Log(LogEntry::warn(format!(
                                        "Queue reconciliation pending for '{}': candidate_load_failed ({})",
                                        dynamic_id, e
                                    ))),
                                )
                                .await;
                            }
                            queue.requeue_front(dynamic_id).await;
                            break;
                        }
                    }
                } else if in_flight.contains(&dynamic_id) {
                    debug!(
                        "Ignoring dynamic queue entry '{}' because it is already in-flight",
                        dynamic_id
                    );
                } else {
                    debug!(
                        "Ignoring dynamic queue entry '{}' because it is already queued",
                        dynamic_id
                    );
                }
            }

            // Update queue change timestamp if items were added
            if queue_changed {
                let mut last_change = self.last_queue_change_at.lock().await;
                *last_change = Some(std::time::Instant::now());
                *reanalysis_reason = ReanalysisReason::QueueNotification;
                info!("Queue changed, scheduler state now requires re-analysis");
            }
            queue_changed
        } else {
            false
        }
    }

    pub(super) fn should_emit_queue_reconciliation_diagnostic(
        &mut self,
        change_id: &str,
        reason: &str,
    ) -> bool {
        self.diagnostic_dedup
            .should_emit(DiagnosticDeduplicationKey::QueueReconciliation {
                change_id: change_id.to_string(),
                reason: reason.to_string(),
            })
    }

    async fn emit_queue_reconciliation_diagnostic(
        &mut self,
        level: QueueReconciliationDiagnosticLevel,
        change_id: &str,
        reason: &str,
    ) {
        if !self.should_emit_queue_reconciliation_diagnostic(change_id, reason) {
            debug!(
                change_id,
                reason, "Suppressing repeated queue reconciliation diagnostic"
            );
            return;
        }

        self.emit_queue_reconciliation_diagnostic_without_dedupe(level, change_id, reason)
            .await;
    }

    async fn emit_queue_reconciliation_diagnostic_without_dedupe(
        &self,
        level: QueueReconciliationDiagnosticLevel,
        change_id: &str,
        reason: &str,
    ) {
        let message = match level {
            QueueReconciliationDiagnosticLevel::Info => LogEntry::info(format!(
                "Queue reconciliation deferred for '{}': {}",
                change_id, reason
            )),
            QueueReconciliationDiagnosticLevel::Admitted => LogEntry::info(format!(
                "Queue reconciliation admitted '{}': {}",
                change_id, reason
            )),
            QueueReconciliationDiagnosticLevel::Settled => LogEntry::info(format!(
                "Queue reconciliation settled '{}': {}",
                change_id, reason
            )),
            QueueReconciliationDiagnosticLevel::Warn => LogEntry::warn(format!(
                "Queue reconciliation pending for '{}': {}",
                change_id, reason
            )),
        };

        send_event(&self.event_tx, ParallelEvent::Log(message)).await;
    }

    /// Base branch used to prove archived-dirty repair evidence for one
    /// already-eligible change.
    ///
    /// Returns `None` when the base branch cannot be read; the caller then
    /// leaves the change unresolved instead of guessing a resume phase.
    async fn resolve_archived_dirty_repair_base_branch(&self) -> Option<String> {
        match self
            .workspace_manager
            .ensure_original_branch_initialized()
            .await
        {
            Ok(branch) => Some(branch),
            Err(error) => {
                warn!(
                    "Failed to determine base branch during archived dirty queue reconciliation: {}",
                    error
                );
                send_event(
                    &self.event_tx,
                    ParallelEvent::Log(LogEntry::warn(format!(
                        "Queue reconciliation skipped archived-dirty repair evidence: failed_to_determine_base_branch ({})",
                        error
                    ))),
                )
                .await;
                None
            }
        }
    }

    /// Take every pending explicit-retry edge before reconciliation and
    /// classification, and report what this pass may act on.
    ///
    /// An edge is produced only by an accepted, state-changing retry carrying the
    /// retried change ID, and it arrives with the authority its route granted.
    ///
    /// Two different things happen here, and keeping them apart is the whole
    /// point:
    ///
    /// - The **route-scoped releases** are applied once, immediately, and only for
    ///   [`RetryEdgeAuthority::TerminalError`]. Clearing an ephemeral failed
    ///   classification removes a fast dispatch gate and the dependents'
    ///   blocker-notification epoch — never more: whether the dependency is
    ///   actually resolved is still decided by ordinary repository and dependency
    ///   evidence, so a dependent stays blocked while its blocker is queued, in
    ///   flight, or unmerged. The retried change's Apply budget is released here
    ///   too, because that budget is owned by this scheduler boundary rather than
    ///   by the invocation which spent the ceiling. An acceptance-stall edge
    ///   performs neither, and reconciliation, notification, ordinary queue
    ///   additions, and mark settlement produce no edge at all.
    /// - The **analysis-bypass authority** is latched rather than spent. Every
    ///   drained target joins [`Self::pending_retry_bypass`], where it stays until
    ///   an eligible dependency-analysis evaluation actually runs. A pass that
    ///   ends before that evaluation therefore leaves the authority available to
    ///   the next one instead of discarding it.
    ///
    /// The returned consumption separates "this pass drained new edges" from
    /// "bypass authority is armed", because only the former is committed operator
    /// intent this pass observed: a retained latch must not keep rearming the
    /// persistent-idle episode on every later wake.
    pub(super) async fn consume_explicit_retry_edges(&mut self) -> RetryEdgeConsumption {
        let drained = match self.dynamic_queue.clone() {
            Some(queue) => queue.drain_explicit_retries().await,
            None => Vec::new(),
        };

        for edge in &drained {
            let change_id = edge.change_id.as_str();
            self.pending_retry_bypass.insert(change_id.to_string());
            if !edge.authority.releases_terminal_error_state() {
                info!(
                    change_id = %change_id,
                    "Consumed acceptance-stall retry edge before queue reconciliation; it grants analysis bypass only"
                );
                continue;
            }
            let cleared = self.failed_tracker.clear_failed(change_id);
            // A retried change's own blocked fingerprint is dropped too, so a
            // later blocked observation for it is a new transition rather than a
            // suppressed duplicate.
            self.dependency_blocker_fingerprints.remove(change_id);
            let released_attempts = self.apply_budget.attempts(change_id);
            self.apply_budget.reset(change_id);
            info!(
                change_id = %change_id,
                cleared_failed_marker = cleared,
                released_apply_attempts = released_attempts,
                "Consumed terminal-error retry edge before queue reconciliation"
            );
        }

        RetryEdgeConsumption {
            newly_drained: drained.len(),
            bypass_armed: !self.pending_retry_bypass.is_empty(),
        }
    }

    /// Spend the latched analysis-bypass authority once an eligible evaluation
    /// has really run.
    ///
    /// Called from the one place the analyzer is actually invoked, so the
    /// authority is one-shot with respect to *evaluations* rather than to passes:
    /// later timer wakes observing the same analysis input fall back to ordinary
    /// unchanged-input suppression.
    fn spend_pending_retry_bypass(&mut self) {
        if self.pending_retry_bypass.is_empty() {
            return;
        }
        let mut spent: Vec<&str> = self
            .pending_retry_bypass
            .iter()
            .map(String::as_str)
            .collect();
        spent.sort_unstable();
        info!(
            targets = %spent.join(","),
            "Dependency analysis consumed the pending explicit-retry bypass"
        );
        self.pending_retry_bypass.clear();
    }

    /// Targets whose explicit-retry bypass authority is still armed (tests).
    #[cfg(test)]
    pub(super) fn pending_retry_bypass_targets(&self) -> Vec<String> {
        let mut targets: Vec<String> = self.pending_retry_bypass.iter().cloned().collect();
        targets.sort_unstable();
        targets
    }

    /// Test-only convenience: production reconciliation runs through
    /// [`Self::reconcile_queued_candidates_with_snapshot`] on the evaluation's
    /// own view. This awaits an equivalent one.
    #[cfg(test)]
    pub(super) async fn reconcile_queued_candidates_from_shared_state(
        &mut self,
        queued: &mut Vec<crate::openspec::Change>,
        in_flight: &HashSet<String>,
    ) -> QueueReconciliationOutcome {
        let snapshot = self.capture_reducer_work_snapshot().await;
        self.reconcile_queued_candidates_with_snapshot(queued, in_flight, &snapshot)
            .await
    }

    /// Reconcile reducer-visible queue intent from an already-captured view.
    ///
    /// An incomplete view returns an empty outcome, but that is *not* the same
    /// empty outcome contention used to produce: the caller may not read it as
    /// drain or blocked-only evidence, because
    /// [`ReducerWorkSnapshot::is_complete`] independently withholds every
    /// termination and idle decision for the same pass.
    pub(super) async fn reconcile_queued_candidates_with_snapshot(
        &mut self,
        queued: &mut Vec<crate::openspec::Change>,
        in_flight: &HashSet<String>,
        snapshot: &ReducerWorkSnapshot,
    ) -> QueueReconciliationOutcome {
        if !snapshot.reducer_present() || !snapshot.is_complete() {
            return QueueReconciliationOutcome::default();
        }

        // A terminal error is not filtered here: `queued_change_ids()` already
        // excludes every terminal state, so a terminal-error change carries no
        // ordinary queue intent until `RetryError` clears it. The terminal-error
        // stop gate itself stays where it can still be reached, in dispatch
        // selection.
        let queued_intent_ids = snapshot.queued_intent_ids().to_vec();
        let lane_wait_ids: std::collections::HashSet<String> = snapshot
            .resolve_wait_ids()
            .iter()
            .chain(snapshot.reject_wait_ids())
            .cloned()
            .collect();
        let revoked_local_ids: Vec<String> = queued
            .iter()
            .filter(|change| !snapshot.is_ordinary_queue_eligible(&change.id))
            .map(|change| change.id.clone())
            .collect();

        let reducer_active_set = snapshot.active_ids();
        let reducer_merge_wait_set = snapshot.merge_wait_ids();

        let mut outcome = QueueReconciliationOutcome::default();

        // Reconciliation runs in both directions. Revoked work must leave the
        // scheduler-local candidate list, not merely be refused at dispatch:
        // `RemoveFromQueue` and `DequeueChange` clear intent without touching
        // this list, so a candidate admitted on an earlier pass would otherwise
        // keep being analysed and keep an otherwise drained run alive. Anything
        // the reducer or the scheduler still owns as waiting work is kept —
        // merge, resolve, and reject lanes are reducer-owned intent of their
        // own, and an in-flight change is already running. A later explicit
        // `AddToQueue` re-adds the change from `queued_change_ids()` below, so
        // dropping it here loses nothing an operator asked for.
        let revoked_local_ids: Vec<String> = revoked_local_ids
            .into_iter()
            .filter(|change_id| {
                !in_flight.contains(change_id)
                    && !reducer_active_set.contains(change_id)
                    && !reducer_merge_wait_set.contains(change_id)
                    && !lane_wait_ids.contains(change_id)
                    && !self.merge_wait_changes.contains(change_id)
                    && !self.resolve_wait_changes.contains(change_id)
                    && !self.reject_wait_changes.contains(change_id)
            })
            .collect();
        if !revoked_local_ids.is_empty() {
            let revoked: std::collections::HashSet<&String> = revoked_local_ids.iter().collect();
            queued.retain(|change| !revoked.contains(&change.id));
            outcome.revoked_removed = revoked_local_ids.len();
            for change_id in &revoked_local_ids {
                // Revocation ends this candidate's failed-blocker epoch: a later
                // explicit re-add is a genuine queue addition and may announce
                // one new bounded blocker transition.
                self.failed_tracker.clear_blocker_epoch(change_id);
                info!(
                    change_id = %change_id,
                    "Queue reconciliation dropped candidate because reducer queue intent was revoked"
                );
                self.emit_queue_reconciliation_diagnostic(
                    QueueReconciliationDiagnosticLevel::Info,
                    change_id,
                    "queue_intent_revoked",
                )
                .await;
            }
        }

        // Ordinary execution intent is explicit, never discovered.
        //
        // `queued_intent_ids` is the *only* source of ordinary candidates here.
        // It carries every explicit path: TUI and remote Start apply
        // `AddToQueue` for their resolved targets before dispatch, Running-mode
        // queue additions and `RetryError` set the same reducer intent, and CLI
        // explicit targets are classified into the initial candidate vector
        // before this function ever runs. `RemoveFromQueue` and `DequeueChange`
        // drop an ID from this list on the very next pass, which is what makes
        // revocation immediate.
        //
        // Repository-wide worktree enumeration deliberately does not happen:
        // an archived-dirty worktree is recoverable *evidence*, not an operator
        // command, and appending it here turned an unselected interrupted
        // workspace into implicit execution intent. Preserved workspaces are
        // still inspected below, but only for an ID that is already eligible.
        if queued_intent_ids.is_empty() {
            return outcome;
        }

        let mut known_changes = match crate::openspec::list_changes_native_from(&self.repo_root) {
            Ok(changes) => changes,
            Err(e) => {
                warn!(
                    "Failed to load OpenSpec changes during queue reconciliation: {}",
                    e
                );
                send_event(
                    &self.event_tx,
                    ParallelEvent::Log(LogEntry::warn(format!(
                        "Queue reconciliation skipped: failed_to_load_changes ({})",
                        e
                    ))),
                )
                .await;
                return outcome;
            }
        };

        let mut known_by_id: std::collections::HashMap<String, crate::openspec::Change> =
            known_changes
                .drain(..)
                .map(|change| (change.id.clone(), change))
                .collect();

        // Resolved on first use only. Without a repository-wide scan the base
        // branch is needed only when an already-eligible ID is missing from the
        // active catalog, so an ordinary reconciliation pass no longer pays for
        // it at all.
        let mut base_branch_for_archived_dirty_repair: Option<Option<String>> = None;

        for queued_id in queued_intent_ids {
            // Manual merge wait survives explicit queue intent: a resolve failure
            // restores `MergeWait` without clearing it, so an explicitly queued
            // change can sit here waiting for an accepted `ResolveMerge`.
            if reducer_merge_wait_set.contains(&queued_id) {
                self.emit_queue_reconciliation_diagnostic(
                    QueueReconciliationDiagnosticLevel::Info,
                    &queued_id,
                    "manual_merge_wait",
                )
                .await;
                continue;
            }
            if queued.iter().any(|change| change.id == queued_id) {
                continue;
            }
            if in_flight.contains(&queued_id) || reducer_active_set.contains(&queued_id) {
                self.emit_queue_reconciliation_diagnostic(
                    QueueReconciliationDiagnosticLevel::Info,
                    &queued_id,
                    "already_active",
                )
                .await;
                continue;
            }
            if Self::is_post_archive_merge_active_for(&queued_id) {
                self.emit_queue_reconciliation_diagnostic(
                    QueueReconciliationDiagnosticLevel::Info,
                    &queued_id,
                    "post_archive_merge_active",
                )
                .await;
                continue;
            }

            match known_by_id.remove(&queued_id) {
                Some(change) => {
                    info!(
                        "Queue reconciliation adding reducer-queued change candidate: {}",
                        queued_id
                    );
                    queued.push(change);
                    outcome.queued_added += 1;
                }
                None => {
                    if base_branch_for_archived_dirty_repair.is_none() {
                        base_branch_for_archived_dirty_repair =
                            Some(self.resolve_archived_dirty_repair_base_branch().await);
                    }
                    let archived_dirty_evidence = if let Some(base_branch) =
                        base_branch_for_archived_dirty_repair
                            .as_ref()
                            .and_then(|branch| branch.as_ref())
                    {
                        match self
                            .workspace_manager
                            .find_existing_workspace(&queued_id)
                            .await
                        {
                            Ok(Some(workspace)) => {
                                match archived_dirty_repair_candidate_from_workspace(
                                    &queued_id,
                                    &workspace.path,
                                    base_branch,
                                )
                                .await
                                {
                                    Some(change) => {
                                        ArchivedDirtyRepairEvidence::Candidate(Box::new(change))
                                    }
                                    None => ArchivedDirtyRepairEvidence::NotApplicable,
                                }
                            }
                            Ok(None) => ArchivedDirtyRepairEvidence::NotApplicable,
                            Err(error) => {
                                // The workspace question was asked and produced
                                // no answer. "This change has no repairable
                                // workspace" is exactly what was *not* proven,
                                // so the row keeps its intent.
                                warn!(
                                    change_id = %queued_id,
                                    "Failed to find workspace for reducer-queued archived dirty repair candidate: {}",
                                    error
                                );
                                ArchivedDirtyRepairEvidence::Undetermined("workspace_lookup_failed")
                            }
                        }
                    } else {
                        // Base identity is a precondition of the repair probe,
                        // not a repair verdict: without it the probe never ran.
                        ArchivedDirtyRepairEvidence::Undetermined("base_branch_unresolved")
                    };

                    match archived_dirty_evidence {
                        ArchivedDirtyRepairEvidence::Candidate(change) => {
                            info!(
                                "Queue reconciliation adding archived dirty repair candidate: {}",
                                queued_id
                            );
                            self.emit_queue_reconciliation_diagnostic(
                                QueueReconciliationDiagnosticLevel::Info,
                                &queued_id,
                                "archived_dirty_repair_candidate",
                            )
                            .await;
                            queued.push(*change);
                            outcome.repair_added += 1;
                        }
                        ArchivedDirtyRepairEvidence::Undetermined(reason) => {
                            // Settling here would build the ghost-row verdict on
                            // a precondition nobody proved, and because mark
                            // settlement is edge-triggered the revoked intent
                            // would need an explicit operator Start to come
                            // back. Defer exactly as an unreadable catalog does:
                            // queue intent and the wake edge stay untouched, and
                            // a pass that can read the evidence decides.
                            debug!(
                                change_id = %queued_id,
                                reason,
                                "Deferring missing queued candidate whose archived-dirty repair evidence is undetermined"
                            );
                            self.emit_queue_reconciliation_diagnostic(
                                QueueReconciliationDiagnosticLevel::Info,
                                &queued_id,
                                "repair_evidence_unavailable",
                            )
                            .await;
                            outcome.repair_evidence_deferred += 1;
                        }
                        ArchivedDirtyRepairEvidence::NotApplicable => {
                            self.resolve_missing_queued_candidate(
                                &queued_id,
                                queued,
                                &mut outcome,
                                snapshot,
                            )
                            .await;
                        }
                    }
                }
            }
        }

        if outcome.queued_added > 0 {
            let mut last_change = self.last_queue_change_at.lock().await;
            if last_change.is_none() {
                *last_change = Some(std::time::Instant::now());
                debug!(
                    queued_added = outcome.queued_added,
                    "Queue reconciliation initialized reducer-visible queue debounce timestamp"
                );
            } else {
                debug!(
                    queued_added = outcome.queued_added,
                    "Queue reconciliation preserved existing queue debounce timestamp"
                );
            }
        }

        outcome
    }

    /// Re-read the repository-visible active-change catalog for one ID.
    ///
    /// Deliberately a *second* read rather than a reuse of this pass's map. A
    /// long-lived owner can observe the reducer command and the repository
    /// update in either order, so the point of this lookup is that it happens
    /// later in wall-clock time than the one that missed — a proposal committed
    /// or merged into the base in between is visible here, and that is what lets
    /// the same owner admit it without a restart.
    fn refresh_active_change_candidate(&self, change_id: &str) -> FreshCandidateLookup {
        match crate::openspec::list_changes_native_from(&self.repo_root) {
            Ok(changes) => match changes.into_iter().find(|change| change.id == change_id) {
                Some(change) => FreshCandidateLookup::Loadable(Box::new(change)),
                None => FreshCandidateLookup::Absent,
            },
            Err(error) => FreshCandidateLookup::Unreadable(error.to_string()),
        }
    }

    /// Classify one reducer-queued change this pass's catalog view could not load.
    ///
    /// The first lookup is never the verdict. It is re-checked against a fresh
    /// repository-visible view, and only that decides:
    ///
    /// - loadable now — the proposal landed after this pass's first read, so the
    ///   same owner admits it as ordinary scheduler-local work;
    /// - unreadable — absence is unproven, so queue intent is left exactly as it
    ///   was for a pass that can actually read the repository;
    /// - genuinely absent — the queued projection has no scheduler-local work,
    ///   no wake edge, and no typed wait behind it, so it is explicitly
    ///   reconciled out through the reducer instead of being left as a
    ///   permanently queued ghost row.
    ///
    /// Execution marks are a separate axis and are never touched here: a failed
    /// admission is not an operator decision to unselect the proposal.
    ///
    /// Callers must only reach this with *conclusive* archived-dirty repair
    /// evidence ([`ArchivedDirtyRepairEvidence::NotApplicable`]). An
    /// undetermined repair probe is deferred by the caller, because the settle
    /// path below spends accepted queue intent and an edge-triggered mark will
    /// not restore it.
    pub(super) async fn resolve_missing_queued_candidate(
        &mut self,
        queued_id: &str,
        queued: &mut Vec<crate::openspec::Change>,
        outcome: &mut QueueReconciliationOutcome,
        snapshot: &ReducerWorkSnapshot,
    ) {
        match self.refresh_active_change_candidate(queued_id) {
            FreshCandidateLookup::Loadable(change) => {
                info!(
                    change_id = %queued_id,
                    "Queue reconciliation admitted a candidate that a refreshed active-change view made loadable"
                );
                self.emit_queue_reconciliation_diagnostic(
                    QueueReconciliationDiagnosticLevel::Admitted,
                    queued_id,
                    "candidate_refreshed",
                )
                .await;
                // The missing-candidate epoch ends with the miss it described:
                // if this change ever goes missing again, that is a new fact and
                // must be reportable rather than suppressed as a repeat.
                self.clear_missing_candidate_diagnostic_epoch(queued_id);
                queued.push(*change);
                outcome.queued_added += 1;
            }
            FreshCandidateLookup::Unreadable(error) => {
                if self
                    .should_emit_queue_reconciliation_diagnostic(queued_id, "candidate_load_failed")
                {
                    warn!(
                        change_id = %queued_id,
                        "Queue reconciliation could not read the active-change catalog: {}",
                        error
                    );
                    send_event(
                        &self.event_tx,
                        ParallelEvent::Log(LogEntry::warn(format!(
                            "Queue reconciliation pending for '{}': candidate_load_failed ({})",
                            queued_id, error
                        ))),
                    )
                    .await;
                } else {
                    debug!(
                        change_id = %queued_id,
                        reason = "candidate_load_failed",
                        "Suppressing repeated queue reconciliation catalog read failure"
                    );
                }
            }
            FreshCandidateLookup::Absent => {
                if self
                    .should_emit_queue_reconciliation_diagnostic(queued_id, "candidate_not_found")
                {
                    warn!(
                        "Queue reconciliation could not load reducer-queued change '{}': candidate_not_found",
                        queued_id
                    );
                    self.emit_queue_reconciliation_diagnostic_without_dedupe(
                        QueueReconciliationDiagnosticLevel::Warn,
                        queued_id,
                        "candidate_not_found",
                    )
                    .await;
                } else {
                    debug!(
                        change_id = %queued_id,
                        reason = "candidate_not_found",
                        "Suppressing repeated queue reconciliation candidate_not_found warning"
                    );
                }

                if self
                    .reconcile_unavailable_queue_intent(queued_id, snapshot)
                    .await
                {
                    outcome.unavailable_reconciled += 1;
                    info!(
                        change_id = %queued_id,
                        "Queue reconciliation removed queue intent whose candidate a fresh repository view proved unavailable"
                    );
                    self.emit_queue_reconciliation_diagnostic(
                        QueueReconciliationDiagnosticLevel::Settled,
                        queued_id,
                        "candidate_unavailable",
                    )
                    .await;
                }
            }
        }
    }

    /// Forget the bounded missing-candidate diagnostics recorded for one change.
    fn clear_missing_candidate_diagnostic_epoch(&mut self, change_id: &str) {
        self.diagnostic_dedup.reset_matching(|key| {
            matches!(
                key,
                DiagnosticDeduplicationKey::QueueReconciliation {
                    change_id: recorded,
                    reason,
                } if recorded == change_id
                    && (reason == "candidate_not_found" || reason == "candidate_load_failed")
            )
        });
    }

    /// Submit the explicit reducer transition that ends an unavailable queued row.
    ///
    /// Returns whether queue intent actually changed.
    ///
    /// Three properties make this safe to run from the scheduler:
    ///
    /// *It is refused whenever some other state already explains the row.* A
    /// merge, resolve, reject, resolving, held, or externally blocked change is
    /// a typed wait, not a ghost, and stays reducer-owned.
    ///
    /// *It reads the revision it mutates.* The absence check and this transition
    /// are separate observations, so eligibility is re-read under the write
    /// guard rather than trusted from the pass's snapshot. A change that became
    /// active, terminal, or newly re-queued in between is a concurrent decision
    /// this pass may not overwrite, and `RemoveFromQueue` is a single targeted
    /// command rather than a state overwrite.
    ///
    /// *It leaves the change re-admittable.* `RemoveFromQueue` clears queue
    /// intent without setting the dequeued flag and without touching the
    /// execution mark, so the next explicit Start or mark-settlement edge can
    /// admit the proposal once it really exists. Nothing here launches Apply.
    async fn reconcile_unavailable_queue_intent(
        &self,
        change_id: &str,
        snapshot: &ReducerWorkSnapshot,
    ) -> bool {
        if snapshot.merge_wait_ids().contains(change_id)
            || snapshot.resolve_wait_ids().contains(change_id)
            || snapshot.reject_wait_ids().contains(change_id)
            || snapshot.resolving_ids().contains(change_id)
            || snapshot.held_ids().contains(change_id)
            || self.merge_wait_changes.contains(change_id)
            || self.resolve_wait_changes.contains(change_id)
            || self.reject_wait_changes.contains(change_id)
        {
            debug!(
                change_id,
                "Leaving unavailable queued candidate to its typed wait state"
            );
            return false;
        }

        let Some(shared) = &self.shared_orchestrator_state else {
            return false;
        };

        let mut guard = shared.write().await;
        if !guard.queued_change_ids().iter().any(|id| id == change_id) {
            return false;
        }
        if guard.active_change_ids().iter().any(|id| id == change_id) {
            return false;
        }

        matches!(
            guard.apply_command(
                crate::orchestration::state::ReducerCommand::RemoveFromQueue(change_id.to_string(),)
            ),
            crate::orchestration::state::ReduceOutcome::Changed(_)
        )
    }

    /// Test-only convenience: production classification runs through
    /// [`Self::classify_queued_work_with_snapshot`] on the evaluation's own
    /// view. This awaits an equivalent one.
    #[cfg(test)]
    pub(super) async fn classify_queued_work(
        &self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
    ) -> BlockedOnlyQueueClassification {
        let snapshot = self.capture_reducer_work_snapshot().await;
        self.classify_queued_work_with_snapshot(queued, in_flight, &snapshot)
            .await
    }

    /// Classify queued work from an already-captured coherent reducer view.
    ///
    /// Acceptance holds, validated external prerequisite waits, and every wait
    /// lane come from the same view as ordinary eligibility, so no two of these
    /// facts can be read from different reducer revisions. Nothing is loaded
    /// from disk: after a restart the reducer holds neither hold kind, the
    /// change is dispatched again, and workspace evidence routes a complete
    /// unarchived apply revision back to acceptance.
    pub(super) async fn classify_queued_work_with_snapshot(
        &self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        snapshot: &ReducerWorkSnapshot,
    ) -> BlockedOnlyQueueClassification {
        let mut classification = BlockedOnlyQueueClassification::default();
        let mut seen_ids: HashSet<String> = HashSet::new();
        let mut dependency_context = DependencyContext::from_snapshot(
            self.repo_root.clone(),
            queued.iter().map(|change| change.id.as_str()),
            in_flight,
            snapshot,
        );

        let reducer_queued = snapshot.queued_intent_ids().to_vec();
        // An apply-origin external blocker suppresses dispatch exactly like an
        // acceptance-origin one; only the explanation differs.
        let acceptance_stalled = snapshot.held_ids();

        let merge_wait_set: HashSet<String> = snapshot.merge_wait_ids().clone();
        let resolve_wait_set: HashSet<String> = snapshot.resolve_wait_ids().clone();
        let reject_wait_set: HashSet<String> = snapshot.reject_wait_ids().clone();

        for change in queued {
            seen_ids.insert(change.id.clone());
            if acceptance_stalled.contains(&change.id) {
                classification.acceptance_stalled.push(change.id.clone());
                continue;
            }
            if merge_wait_set.contains(&change.id) || self.merge_wait_changes.contains(&change.id) {
                classification.manual_merge_wait.push(change.id.clone());
                continue;
            }
            if resolve_wait_set.contains(&change.id)
                || reject_wait_set.contains(&change.id)
                || self.resolve_wait_changes.contains(&change.id)
                || self.reject_wait_changes.contains(&change.id)
            {
                classification.scheduler_lane_wait.push(change.id.clone());
                continue;
            }
            if dependency_context.is_terminal_error_change(&change.id) {
                classification
                    .terminal_error_retry_required
                    .push(change.id.clone());
                continue;
            }
            // Ordinary dispatch requires *current* reducer intent. Sitting in the
            // scheduler-local candidate list only proves the change was admitted
            // on an earlier pass: `RemoveFromQueue` and `DequeueChange` revoke
            // intent without touching that list, so a candidate added before the
            // revocation would otherwise stay dispatchable for the rest of the
            // run. Reducer-owned lane waits are classified above, so this gate
            // removes ordinary eligibility only, and an explicit `AddToQueue`
            // restores it on the very next pass.
            if dependency_context.withholds_ordinary_queue_intent(&change.id) {
                debug!(
                    change_id = %change.id,
                    "Withholding queued candidate because current reducer intent does not admit ordinary work"
                );
                classification.candidate_unavailable.push(change.id.clone());
                continue;
            }
            if self.failed_tracker.should_skip(&change.id).is_some() {
                classification.dependency_blocked.push(change.id.clone());
                continue;
            }

            let proposal_path = self
                .repo_root
                .join("openspec/changes")
                .join(&change.id)
                .join("proposal.md");
            let metadata_dependencies = if self.repo_root.join("openspec/changes").exists() {
                match crate::openspec::parse_proposal_dependencies_strict_from_file(&proposal_path)
                {
                    Ok(dependencies) => dependencies,
                    Err(error) => {
                        warn!(
                            change_id = %change.id,
                            proposal = %proposal_path.display(),
                            error = %error,
                            "Blocking queue classification because dependency metadata could not be read"
                        );
                        classification.dependency_blocked.push(change.id.clone());
                        continue;
                    }
                }
            } else {
                Vec::new()
            };
            let mut dependencies = change.dependencies.clone();
            for dep_id in metadata_dependencies {
                if dep_id != change.id && !dependencies.contains(&dep_id) {
                    dependencies.push(dep_id);
                }
            }
            let blocked = dependency_context
                .is_blocked(&dependencies, self.workspace_manager.as_ref())
                .await
                .is_some();

            if blocked {
                classification.dependency_blocked.push(change.id.clone());
            } else {
                classification.dispatchable.push(change.clone());
            }
        }

        for queued_id in reducer_queued {
            if !seen_ids.insert(queued_id.clone()) {
                continue;
            }
            if merge_wait_set.contains(&queued_id) || self.merge_wait_changes.contains(&queued_id) {
                classification.manual_merge_wait.push(queued_id);
            } else if resolve_wait_set.contains(&queued_id)
                || reject_wait_set.contains(&queued_id)
                || self.resolve_wait_changes.contains(&queued_id)
                || self.reject_wait_changes.contains(&queued_id)
            {
                classification.scheduler_lane_wait.push(queued_id);
            } else if dependency_context.is_terminal_error_change(&queued_id) {
                classification.terminal_error_retry_required.push(queued_id);
            } else {
                classification.candidate_unavailable.push(queued_id);
            }
        }

        for change_id in merge_wait_set
            .into_iter()
            .chain(self.merge_wait_changes.iter().cloned())
        {
            if seen_ids.insert(change_id.clone()) {
                classification.manual_merge_wait.push(change_id);
            }
        }
        for change_id in resolve_wait_set
            .into_iter()
            .chain(reject_wait_set)
            .chain(self.resolve_wait_changes.iter().cloned())
            .chain(self.reject_wait_changes.iter().cloned())
        {
            if seen_ids.insert(change_id.clone()) {
                classification.scheduler_lane_wait.push(change_id);
            }
        }

        classification
    }

    #[cfg(test)]
    pub(super) async fn is_blocked_only_scheduler_state(
        &self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
    ) -> bool {
        let snapshot = self.capture_reducer_work_snapshot().await;
        self.is_blocked_only_scheduler_state_with_snapshot(queued, in_flight, &snapshot)
            .await
    }

    /// Whether every remaining candidate is stably held, judged from one
    /// coherent reducer view.
    ///
    /// Incomplete evidence is never "stably blocked". Blocked-only is a
    /// scheduler *conclusion* — it ends a finite run as `BlockedOrStalled` and
    /// puts a persistent run into an event-driven idle wait with no timer — so
    /// it requires evidence that actually exists.
    pub(super) async fn is_blocked_only_scheduler_state_with_snapshot(
        &self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        snapshot: &ReducerWorkSnapshot,
    ) -> bool {
        if !snapshot.is_complete() {
            return false;
        }
        if !self.resolve_wait_changes.is_empty() || !self.reject_wait_changes.is_empty() {
            return false;
        }

        in_flight.is_empty()
            && self.manual_resolve_active() == 0
            && self.pending_merge_count.load(Ordering::Relaxed) == 0
            && self
                .classify_queued_work_with_snapshot(queued, in_flight, snapshot)
                .await
                .is_blocked_only()
    }

    async fn emit_log_diagnostic_once(
        &mut self,
        key: DiagnosticDeduplicationKey,
        log_entry: LogEntry,
        suppressed_message: &'static str,
    ) {
        let event_tx = self.event_tx.clone();
        self.diagnostic_dedup
            .emit_or_suppress(
                key,
                || async move {
                    send_event(&event_tx, ParallelEvent::Log(log_entry)).await;
                },
                || debug!("{}", suppressed_message),
            )
            .await;
    }

    #[cfg(test)]
    async fn emit_error_diagnostic_once(
        &mut self,
        key: DiagnosticDeduplicationKey,
        message: String,
        suppressed_message: &'static str,
    ) {
        let event_tx = self.event_tx.clone();
        self.diagnostic_dedup
            .emit_or_suppress(
                key,
                || async move {
                    warn!("{}", message);
                    send_event(&event_tx, ParallelEvent::Log(LogEntry::warn(&message))).await;
                    send_event(&event_tx, ParallelEvent::Error { message }).await;
                },
                || debug!("{}", suppressed_message),
            )
            .await;
    }

    #[cfg(test)]
    fn analysis_failure_diagnostic(
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        error: &str,
    ) -> (DiagnosticDeduplicationKey, String) {
        let mut queued_ids: Vec<String> = queued.iter().map(|change| change.id.clone()).collect();
        queued_ids.sort();
        let mut in_flight_ids: Vec<String> = in_flight.iter().cloned().collect();
        in_flight_ids.sort();
        let normalized_error = error.trim().to_string();
        let key = DiagnosticDeduplicationKey::AnalysisFailure {
            queued_ids: queued_ids.clone(),
            in_flight_ids: in_flight_ids.clone(),
            error: normalized_error.clone(),
        };
        let message = format!(
            "Dependency analysis failed: error={}, queued={:?}, in_flight={:?}",
            normalized_error, queued_ids, in_flight_ids
        );
        (key, message)
    }

    /// Awaited rather than attempted: this diagnostic is the operator's only
    /// notice that reducer-visible queued work produced no analysis, so losing it
    /// to a concurrent reducer write is exactly the silence it exists to prevent.
    async fn no_analysis_diagnostic(
        &self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        reason: &str,
    ) -> Option<(DiagnosticDeduplicationKey, LogEntry)> {
        let reducer_queued = self
            .capture_reducer_work_snapshot()
            .await
            .queued_intent_ids()
            .to_vec();
        if reducer_queued.is_empty() {
            return None;
        }

        let queued_len = queued.len();
        let in_flight_len = in_flight.len();
        let key = DiagnosticDeduplicationKey::NoAnalysis {
            reducer_queued: reducer_queued.clone(),
            queued_len,
            in_flight_len,
            reason: reason.to_string(),
        };
        let log_entry = LogEntry::info(format!(
            "No analysis started despite reducer-visible queued work: reason={}, reducer_queued={:?}, local_queued={}, in_flight={}, max_parallelism={}",
            reason, reducer_queued, queued_len, in_flight_len, max_parallelism
        ));
        Some((key, log_entry))
    }

    fn capacity_zero_dispatch_diagnostic(
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        lifecycle_occupancy: &HashSet<String>,
        max_parallelism: usize,
        analysis_order: &[String],
    ) -> (DiagnosticDeduplicationKey, LogEntry) {
        const REASON: &str = "dispatch_capacity_zero_after_analysis";
        let order = if analysis_order.is_empty() {
            let mut queued_ids = queued
                .iter()
                .map(|change| change.id.clone())
                .collect::<Vec<_>>();
            queued_ids.sort();
            queued_ids
        } else {
            analysis_order.to_vec()
        };
        let queued_len = queued.len();
        let in_flight_len = in_flight.len();
        let mut occupancy: Vec<String> = lifecycle_occupancy.iter().cloned().collect();
        occupancy.sort();
        let key = DiagnosticDeduplicationKey::DispatchCapacityZero {
            order,
            queued_len,
            in_flight_len,
            lifecycle_occupancy: occupancy.clone(),
            max_parallelism,
            reason: REASON.to_string(),
        };
        // The occupancy list is the answer to the operator's actual question at
        // zero capacity: which admitted changes own the slots. A change that is
        // merging or waiting for manual resolution is not in flight, so
        // `in_flight` alone reads as "nothing is running, yet nothing starts".
        let log_entry = LogEntry::info(format!(
            "Dispatch suppressed after dependency analysis: reason={}, local_queued={}, in_flight={}, lifecycle_slots={:?}, max_parallelism={}",
            REASON, queued_len, in_flight_len, occupancy, max_parallelism
        ));
        (key, log_entry)
    }

    #[cfg(test)]
    pub(super) async fn emit_analysis_failure_diagnostic_once(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        error: &str,
    ) {
        let (key, message) = Self::analysis_failure_diagnostic(queued, in_flight, error);
        self.emit_error_diagnostic_once(
            key,
            message,
            "Suppressing repeated analysis failure diagnostic",
        )
        .await;
    }

    pub(super) async fn emit_no_analysis_diagnostic(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        reason: &str,
    ) {
        if let Some((key, log_entry)) = self
            .no_analysis_diagnostic(queued, in_flight, max_parallelism, reason)
            .await
        {
            self.emit_log_diagnostic_once(
                key,
                log_entry,
                "Suppressing repeated no-analysis diagnostic",
            )
            .await;
        }
    }

    pub(super) async fn emit_capacity_zero_dispatch_diagnostic_once(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        analysis_order: &[String],
    ) {
        let occupancy: HashSet<String> = self
            .lifecycle_slots
            .members()
            .union(in_flight)
            .cloned()
            .collect();
        let (key, log_entry) = Self::capacity_zero_dispatch_diagnostic(
            queued,
            in_flight,
            &occupancy,
            max_parallelism,
            analysis_order,
        );
        self.emit_log_diagnostic_once(
            key,
            log_entry,
            "Suppressing repeated capacity-zero dispatch diagnostic",
        )
        .await;
    }

    /// Suppress the analyzer for one zero-capacity evaluation.
    ///
    /// Records nothing. A signature says "this exact input was analysed", and a
    /// suppression record says "this exact input was analysed and produced
    /// nothing new" — neither is true of an input the analyzer never saw. Writing
    /// either here is what would swallow the queue, completion, repair, or
    /// slot-recovery edge that is still waiting to be evaluated, so the flag it
    /// sets instead tells the loop step to leave that edge unconsumed.
    async fn suppress_analysis_for_zero_capacity(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
    ) {
        self.analyzer_capacity_suppressed = true;
        info!(
            max_parallelism,
            in_flight = in_flight.len(),
            queued = queued.len(),
            manual_resolve_active = self.manual_resolve_active(),
            auto_resolve_active = self.auto_resolve_count.load(Ordering::Relaxed),
            "Skipping dependency analysis because no execution slots are available"
        );
        self.emit_no_analysis_diagnostic(
            queued,
            in_flight,
            max_parallelism,
            "analysis_capacity_zero",
        )
        .await;
    }

    /// Whether this pass suppressed the analyzer purely for lack of capacity.
    ///
    /// Read once by the loop step that owns trigger lifetime, immediately after
    /// the pass it describes.
    pub(super) fn analyzer_capacity_suppressed(&self) -> bool {
        self.analyzer_capacity_suppressed
    }

    async fn prepare_dispatch_candidates(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        iteration: u32,
        work_snapshot: &ReducerWorkSnapshot,
    ) -> ReanalysisFlowDecision {
        // Fail-closed: an incomplete view proves nothing about what is
        // dispatchable, so no analyzer, workspace preparation, or dispatch may
        // start from it. The pass ends without recording any classification.
        if !work_snapshot.is_complete() {
            debug!(
                queued = queued.len(),
                "Skipping dependency analysis because reducer evidence for this evaluation is incomplete"
            );
            return ReanalysisFlowDecision::done(false, iteration);
        }

        let classification = self
            .classify_queued_work_with_snapshot(queued, in_flight, work_snapshot)
            .await;
        if classification.is_blocked_only() {
            info!(
                dispatchable = classification.dispatchable.len(),
                manual_merge_wait = classification.manual_merge_wait.len(),
                scheduler_lane_wait = classification.scheduler_lane_wait.len(),
                terminal_error_retry_required = classification.terminal_error_retry_required.len(),
                dependency_blocked = classification.dependency_blocked.len(),
                candidate_unavailable = classification.candidate_unavailable.len(),
                acceptance_stalled = classification.acceptance_stalled.len(),
                "Skipping dependency analysis because queued work is blocked-only"
            );
            self.emit_no_analysis_diagnostic(
                queued,
                in_flight,
                max_parallelism,
                "blocked_only_no_dispatchable_candidates",
            )
            .await;
            return ReanalysisFlowDecision::done(false, iteration);
        }

        if !classification.has_dispatchable_apply() && !queued.is_empty() {
            info!(
                queued = queued.len(),
                "Skipping dependency analysis because no ordinary dispatchable candidates are available"
            );
            self.emit_no_analysis_diagnostic(
                queued,
                in_flight,
                max_parallelism,
                "no_dispatchable_apply_candidates",
            )
            .await;
            return ReanalysisFlowDecision::done(false, iteration);
        }

        // Keep temporarily dependency-blocked candidates in `queued`: a later reanalysis,
        // such as ResolveCompletion, must re-evaluate them after repository-visible evidence
        // changes. Dispatch selection applies the per-pass gate without dropping them.
        ReanalysisFlowDecision::Continue
    }

    fn compute_effective_reanalysis_reason(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        reanalysis_reason: ReanalysisReason,
    ) -> ReanalysisExecutionDecision {
        let available_slots = self.calculate_available_slots(max_parallelism, in_flight);
        let previous_available_slots = self.last_available_slots.replace(available_slots);
        let slot_recovered = matches!(previous_available_slots, Some(0)) && available_slots > 0;

        if slot_recovered && matches!(reanalysis_reason, ReanalysisReason::QueueNotification) {
            info!(
                previous_available_slots = previous_available_slots.unwrap_or_default(),
                available_slots,
                queued = queued.len(),
                "Execution capacity recovered; promoting queue re-analysis trigger"
            );
        }

        if available_slots == 0 {
            info!(
                max_parallelism,
                in_flight = in_flight.len(),
                queued = queued.len(),
                manual_resolve_active = self.manual_resolve_active(),
                auto_resolve_active = self.auto_resolve_count.load(Ordering::Relaxed),
                "Re-analysis will continue with zero dispatch capacity; ordinary apply dispatch remains suppressed"
            );
        }

        let effective_reason =
            if slot_recovered && matches!(reanalysis_reason, ReanalysisReason::QueueNotification) {
                ReanalysisReason::SlotRecovery
            } else {
                reanalysis_reason
            };

        ReanalysisExecutionDecision {
            effective_reason,
            available_slots,
        }
    }

    /// Probe the repository-visible material a signature needs.
    ///
    /// Returns `Err` with an operator-visible reason on any read or revision failure so the
    /// caller can fail open rather than suppress analysis on partial input.
    pub(super) async fn probe_analysis_signature_materials(
        &self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
    ) -> std::result::Result<(String, std::collections::HashMap<String, String>), String> {
        let referenced_ids: HashSet<&str> = queued
            .iter()
            .map(|change| change.id.as_str())
            .chain(in_flight.iter().map(String::as_str))
            .collect();

        match &self.analysis_input_probe {
            Some(probe) => {
                let base_revision = probe.base_revision().await?;
                let mut digests = std::collections::HashMap::new();
                for change_id in referenced_ids {
                    digests.insert(change_id.to_string(), probe.proposal_digest(change_id)?);
                }
                Ok((base_revision, digests))
            }
            None => {
                // The named effective dependency base, not the checkout commit: dependency
                // merge evidence is read from that ref, and it can advance while `HEAD` does
                // not.
                let base_revision = self
                    .effective_dependency_base_evidence()
                    .await
                    .map_err(|error| error.to_string())?
                    .signature_material();
                let mut digests = std::collections::HashMap::new();
                for change_id in referenced_ids {
                    let path = self
                        .repo_root
                        .join("openspec/changes")
                        .join(change_id)
                        .join("proposal.md");
                    digests.insert(change_id.to_string(), proposal_digest_from_path(&path)?);
                }
                Ok((base_revision, digests))
            }
        }
    }

    /// Build the current analysis-input signature, or `None` when it cannot be built.
    ///
    /// Signature construction is fail-open by design: a proposal read or revision resolution
    /// error emits a deduplicated warning and yields no signature, which permits analysis and
    /// records nothing. It must never terminate the scheduler loop.
    async fn current_analysis_input_signature(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        available_slots: usize,
    ) -> Option<AnalysisInputSignature> {
        match self
            .probe_analysis_signature_materials(queued, in_flight)
            .await
        {
            Ok((base_revision, proposal_digests)) => {
                let in_flight_ids: Vec<String> = in_flight.iter().cloned().collect();
                Some(build_analysis_input_signature(&AnalysisInputMaterials {
                    queued,
                    in_flight_ids: &in_flight_ids,
                    available_slots,
                    max_parallelism,
                    base_revision: &base_revision,
                    proposal_digests: &proposal_digests,
                }))
            }
            Err(error) => {
                warn!(
                    error = %error,
                    queued = queued.len(),
                    in_flight = in_flight.len(),
                    "Could not build dependency-analysis input signature; permitting analysis without suppression"
                );
                let key = DiagnosticDeduplicationKey::AnalysisSignatureUnavailable {
                    queued_len: queued.len(),
                    in_flight_len: in_flight.len(),
                    error: error.clone(),
                };
                let log_entry = LogEntry::warn(format!(
                    "Dependency-analysis input signature unavailable; analysis proceeds without unchanged-input suppression: error={}, queued={}, in_flight={}",
                    error,
                    queued.len(),
                    in_flight.len()
                ));
                self.emit_log_diagnostic_once(
                    key,
                    log_entry,
                    "Suppressing repeated analysis-signature-unavailable diagnostic",
                )
                .await;
                None
            }
        }
    }

    /// Decide whether an ordinary timer evaluation may invoke the expensive analyzer.
    ///
    /// This runs *after* queue classification, reconciliation, and debounce eligibility, so
    /// cheap repository-visible work is never cached — only duplicate analyzer invocation is
    /// removed. Explicit scheduler edges never reach this gate.
    async fn check_unchanged_analysis_input(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        effective_reason: ReanalysisReason,
    ) -> UnchangedInputGate {
        let available_slots = self.calculate_available_slots(max_parallelism, in_flight);

        if bypasses_unchanged_input_gate(effective_reason) {
            // An explicit edge gets one immediate evaluation per event even when its input
            // signature matches or a bounded retry deadline is still pending. Its signature is
            // still captured so that once the edge is consumed, later timer-only wakes fall
            // back to unchanged-input suppression or to bounded fail-open retry.
            let captured = self
                .current_analysis_input_signature(
                    queued,
                    in_flight,
                    max_parallelism,
                    available_slots,
                )
                .await;
            self.note_signature_probe_outcome(captured.is_some());
            return UnchangedInputGate::Analyze { captured };
        }

        let now = tokio::time::Instant::now();

        // A previous attempt established no completed input. Fail-open must not mean fail-hot:
        // before the bounded deadline, an ordinary wake performs no probe and no analysis.
        if let Some(throttle) = self.analysis_retry_throttle {
            if throttle.blocks(now) {
                self.emit_bounded_retry_diagnostic(
                    queued,
                    in_flight,
                    max_parallelism,
                    throttle.cause(),
                )
                .await;
                return UnchangedInputGate::Suppress;
            }
            // The deadline passed: this evaluation retries both signature construction and
            // analysis.
            self.analysis_retry_throttle = None;
        }

        // While an input is suppressed, only re-probe at the bounded cadence. A 500 ms wake
        // before the deadline must not read proposal files or spawn a VCS subprocess.
        if let Some(probe_deadline) = self.next_analysis_signature_probe_at {
            if now < probe_deadline {
                debug!("Skipping dependency-analysis signature probe before its bounded deadline");
                return UnchangedInputGate::Suppress;
            }
        }

        let available_slots = self.calculate_available_slots(max_parallelism, in_flight);
        let Some(signature) = self
            .current_analysis_input_signature(queued, in_flight, max_parallelism, available_slots)
            .await
        else {
            self.note_signature_probe_outcome(false);
            return UnchangedInputGate::Analyze { captured: None };
        };

        let suppressing_record = self
            .last_completed_analysis_input
            .as_ref()
            .filter(|record| record.suppresses(&signature, now));

        match suppressing_record {
            Some(record) => {
                let degraded = record.is_degraded();
                self.next_analysis_signature_probe_at = Some(record.next_probe_deadline(now));
                self.emit_unchanged_analysis_input_diagnostic(
                    queued,
                    in_flight,
                    max_parallelism,
                    &signature,
                    degraded,
                )
                .await;
                UnchangedInputGate::Suppress
            }
            None => {
                self.note_signature_probe_outcome(true);
                UnchangedInputGate::Analyze {
                    captured: Some(signature),
                }
            }
        }
    }

    /// Apply the retry policy implied by one signature-probe attempt.
    ///
    /// A usable signature clears any pending fail-open throttle, because the next pass can rely
    /// on ordinary completed-input suppression instead. An unavailable signature arms the
    /// bounded deadline so persistent proposal-read or revision-resolution failure retries at
    /// the ten-second cadence rather than on every 500 ms wake.
    fn note_signature_probe_outcome(&mut self, signature_available: bool) {
        self.next_analysis_signature_probe_at = None;
        self.analysis_retry_throttle = if signature_available {
            None
        } else {
            Some(BoundedAnalysisRetry::after(
                BoundedRetryCause::SignatureUnavailable,
                tokio::time::Instant::now(),
            ))
        };
    }

    /// Arm the bounded retry deadline after an attempt that produced no usable result.
    fn arm_bounded_analysis_retry(&mut self, cause: BoundedRetryCause) {
        self.next_analysis_signature_probe_at = None;
        self.analysis_retry_throttle = Some(BoundedAnalysisRetry::after(
            cause,
            tokio::time::Instant::now(),
        ));
    }

    /// Operator-visible reason for a wake that is waiting on a bounded fail-open retry.
    async fn emit_bounded_retry_diagnostic(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        cause: BoundedRetryCause,
    ) {
        debug!(
            queued = queued.len(),
            in_flight = in_flight.len(),
            cause = cause.as_str(),
            "Waiting for the bounded dependency-analysis retry deadline"
        );
        let key = DiagnosticDeduplicationKey::BoundedAnalysisRetryPending {
            cause: cause.as_str(),
            queued_len: queued.len(),
            in_flight_len: in_flight.len(),
        };
        let log_entry = LogEntry::info(format!(
            "No analysis started: reason={}, queued={}, in_flight={}, max_parallelism={}",
            cause.as_str(),
            queued.len(),
            in_flight.len(),
            max_parallelism
        ));
        self.emit_log_diagnostic_once(
            key,
            log_entry,
            "Suppressing repeated bounded-analysis-retry diagnostic",
        )
        .await;
        self.emit_no_analysis_diagnostic(queued, in_flight, max_parallelism, cause.as_str())
            .await;
    }

    async fn emit_unchanged_analysis_input_diagnostic(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        signature: &AnalysisInputSignature,
        degraded: bool,
    ) {
        info!(
            queued = queued.len(),
            in_flight = in_flight.len(),
            degraded,
            "Skipping dependency analysis because the analysis input is unchanged since the last completed analysis"
        );
        let key = DiagnosticDeduplicationKey::UnchangedAnalysisInput {
            signature: signature.as_str().to_string(),
            queued_len: queued.len(),
            in_flight_len: in_flight.len(),
            degraded,
        };
        let log_entry = LogEntry::info(format!(
            "No analysis started: reason=unchanged_analysis_input, degraded={}, queued={}, in_flight={}, max_parallelism={}",
            degraded,
            queued.len(),
            in_flight.len(),
            max_parallelism
        ));
        self.emit_log_diagnostic_once(
            key,
            log_entry,
            "Suppressing repeated unchanged-analysis-input diagnostic",
        )
        .await;
        self.emit_no_analysis_diagnostic(
            queued,
            in_flight,
            max_parallelism,
            "unchanged_analysis_input",
        )
        .await;
    }

    /// Record a completed analysis input so unchanged ordinary timer wakes stay quiescent.
    ///
    /// `captured` is the pre-analysis snapshot. A recoverable-failure fallback is recorded as
    /// degraded, so a broken analyzer command is not relaunched on every wake but still gets
    /// one retry after the fixed degraded interval.
    fn record_completed_analysis_input(
        &mut self,
        captured: Option<AnalysisInputSignature>,
        provenance: crate::analyzer::AnalysisProvenance,
    ) {
        let Some(signature) = captured else {
            return;
        };

        let now = tokio::time::Instant::now();
        let record = if provenance.is_degraded() {
            CompletedAnalysisInput::degraded(signature, now)
        } else {
            CompletedAnalysisInput::healthy(signature)
        };
        // Arm the probe deadline at record time so the immediate 500 ms wake after a completed
        // analysis performs no proposal read and spawns no VCS subprocess just to confirm the
        // input it was already derived from.
        self.next_analysis_signature_probe_at = Some(record.next_probe_deadline(now));
        self.last_completed_analysis_input = Some(record);
        // A usable result supersedes any bounded fail-open retry.
        self.analysis_retry_throttle = None;
        debug!(
            degraded = provenance.is_degraded(),
            "Recorded completed dependency-analysis input signature"
        );
    }

    /// Drop any suppression for the current input.
    ///
    /// Used when an analysis pass produced no usable dispatch decision while the scheduler was
    /// otherwise idle: an erroneous non-deterministic dependency result must not be able to
    /// freeze an idle scheduler permanently.
    fn forget_completed_analysis_input(&mut self) {
        self.last_completed_analysis_input = None;
        self.next_analysis_signature_probe_at = None;
        self.analysis_retry_throttle = None;
    }

    async fn should_run_analysis_now(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        iteration: u32,
        effective_reason: ReanalysisReason,
    ) -> ReanalysisFlowDecision {
        let bypass_debounce = bypasses_unchanged_input_gate(effective_reason);

        let should_analyze = if iteration == 1 {
            info!("First iteration, skipping debounce check");
            true
        } else {
            self.should_reanalyze(bypass_debounce).await
        };

        if should_analyze {
            ReanalysisFlowDecision::Continue
        } else {
            info!("Debounce active, waiting for timer or queue notification");
            self.emit_no_analysis_diagnostic(queued, in_flight, max_parallelism, "debounce_active")
                .await;
            ReanalysisFlowDecision::done(false, iteration)
        }
    }

    /// Announce failed-dependency blocking once per blocker epoch and decide
    /// whether this pass still has non-blocked candidates to analyse.
    ///
    /// Candidates are never removed from `queued`: they stay locally
    /// represented as dependency-blocked queued work, which is what makes
    /// reconciliation report `queued_added == 0` on every later wake instead of
    /// synthesising a fresh queue edge. `select_changes_for_dispatch` and
    /// `classify_queued_work` apply the dispatch gate itself.
    ///
    /// Exactly two events are emitted per `(change, failed blocker set)` epoch:
    /// a compatibility `ChangeSkipped` — which observes dispatch exclusion and
    /// never revokes accepted queue intent — and the authoritative
    /// `DependencyBlocked` transition.
    async fn observe_failed_dependency_blocks(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        iteration: u32,
    ) -> ReanalysisFlowDecision {
        let blocked = self.failed_dependency_blocked_candidates(queued);

        // A candidate that is present but no longer failed-dependency blocked
        // ends its epoch here, so a later refailure is announced as a genuinely
        // new transition rather than suppressed as a duplicate.
        for change in queued {
            if !blocked.iter().any(|(id, _)| id == &change.id) {
                self.failed_tracker.clear_blocker_epoch(&change.id);
            }
        }

        for (change_id, blockers) in &blocked {
            // The blocker epoch bounds *events*, not state: the current failed
            // dependency set is reconciled into the projection on every pass so
            // a rebuilt reducer still reports the wait.
            self.reconcile_dependency_blocker_projection(change_id, blockers)
                .await;

            if !self.failed_tracker.begin_blocker_epoch(change_id, blockers) {
                debug!(
                    change_id = %change_id,
                    blockers = ?blockers,
                    "Suppressing repeated failed-dependency observation for an unchanged blocker epoch"
                );
                continue;
            }

            // `should_skip` picks the first declared failed dependency; the
            // compatibility reason keeps that exact wording so existing
            // consumers are unaffected.
            let primary = blockers
                .first()
                .cloned()
                .unwrap_or_else(|| change_id.clone());
            warn!(
                "Skipping change-{} because dependency change-{} failed",
                change_id, primary
            );
            send_event(
                &self.event_tx,
                ParallelEvent::ChangeSkipped {
                    change_id: change_id.clone(),
                    reason: format!("Dependency '{}' failed", primary),
                },
            )
            .await;
            send_event(
                &self.event_tx,
                ParallelEvent::DependencyBlocked {
                    change_id: change_id.clone(),
                    dependency_ids: blockers.clone(),
                },
            )
            .await;
        }

        if !queued.is_empty() && blocked.len() == queued.len() {
            info!("All queued changes are blocked by failed dependencies");
            self.emit_no_analysis_diagnostic(
                queued,
                in_flight,
                max_parallelism,
                "all_candidates_failed_dependency_blocked",
            )
            .await;
            // Never `should_break` here: blocked work is not drained work. The
            // canonical drain and blocked-only lifetime checks own termination,
            // so a finite run reports blocked/stalled and a persistent run waits
            // for an explicit notification.
            return ReanalysisFlowDecision::done(false, iteration);
        }

        ReanalysisFlowDecision::Continue
    }

    async fn run_dependency_analysis_attempt<F>(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        iteration: u32,
        effective_reason: ReanalysisReason,
        analyzer: &F,
    ) -> DependencyAnalysisAttempt
    where
        for<'a> F: Fn(
                &'a [crate::openspec::Change],
                &'a [String],
                u32,
            )
                -> std::pin::Pin<Box<dyn std::future::Future<Output = AnalysisOutcome> + Send + 'a>>
            + Send
            + Sync,
    {
        info!(
            "Re-analysis triggered: iteration={}, queued={}, in_flight={}, trigger={}",
            iteration,
            queued.len(),
            in_flight.len(),
            effective_reason
        );
        let attempt_id = analysis_attempt_id(iteration, effective_reason, queued);
        send_event(
            &self.event_tx,
            ParallelEvent::AnalysisStarted {
                remaining_changes: queued.len(),
                attempt_id,
            },
        )
        .await;

        // The analyzer prompt must not depend on `HashSet` iteration order: the signature treats
        // in-flight membership as a set, so the consumer has to see one deterministic ordering
        // of that same set.
        let mut in_flight_ids: Vec<String> = in_flight.iter().cloned().collect();
        in_flight_ids.sort_unstable();
        let analysis_outcome = analyzer(queued, &in_flight_ids, iteration).await;

        if analysis_outcome.result.order.is_empty() {
            // An unusable analysis decision must not establish a completed input, or it would
            // suppress every later attempt. It is also not evidence that queued work finished:
            // ending the scheduler here would strand reducer-visible queued intent. Keep the
            // loop alive, let the canonical drain checks own termination, and rate-limit the
            // retry so a persistently unusable analyzer is not relaunched every 500 ms.
            warn!("No order returned from analysis");
            let queued_work_remains = !queued.is_empty();
            if queued_work_remains {
                self.arm_bounded_analysis_retry(BoundedRetryCause::UnusableAnalysisResult);
            }
            return DependencyAnalysisAttempt::EmptyOrder(ReanalysisFlowDecision::done(
                !queued_work_remains && in_flight.is_empty(),
                iteration,
            ));
        }

        DependencyAnalysisAttempt::Completed(analysis_outcome)
    }

    async fn handle_post_analysis_capacity(
        &mut self,
        queued: &[crate::openspec::Change],
        in_flight: &HashSet<String>,
        max_parallelism: usize,
        _iteration: u32,
        analysis_result: &AnalysisResult,
    ) -> Result<Option<usize>> {
        self.failed_tracker
            .set_dependencies(analysis_result.dependencies.clone());
        self.change_dependencies = analysis_result.dependencies.clone();

        let available_slots = self.calculate_available_slots(max_parallelism, in_flight);
        info!(
            "Available slots after analysis: {} (max: {}, in_flight: {}, queued: {})",
            available_slots,
            max_parallelism,
            in_flight.len(),
            queued.len()
        );

        if available_slots == 0 {
            info!(
                max_parallelism,
                in_flight = in_flight.len(),
                queued = queued.len(),
                order = ?analysis_result.order,
                "Dependency analysis completed, but dispatch is suppressed because no execution slots are available"
            );
            self.emit_capacity_zero_dispatch_diagnostic_once(
                queued,
                in_flight,
                max_parallelism,
                &analysis_result.order,
            )
            .await;
            self.emit_no_analysis_diagnostic(
                queued,
                in_flight,
                max_parallelism,
                "dispatch_capacity_zero_after_analysis",
            )
            .await;
            return Ok(None);
        }

        Ok(Some(available_slots))
    }

    async fn dispatch_selected_candidates(
        &mut self,
        ctx: DispatchSelectedCandidatesContext<'_>,
    ) -> Result<DispatchSelectionOutcome> {
        let DispatchSelectedCandidatesContext {
            queued,
            in_flight,
            iteration,
            analysis_result,
            available_slots,
            join_set,
            cleanup_guard,
        } = ctx;
        let selected_changes = self
            .select_changes_for_dispatch(analysis_result, available_slots, in_flight)
            .await;

        if selected_changes.is_empty() {
            return Ok(DispatchSelectionOutcome {
                iteration,
                selected: 0,
            });
        }

        let base_revision = self
            .workspace_manager
            .get_current_revision()
            .await
            .map_err(OrchestratorError::from)?;

        info!(
            "Dispatching {} changes (iteration {}): {:?}",
            selected_changes.len(),
            iteration,
            selected_changes
        );

        for change_id in &selected_changes {
            if self.is_cancelled() {
                info!("Stopping selected-change dispatch loop because parallel execution is cancelled");
                break;
            }
            if let Err(e) = self
                .dispatch_change_to_workspace(
                    change_id.clone(),
                    base_revision.clone(),
                    join_set,
                    in_flight,
                    cleanup_guard,
                )
                .await
            {
                let message = format!("Failed to dispatch change '{}': {}", change_id, e);
                self.failed_tracker.mark_failed(change_id);
                send_event(
                    &self.event_tx,
                    ParallelEvent::ProcessingError {
                        id: change_id.clone(),
                        error: message.clone(),
                    },
                )
                .await;
                send_event(
                    &self.event_tx,
                    ParallelEvent::Log(LogEntry::error(message.clone())),
                )
                .await;
                error!("{}", message);
            }
        }

        let dispatched_set: HashSet<_> = selected_changes.iter().collect();
        queued.retain(|c| !dispatched_set.contains(&c.id));

        Ok(DispatchSelectionOutcome {
            iteration: iteration + 1,
            selected: selected_changes.len(),
        })
    }

    async fn before_dependency_analysis<F>(
        &mut self,
        inputs: DependencyAnalysisInputs<'_, F>,
    ) -> Result<DependencyAnalysisPass>
    where
        for<'a> F: Fn(
                &'a [crate::openspec::Change],
                &'a [String],
                u32,
            )
                -> std::pin::Pin<Box<dyn std::future::Future<Output = AnalysisOutcome> + Send + 'a>>
            + Send
            + Sync,
    {
        let DependencyAnalysisInputs {
            queued,
            in_flight,
            max_parallelism,
            iteration,
            reanalysis_reason,
            analyzer,
            work_snapshot,
        } = inputs;

        // Failed-dependency observation runs before classification and before
        // the debounce gate: it owns bounded, once-per-epoch operator
        // notification, so a blocked-only pass that stops at classification
        // still reports why.
        if let Some((should_break, iteration)) = self
            .observe_failed_dependency_blocks(queued, in_flight, max_parallelism, iteration)
            .await
            .into_result()
        {
            return Ok(DependencyAnalysisPass::terminal(
                ReanalysisFlowDecision::done(should_break, iteration),
            ));
        }

        if let Some((should_break, iteration)) = self
            .prepare_dispatch_candidates(
                queued,
                in_flight,
                max_parallelism,
                iteration,
                work_snapshot,
            )
            .await
            .into_result()
        {
            return Ok(DependencyAnalysisPass::terminal(
                ReanalysisFlowDecision::done(should_break, iteration),
            ));
        }

        // Capacity is recomputed inside this call, and `last_available_slots` is
        // recorded there — which is what lets the *next* pass recognise zero to
        // positive recovery and promote its queue trigger. So the gate below has
        // to run after it, never instead of it.
        let analysis_decision = self.compute_effective_reanalysis_reason(
            queued,
            in_flight,
            max_parallelism,
            reanalysis_reason,
        );

        // Classification, reducer-intent reconciliation, and operator diagnostics
        // have all run by now; only the expensive analyzer and ordinary dispatch
        // are gated on capacity. Nothing below this point may execute at zero
        // slots, because everything below it either invokes the analyzer or
        // records suppression state derived from having invoked it.
        if analysis_decision.available_slots == 0 {
            self.suppress_analysis_for_zero_capacity(queued, in_flight, max_parallelism)
                .await;
            return Ok(DependencyAnalysisPass::terminal(
                ReanalysisFlowDecision::done(false, iteration),
            ));
        }

        if let Some((should_break, iteration)) = self
            .should_run_analysis_now(
                queued,
                in_flight,
                max_parallelism,
                iteration,
                analysis_decision.effective_reason,
            )
            .await
            .into_result()
        {
            return Ok(DependencyAnalysisPass::terminal(
                ReanalysisFlowDecision::done(should_break, iteration),
            ));
        }

        // Queue classification, reconciliation, dependency blocker checks, and debounce
        // eligibility have all run against fresh repository state by now. Only the expensive
        // analyzer invocation is subject to unchanged-input suppression.
        let captured_signature = match self
            .check_unchanged_analysis_input(
                queued,
                in_flight,
                max_parallelism,
                analysis_decision.effective_reason,
            )
            .await
        {
            UnchangedInputGate::Analyze { captured } => captured,
            UnchangedInputGate::Suppress => {
                return Ok(DependencyAnalysisPass::terminal(
                    ReanalysisFlowDecision::done(false, iteration),
                ));
            }
        };

        // The evaluation the edge authorized is now certain to happen, so this is
        // where its authority is spent. Everything above may end the pass early,
        // and every one of those exits must leave the latch armed for the next
        // eligible evaluation instead of discarding it with this pass.
        self.spend_pending_retry_bypass();

        let attempt = self
            .run_dependency_analysis_attempt(
                queued,
                in_flight,
                iteration,
                analysis_decision.effective_reason,
                analyzer,
            )
            .await;

        Ok(DependencyAnalysisPass {
            attempt,
            captured_signature,
        })
    }

    pub(super) async fn perform_reanalysis_and_dispatch<F>(
        &mut self,
        ctx: ReanalysisDispatchContext<'_, F>,
    ) -> Result<(bool, u32)>
    where
        for<'a> F: Fn(
                &'a [crate::openspec::Change],
                &'a [String],
                u32,
            ) -> std::pin::Pin<
                Box<dyn std::future::Future<Output = crate::analyzer::AnalysisOutcome> + Send + 'a>,
            > + Send
            + Sync,
    {
        let ReanalysisDispatchContext {
            queued,
            in_flight,
            max_parallelism,
            iteration,
            reanalysis_reason,
            analyzer,
            join_set,
            cleanup_guard,
            work_snapshot,
        } = ctx;

        // The evaluation's own limit is the ceiling admission must enforce, so a
        // scheduler configured above this executor's construction-time limit
        // grows the slot ceiling rather than silently capping dispatch.
        self.lifecycle_slots.ensure_capacity(max_parallelism);

        // Per-pass, and cleared before anything can set it: a stale flag from an
        // earlier evaluation would preserve an edge this pass really did consume.
        self.analyzer_capacity_suppressed = false;

        if self.is_cancelled() {
            info!(
                "Skipping dependency analysis and dispatch because parallel execution is cancelled"
            );
            return Ok((true, iteration));
        }

        // The loop hands down the view it already captured for this pass; a
        // direct caller gets an equivalent awaited one. Either way classification
        // never re-derives reducer facts from a second, possibly different
        // revision.
        let captured_work_snapshot;
        let work_snapshot = match work_snapshot {
            Some(snapshot) => snapshot,
            None => {
                captured_work_snapshot = self.capture_reducer_work_snapshot().await;
                &captured_work_snapshot
            }
        };

        let pass = self
            .before_dependency_analysis(DependencyAnalysisInputs {
                queued,
                in_flight,
                max_parallelism,
                iteration,
                reanalysis_reason,
                analyzer,
                work_snapshot,
            })
            .await?;
        let captured_signature = pass.captured_signature;
        let analysis_outcome = match pass.attempt {
            DependencyAnalysisAttempt::Completed(outcome) => outcome,
            DependencyAnalysisAttempt::EmptyOrder(decision) => {
                if let Some(result) = decision.into_result() {
                    return Ok(result);
                }
                unreachable!("empty analysis order always produces a terminal flow decision")
            }
        };
        let provenance = analysis_outcome.provenance;
        let analysis_result = analysis_outcome.result;

        let was_idle_before_dispatch = in_flight.is_empty();
        let available_slots = match self
            .handle_post_analysis_capacity(
                queued,
                in_flight,
                max_parallelism,
                iteration,
                &analysis_result,
            )
            .await
        {
            Ok(Some(slots)) => slots,
            Ok(None) => {
                // Zero capacity with a usable analysis result is the state that produced the
                // live no-progress loop: the input was genuinely analyzed, so record it and
                // let capacity recovery change the signature later.
                self.record_completed_analysis_input(captured_signature, provenance);
                return Ok((false, iteration));
            }
            Err(error) => return Err(error),
        };

        let dispatch = self
            .dispatch_selected_candidates(DispatchSelectedCandidatesContext {
                queued,
                in_flight,
                iteration,
                analysis_result: &analysis_result,
                available_slots,
                join_set,
                cleanup_guard,
            })
            .await?;

        if dispatch.selected == 0 && was_idle_before_dispatch {
            // Positive capacity, nothing in flight, and nothing selected: an erroneous
            // non-deterministic dependency result must not be able to freeze an otherwise idle
            // scheduler, so leave this input eligible for the next debounced evaluation.
            info!(
                queued = queued.len(),
                available_slots,
                "Analysis selected no dispatch while idle with positive capacity; leaving the input eligible for re-analysis"
            );
            self.forget_completed_analysis_input();
        } else {
            self.record_completed_analysis_input(captured_signature, provenance);
        }

        Ok((false, dispatch.iteration))
    }
}