cflx 0.6.327

//! Shared parallel execution service for CLI and TUI modes.
//!
//! This module provides a unified service for running parallel execution
//! that can be used by both CLI and TUI orchestrators, eliminating
//! code duplication between the two modes.

use crate::ai_command_runner::{AiCommandRunner, RunCommandScope, SharedStaggerState};
use crate::analyzer::{ParallelGroup, ParallelizationAnalyzer};
use crate::config::OrchestratorConfig;
use crate::dependency_targets::union_metadata_dependencies;
use crate::error::Result;
use crate::hooks::HookRunner;
use crate::openspec::Change;
use crate::parallel::dedup::{DiagnosticDeduplicationKey, DiagnosticDeduplicationStore};
use crate::parallel::{ParallelEvent, ParallelExecutor, PostArchiveAction, SchedulerRunReport};
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};

type AnalysisDiagnosticStore = Arc<Mutex<DiagnosticDeduplicationStore<DiagnosticDeduplicationKey>>>;

/// Service for parallel execution of changes.
///
/// This service encapsulates the shared logic between CLI and TUI
/// parallel execution modes, including:
/// - Git availability checking
/// - Dependency-based analysis
/// - ParallelExecutor coordination
pub struct ParallelRunService {
    /// Configuration for the orchestrator
    config: OrchestratorConfig,
    /// Repository root directory
    repo_root: PathBuf,
    /// Disable automatic workspace resume (always create new workspaces)
    no_resume: bool,
    /// Shared stagger state for coordinating AI command execution delays
    shared_stagger_state: SharedStaggerState,
    /// Terminal action after successful archive.
    post_archive_action: PostArchiveAction,
    /// Shared reducer state used by CLI/server/TUI paths for base-mutating lane scheduling.
    shared_orchestrator_state:
        Arc<tokio::sync::RwLock<crate::orchestration::state::OrchestratorState>>,
    /// AI command runner for analyze commands
    ai_runner: AiCommandRunner,
    /// Invocation-scoped ownership of every AI command this run launches.
    ///
    /// Created once per service and shared with the executor it builds, so
    /// analyze, Apply, Archive, Acceptance, cleanup review, rejection review,
    /// conflict resolve, and upstream repair all report into one barrier.
    run_command_scope: RunCommandScope,
    /// Runtime-only observability dedupe for stable analyzer failure signatures.
    diagnostic_dedup: AnalysisDiagnosticStore,
    /// Invocation-scoped upstream integration runtime.
    ///
    /// `None` installs no checkpoint at all; it is never sourced from persistent
    /// orchestration config.
    upstream_integration: Option<crate::upstream::UpstreamRuntime>,
    /// Explicit-target classification deferred to the executor's post-checkpoint
    /// boundary. Only an enabled real `-u` run installs one.
    explicit_target_plan: Option<crate::orchestration::target_resolution::ExplicitTargetPlan>,
    /// The run owner's pending graceful-stop request, when one is bound.
    ///
    /// Handed to the executor this service builds so the scheduler loop reads the
    /// same request shared run control records through the scheduler port. A
    /// service no owner bound one to builds an executor that observes none.
    graceful_stop: Option<Arc<std::sync::atomic::AtomicBool>>,
}

impl ParallelRunService {
    /// Create a new parallel run service
    pub fn new(repo_root: PathBuf, config: OrchestratorConfig) -> Self {
        let shared_stagger_state: SharedStaggerState = Arc::new(Mutex::new(None));
        let run_command_scope = RunCommandScope::new();
        let ai_runner = AiCommandRunner::for_run(
            &config,
            shared_stagger_state.clone(),
            run_command_scope.clone(),
        );

        let shared_orchestrator_state = Arc::new(tokio::sync::RwLock::new(
            crate::orchestration::state::OrchestratorState::new(Vec::new(), 1),
        ));

        Self {
            config,
            repo_root,
            no_resume: false,
            shared_stagger_state,
            post_archive_action: PostArchiveAction::MergeToBase,
            shared_orchestrator_state,
            ai_runner,
            run_command_scope,
            diagnostic_dedup: Arc::new(Mutex::new(DiagnosticDeduplicationStore::new())),
            upstream_integration: None,
            explicit_target_plan: None,
            graceful_stop: None,
        }
    }

    /// Create a new parallel run service with a shared stagger state
    pub fn new_with_shared_state(
        repo_root: PathBuf,
        config: OrchestratorConfig,
        shared_stagger_state: SharedStaggerState,
    ) -> Self {
        let run_command_scope = RunCommandScope::new();
        let ai_runner = AiCommandRunner::for_run(
            &config,
            shared_stagger_state.clone(),
            run_command_scope.clone(),
        );

        let shared_orchestrator_state = Arc::new(tokio::sync::RwLock::new(
            crate::orchestration::state::OrchestratorState::new(Vec::new(), 1),
        ));

        Self {
            config,
            repo_root,
            no_resume: false,
            shared_stagger_state,
            post_archive_action: PostArchiveAction::MergeToBase,
            shared_orchestrator_state,
            ai_runner,
            run_command_scope,
            diagnostic_dedup: Arc::new(Mutex::new(DiagnosticDeduplicationStore::new())),
            upstream_integration: None,
            explicit_target_plan: None,
            graceful_stop: None,
        }
    }

    /// Set whether to disable automatic workspace resume.
    ///
    /// When `no_resume` is true, existing workspaces are always deleted
    /// and new ones are created. When false (default), existing workspaces
    /// are reused to resume interrupted work.
    pub fn set_no_resume(&mut self, no_resume: bool) {
        self.no_resume = no_resume;
    }

    pub fn set_post_archive_action(&mut self, action: PostArchiveAction) {
        self.post_archive_action = action;
    }

    /// Install invocation-scoped upstream integration.
    pub fn set_upstream_integration(&mut self, runtime: crate::upstream::UpstreamRuntime) {
        self.upstream_integration = Some(runtime);
    }

    #[cfg(test)]
    pub fn upstream_integration(&self) -> Option<&crate::upstream::UpstreamRuntime> {
        self.upstream_integration.as_ref()
    }

    /// Bind the run owner's pending graceful-stop request.
    ///
    /// Only a run owner that already owns the flag shared run control writes
    /// through calls this; every other caller leaves the executor with no request
    /// to observe.
    pub fn set_graceful_stop_flag(&mut self, graceful_stop: Arc<std::sync::atomic::AtomicBool>) {
        self.graceful_stop = Some(graceful_stop);
    }

    /// Install deferred explicit-target classification.
    ///
    /// Used by an enabled real `-u` run, where the requested set must be
    /// classified against the cumulative base produced by the initial upstream
    /// checkpoint rather than the pre-checkpoint base.
    pub fn set_explicit_target_plan(
        &mut self,
        plan: crate::orchestration::target_resolution::ExplicitTargetPlan,
    ) {
        self.explicit_target_plan = Some(plan);
    }

    /// Install the coordinator on an executor when the option is enabled.
    ///
    /// A disabled service leaves the executor untouched, so no upstream object
    /// is constructed on the default-off path.
    fn install_upstream_integration(&self, executor: &mut ParallelExecutor) {
        if let Some(runtime) = &self.upstream_integration {
            executor.set_upstream_integration(runtime.clone());
        }
        if let Some(plan) = &self.explicit_target_plan {
            executor.set_explicit_target_plan(plan.clone());
        }
    }

    #[cfg(test)]
    pub fn post_archive_action(&self) -> &PostArchiveAction {
        &self.post_archive_action
    }

    /// Set shared reducer state for callers that already own UI/server state.
    pub fn set_shared_orchestrator_state(
        &mut self,
        shared_state: Arc<tokio::sync::RwLock<crate::orchestration::state::OrchestratorState>>,
    ) {
        self.shared_orchestrator_state = shared_state;
    }

    /// The invocation scope owning every AI command this service launches.
    #[allow(dead_code)] // Read by scope-ownership coverage; the run owner injects its own.
    pub fn run_command_scope(&self) -> RunCommandScope {
        self.run_command_scope.clone()
    }

    /// Adopt a scope created by the run owner (the local TUI supervisor), so a
    /// clone survives outside the spawned orchestrator task.
    pub fn set_run_command_scope(&mut self, scope: RunCommandScope) {
        self.ai_runner.set_run_command_scope(scope.clone());
        self.run_command_scope = scope;
    }

    /// Check that a usable Git workspace is available for worktree execution.
    ///
    /// Startup already refuses an unusable workspace; this is the executor's own
    /// guard so a programmatic caller cannot bypass it.
    pub async fn check_vcs_available(&self) -> Result<()> {
        if !crate::cli::check_git_workspace_usable() {
            return Err(crate::error::OrchestratorError::GitCommand(
                "Git repository not available for worktree execution".to_string(),
            ));
        }
        Ok(())
    }

    /// Create a configured ParallelExecutor instance with optional shared queue change state.
    ///
    /// This allows external callers to share queue change timestamps across multiple executors,
    /// enabling debounce logic to work across re-analysis iterations.
    pub fn create_executor_with_queue_state(
        &self,
        event_tx: Option<mpsc::Sender<ParallelEvent>>,
        cancel_token: Option<CancellationToken>,
        shared_queue_change: Option<std::sync::Arc<tokio::sync::Mutex<Option<std::time::Instant>>>>,
        dynamic_queue: Option<std::sync::Arc<crate::tui::queue::DynamicQueue>>,
        manual_resolve_counter: Option<std::sync::Arc<std::sync::atomic::AtomicUsize>>,
        shared_orchestrator_state: Option<
            std::sync::Arc<tokio::sync::RwLock<crate::orchestration::state::OrchestratorState>>,
        >,
    ) -> ParallelExecutor {
        let vcs_backend = self.config.get_vcs_backend();

        // Create hooks before moving event_tx
        let hooks = if let Some(ref tx) = event_tx {
            HookRunner::with_event_tx(self.config.get_hooks(), &self.repo_root, tx.clone())
        } else {
            HookRunner::new(self.config.get_hooks(), &self.repo_root)
        };

        let has_dynamic_queue = dynamic_queue.is_some();
        let mut executor = ParallelExecutor::with_backend_and_queue_and_stagger(
            self.repo_root.clone(),
            self.config.clone(),
            event_tx,
            vcs_backend,
            shared_queue_change,
            Some(self.shared_stagger_state.clone()),
        );
        executor.set_run_command_scope(self.run_command_scope.clone());
        executor.set_no_resume(self.no_resume);
        executor.set_post_archive_action(self.post_archive_action.clone());
        self.install_upstream_integration(&mut executor);

        if has_dynamic_queue {
            // Loop-based frontends (TUI/server) should stay alive when idle
            // and wait for new queue notifications.
            executor.set_persistent_lifetime();
        }

        if let Some(graceful_stop) = &self.graceful_stop {
            executor.set_graceful_stop_flag(graceful_stop.clone());
        }

        executor.set_hooks(hooks);

        if let Some(token) = cancel_token {
            executor.set_cancel_token(token);
        }
        if let Some(queue) = dynamic_queue {
            executor.set_dynamic_queue(queue);
        }
        if let Some(counter) = manual_resolve_counter {
            executor.set_manual_resolve_counter(counter);
        }
        if let Some(shared_state) = shared_orchestrator_state {
            executor.set_shared_orchestrator_state(shared_state);
        }
        executor
    }

    async fn filter_committed_changes(
        &self,
        changes: Vec<Change>,
    ) -> Result<(Vec<Change>, Vec<String>)> {
        filter_committed_changes_at(&self.repo_root, changes).await
    }

    /// Prepare changes for parallel execution: filter committed changes and send warning event if needed.
    ///
    /// This helper consolidates the preparation logic shared across multiple execution paths:
    /// 1. Filters changes to only include those committed to the repository
    /// 2. Sends a warning event if uncommitted changes are skipped (before any state update)
    /// 3. Returns the filtered changes, or None if no committed changes remain
    ///
    /// The event is sent synchronously before returning to maintain proper event ordering.
    async fn prepare_parallel_execution(
        &self,
        changes: Vec<Change>,
        event_tx: &mpsc::Sender<ParallelEvent>,
        allow_empty_when_resolve_wait: bool,
    ) -> Result<Option<Vec<Change>>> {
        let (changes, skipped) = self.filter_committed_changes(changes).await?;

        // Send warning event BEFORE any state update to maintain event order
        if !skipped.is_empty() {
            let message = format!("Skipping uncommitted changes: {}", skipped.join(", "));
            warn!("{}", message);
            let _ = event_tx
                .send(ParallelEvent::Warning {
                    title: "Uncommitted changes skipped".to_string(),
                    message,
                })
                .await;
            // Send explicit rejection event so callers can reconcile user-visible state
            // (e.g. TUI resets Queued rows, CLI reports zero-start)
            let _ = event_tx
                .send(ParallelEvent::ParallelStartRejected {
                    change_ids: skipped.clone(),
                    reason: "uncommitted or not in HEAD".to_string(),
                })
                .await;
        }

        if changes.is_empty() {
            if allow_empty_when_resolve_wait {
                info!(
                    "No committed changes available, but scheduler-owned ResolveWait retry is present; continuing with empty queue"
                );
                return Ok(Some(changes));
            }
            info!("No committed changes available for parallel execution");
            return Ok(None);
        }

        Ok(Some(changes))
    }

    /// Run parallel execution with an event callback
    ///
    /// The event_handler receives ParallelEvents as they occur during execution.
    /// Returns the execution result.
    ///
    /// This method now uses `execute_with_reanalysis` for dynamic re-analysis,
    /// matching the TUI behavior and aligning with the spec requirement for
    /// unified CLI/TUI execution paths.
    pub async fn run_parallel<F>(
        &self,
        changes: Vec<Change>,
        cancel_token: Option<CancellationToken>,
        event_handler: F,
    ) -> Result<()>
    where
        F: Fn(ParallelEvent) + Send + Sync + 'static,
    {
        // Create event channel
        let (event_tx, mut event_rx) = mpsc::channel::<ParallelEvent>(100);

        // Prepare changes using the common helper (sends warning event if needed)
        {
            let mut guard = self.shared_orchestrator_state.write().await;
            for change in &changes {
                guard.add_dynamic_change(change.id.clone());
            }
        }

        let changes = match self
            .prepare_parallel_execution(changes, &event_tx, true)
            .await?
        {
            Some(changes) => changes,
            None => {
                // All changes were rejected before execution started.
                // The forwarding task has not been spawned yet, so drain any buffered
                // rejection events directly and forward them to the caller before returning.
                drop(event_tx);
                while let Some(event) = event_rx.recv().await {
                    event_handler(event);
                }
                return Ok(());
            }
        };

        // A CLI target list is explicit operator intent, exactly like a TUI or
        // remote Start, so it enters the same reducer contract that
        // `initialize_parallel_shared_state` and `RunControlService::start_marked`
        // use: the scheduler reads reducer queue intent — not the local candidate
        // list — when it decides what may still run, and registration alone
        // leaves a change `NotQueued`. Intent is recorded *after* the committed
        // filter, so a change this run just rejected is not granted eligibility
        // that reconciliation would hand straight back.
        {
            let mut guard = self.shared_orchestrator_state.write().await;
            for change in &changes {
                guard.apply_command(crate::orchestration::state::ReducerCommand::AddToQueue(
                    change.id.clone(),
                ));
            }
        }

        // Spawn event forwarding task
        let forward_handle = tokio::spawn(async move {
            while let Some(event) = event_rx.recv().await {
                let is_completed =
                    matches!(event, ParallelEvent::AllCompleted | ParallelEvent::Stopped);
                event_handler(event);
                if is_completed {
                    break;
                }
            }
        });

        // One construction path for every frontend. The CLI used to assemble its
        // own executor here, and that duplicate omitted `post_archive_action`
        // and the run owner's graceful-stop request — which is how a headless
        // `--push` run degraded to the default merge action. No dynamic queue is
        // supplied, so the shared builder leaves the finite lifetime a CLI run
        // depends on untouched.
        let executor = self.create_executor_with_queue_state(
            Some(event_tx.clone()),
            cancel_token,
            None,
            None,
            None,
            Some(self.shared_orchestrator_state.clone()),
        );

        // Delegate to the shared order-based run rather than repeating its
        // analyzer closure. Its `prepare_parallel_execution` runs a second time
        // over an already-filtered list, which is a no-op: nothing is skipped
        // twice, so no duplicate warning or rejection event is emitted.
        //
        // `event_tx` is moved in, leaving no sender clone here. That is what
        // closes the channel — and releases the forwarder — when the delegate
        // drops the executor and the analyzer closure, including on the terminal
        // paths that never emit `AllCompleted` (a scheduler failure, or a
        // blocked/stalled exit).
        let result = self
            .run_parallel_order_based_with_executor(executor, changes, event_tx)
            .await;

        // Wait for event forwarding to complete
        let _ = forward_handle.await;

        // The callback API reports through events; the typed terminal report is
        // consumed by boundaries that publish their own completion transition.
        result.map(|_report| ())
    }

    /// Run parallel execution with an mpsc sender for events and optional shared queue change state.
    ///
    /// This variant is useful when integrating with existing channel-based
    /// event systems (e.g., TUI).
    ///
    /// Uses dynamic re-analysis: after each dispatch iteration completes, the remaining changes
    /// are re-analyzed to determine the next dispatch.
    ///
    /// The `shared_queue_change` parameter allows tracking queue changes across multiple
    /// re-analysis iterations for proper debouncing behavior.
    #[allow(clippy::too_many_arguments)]
    pub async fn run_parallel_with_channel_and_queue_state(
        &self,
        changes: Vec<Change>,
        event_tx: mpsc::Sender<ParallelEvent>,
        cancel_token: Option<CancellationToken>,
        shared_queue_change: Option<std::sync::Arc<tokio::sync::Mutex<Option<std::time::Instant>>>>,
        dynamic_queue: Option<std::sync::Arc<crate::tui::queue::DynamicQueue>>,
        manual_resolve_counter: Option<std::sync::Arc<std::sync::atomic::AtomicUsize>>,
        shared_orchestrator_state: Option<
            std::sync::Arc<tokio::sync::RwLock<crate::orchestration::state::OrchestratorState>>,
        >,
        explicit_retry: bool,
    ) -> Result<SchedulerRunReport> {
        let mut executor = self.create_executor_with_queue_state(
            Some(event_tx.clone()),
            cancel_token,
            shared_queue_change,
            dynamic_queue,
            manual_resolve_counter,
            shared_orchestrator_state,
        );
        executor.set_explicit_retry(explicit_retry);
        // Use order-based execution (aligned with spec)
        self.run_parallel_order_based_with_executor(executor, changes, event_tx)
            .await
    }

    /// Run parallel execution with order-based analysis using a pre-configured executor.
    ///
    /// This is the preferred execution method that aligns with the parallel-execution spec.
    /// Uses `order` directly to select changes based on available slots.
    pub async fn run_parallel_order_based_with_executor(
        &self,
        mut executor: ParallelExecutor,
        changes: Vec<Change>,
        event_tx: mpsc::Sender<ParallelEvent>,
    ) -> Result<SchedulerRunReport> {
        // Prepare changes using the common helper (sends warning event if needed).
        // Preserve caller-provided reducer state: manual TUI retry startup records
        // ResolveWait/RejectWait in the reducer before constructing the executor, and
        // replacing that reducer here creates split-brain retry ownership.
        executor.ensure_shared_orchestrator_state(self.shared_orchestrator_state.clone());
        // An opted-in run must reach the executor's upstream boundary even with
        // an empty queue: a change that is archived and integrated but not yet
        // published owes only publication, and it is no longer an active change,
        // so short-circuiting here would strand it unpublished with no way for
        // an operator retry to resume it.
        let allow_empty_queue = changes.is_empty()
            && (executor.has_resolve_wait().await || executor.has_upstream_integration());
        let changes = match self
            .prepare_parallel_execution(changes, &event_tx, allow_empty_queue)
            .await?
        {
            Some(changes) => changes,
            None => return Ok(SchedulerRunReport::Completed),
        };

        info!(
            "Starting order-based parallel execution with re-analysis for {} changes",
            changes.len()
        );

        let config = self.config.clone();
        let repo_root = self.repo_root.clone();
        let shared_stagger_state = self.shared_stagger_state.clone();

        // Use order-based execution
        executor
            .execute_with_order_based_reanalysis(
                changes,
                move |remaining, in_flight_ids, iteration| {
                    let config = config.clone();
                    let repo_root = repo_root.clone();
                    let event_tx = event_tx.clone();
                    let shared_stagger_state = shared_stagger_state.clone();
                    Box::pin(async move {
                        let service = ParallelRunService::new_with_shared_state(
                            repo_root,
                            config,
                            shared_stagger_state,
                        );
                        service
                            .analyze_order_with_sender(
                                remaining,
                                in_flight_ids,
                                Some(&event_tx),
                                iteration,
                            )
                            .await
                    })
                },
            )
            .await
    }

    /// Analyze changes and group them for parallel execution (public API).
    ///
    /// If `use_llm_analysis` is enabled (default), uses LLM to analyze dependencies.
    /// Otherwise, runs all changes in parallel (no dependency inference).
    pub async fn analyze_and_group_public(&self, changes: &[Change]) -> Vec<ParallelGroup> {
        self.analyze_and_group(changes).await
    }

    /// Analyze changes and group them for parallel execution.
    ///
    /// If `use_llm_analysis` is enabled (default), uses LLM to analyze dependencies.
    /// Otherwise, runs all changes in parallel (no dependency inference).
    async fn analyze_and_group(&self, changes: &[Change]) -> Vec<ParallelGroup> {
        self.analyze_and_group_with_sender(changes, None, 1).await
    }

    /// Analyze changes and return order-based result with optional event sender.
    ///
    /// If `use_llm_analysis` is enabled (default), uses LLM to analyze dependencies.
    /// Otherwise, returns changes in listed order with proposal metadata dependencies preserved.
    /// When a sender is provided, AnalysisOutput events are sent for streaming output.
    ///
    /// The returned [`crate::analyzer::AnalysisOutcome`] carries runtime-only provenance so the
    /// scheduler can tell a healthy or intentionally metadata-only result apart from a
    /// recoverable-failure fallback. Only the failure fallback uses bounded suppression, which
    /// keeps a broken analyzer command from being relaunched on every timer wake while still
    /// allowing recovery from a transient outage.
    ///
    /// # Arguments
    /// * `changes` - Changes to analyze for execution order
    /// * `in_flight_ids` - Currently executing change IDs (not selectable, but available as dependencies)
    /// * `event_tx` - Optional event sender for streaming output
    /// * `iteration` - Current iteration number
    async fn analyze_order_with_sender(
        &self,
        changes: &[Change],
        in_flight_ids: &[String],
        event_tx: Option<&mpsc::Sender<ParallelEvent>>,
        iteration: u32,
    ) -> crate::analyzer::AnalysisOutcome {
        // Check if LLM analysis is enabled (default: true)
        if self.config.use_llm_analysis() {
            info!("Using LLM analysis for parallelization (analyze_command)");
            match self
                .analyze_order_with_llm_streaming(changes, in_flight_ids, event_tx, iteration)
                .await
            {
                Ok(result) => {
                    info!(
                        "LLM analysis successful: {} changes in order",
                        result.order.len()
                    );
                    return crate::analyzer::AnalysisOutcome::healthy(result);
                }
                Err(e) => {
                    // Rejecting an LLM analysis response is recoverable: metadata-dependency-only
                    // fallback below preserves declared dependencies, so this stays warning-visible
                    // instead of being emitted as a terminal workflow error.
                    self.emit_recoverable_analysis_fallback_diagnostic_once(
                        changes,
                        in_flight_ids,
                        event_tx,
                        &e.to_string(),
                    )
                    .await;
                    return crate::analyzer::AnalysisOutcome::recoverable_failure_fallback(
                        Self::metadata_dependency_analysis_result(changes),
                    );
                }
            }
        }

        info!("LLM analysis disabled, using metadata-dependency-only analysis");
        crate::analyzer::AnalysisOutcome::intentional_metadata_only(
            Self::metadata_dependency_analysis_result(changes),
        )
    }

    fn log_recoverable_analysis_fallback(error: &dyn std::fmt::Display) {
        warn!(
            error = %error,
            "LLM analysis failed; falling back to metadata-dependency-only analysis"
        );
    }

    /// Build the deduplication key and operator-facing message for a recoverable
    /// analysis fallback.
    ///
    /// The `AnalysisFailure` key identity is intentionally unchanged so repeated
    /// equivalent fallbacks keep collapsing into a single diagnostic while a
    /// different error or queued/in-flight set stays independently visible.
    ///
    /// Crate-visible so consumer-side tests can assert against the real producer
    /// message instead of a hand-written copy that could silently drift.
    pub(crate) fn recoverable_analysis_fallback_diagnostic(
        changes: &[Change],
        in_flight_ids: &[String],
        error: &str,
    ) -> (DiagnosticDeduplicationKey, String) {
        let mut queued_ids: Vec<String> = changes.iter().map(|change| change.id.clone()).collect();
        queued_ids.sort();
        let mut in_flight = in_flight_ids.to_vec();
        in_flight.sort();
        let normalized_error = error.trim().to_string();
        let key = DiagnosticDeduplicationKey::AnalysisFailure {
            queued_ids: queued_ids.clone(),
            in_flight_ids: in_flight.clone(),
            error: normalized_error.clone(),
        };
        // The shared marker keeps operator-facing wording stable across producer and
        // tests. It is observability text only: non-fatality is carried by emitting
        // this diagnostic as a warning log event, not by its message content.
        let message = format!(
            "{}: error={}, queued={:?}, in_flight={:?}",
            crate::events::RECOVERABLE_ANALYSIS_FALLBACK_MARKER,
            normalized_error,
            queued_ids,
            in_flight
        );
        (key, message)
    }

    /// Emit the recoverable fallback diagnostic as a deduplicated warning.
    ///
    /// The tracing record and the runtime event are two representations of the same
    /// operator-facing diagnostic, so both live behind a single
    /// `DiagnosticDeduplicationKey::AnalysisFailure` decision. Emitting the tracing
    /// record ahead of the gate would leave persistent logs accumulating repetitions
    /// that the event stream already suppresses.
    ///
    /// The gate is evaluated even without an event sender so tracing-only callers get
    /// the same bounded diagnostics.
    ///
    /// Successful metadata fallback is degraded execution, not a terminal failure, so
    /// this path never emits `ParallelEvent::Error`; terminal error events remain
    /// reserved for operations that actually fail without a safe fallback.
    async fn emit_recoverable_analysis_fallback_diagnostic_once(
        &self,
        changes: &[Change],
        in_flight_ids: &[String],
        event_tx: Option<&mpsc::Sender<ParallelEvent>>,
        error: &str,
    ) {
        let (key, message) =
            Self::recoverable_analysis_fallback_diagnostic(changes, in_flight_ids, error);
        let error = error.trim().to_string();
        let mut dedup = self.diagnostic_dedup.lock().await;
        dedup
            .emit_or_suppress(
                key,
                move || async move {
                    Self::log_recoverable_analysis_fallback(&error);
                    if let Some(tx) = event_tx {
                        let _ = tx
                            .send(ParallelEvent::Log(crate::events::LogEntry::warn(&message)))
                            .await;
                    }
                },
                || {
                    debug!("Suppressing repeated analysis fallback diagnostic");
                },
            )
            .await;
    }

    fn metadata_dependency_analysis_result(changes: &[Change]) -> crate::analyzer::AnalysisResult {
        let mut dependencies = HashMap::new();
        for change in changes {
            union_metadata_dependencies(&mut dependencies, &change.id, &change.dependencies);
        }

        crate::analyzer::AnalysisResult {
            order: changes.iter().map(|c| c.id.clone()).collect(),
            dependencies,
            groups: None,
        }
    }

    /// Analyze changes and group them for parallel execution with optional event sender.
    ///
    /// If `use_llm_analysis` is enabled (default), uses LLM to analyze dependencies.
    /// Otherwise, runs all changes in parallel (no dependency inference).
    /// When a sender is provided, AnalysisOutput events are sent for streaming output.
    ///
    /// # Deprecated
    ///
    /// This method converts order-based results to group-based format.
    /// Prefer using `analyze_order_with_sender()` for order-based execution.
    async fn analyze_and_group_with_sender(
        &self,
        changes: &[Change],
        event_tx: Option<&mpsc::Sender<ParallelEvent>>,
        iteration: u32,
    ) -> Vec<ParallelGroup> {
        // Check if LLM analysis is enabled (default: true)
        if self.config.use_llm_analysis() {
            info!("Using LLM analysis for parallelization (analyze_command)");
            match self
                .analyze_with_llm_streaming(changes, event_tx, iteration)
                .await
            {
                Ok(groups) => {
                    info!("LLM analysis successful: {} groups", groups.len());
                    return groups;
                }
                Err(e) => {
                    error!("LLM analysis failed: {}", e);
                    warn!(
                        "Falling back to running all changes in parallel (no dependency analysis)"
                    );
                }
            }
        } else {
            info!("LLM analysis disabled, running all changes in parallel");
        }

        // Fall back: run all changes in a single parallel group
        Self::all_parallel(changes)
    }

    /// Create a single group with all changes (no dependencies, full parallelism)
    fn all_parallel(changes: &[Change]) -> Vec<ParallelGroup> {
        if changes.is_empty() {
            return Vec::new();
        }

        vec![ParallelGroup {
            id: 1,
            changes: changes.iter().map(|c| c.id.clone()).collect(),
            depends_on: Vec::new(),
        }]
    }

    /// Analyze changes using LLM and return raw analysis result (order + dependencies)
    ///
    /// # Arguments
    /// * `changes` - Changes to analyze for execution order
    /// * `in_flight_ids` - Currently executing change IDs (not selectable, but available as dependencies)
    /// * `event_tx` - Optional event sender for streaming output
    /// * `iteration` - Current iteration number
    async fn analyze_order_with_llm_streaming(
        &self,
        changes: &[Change],
        in_flight_ids: &[String],
        event_tx: Option<&mpsc::Sender<ParallelEvent>>,
        iteration: u32,
    ) -> Result<crate::analyzer::AnalysisResult> {
        let analyzer = ParallelizationAnalyzer::new(
            self.ai_runner.clone(),
            self.config.clone(),
            self.repo_root.clone(),
        );

        if let Some(tx) = event_tx {
            let tx = tx.clone();
            analyzer
                .analyze_with_callback(changes, in_flight_ids, move |output| {
                    let _ = tx.try_send(ParallelEvent::AnalysisOutput {
                        output: output.clone(),
                        iteration,
                    });
                })
                .await
        } else {
            analyzer.analyze_with_inflight(changes, in_flight_ids).await
        }
    }

    /// Analyze changes using LLM (analyze_command) with streaming output
    ///
    /// # Deprecated
    ///
    /// This method converts order-based results to group-based format.
    /// Prefer using `analyze_order_with_llm_streaming()` for order-based execution.
    async fn analyze_with_llm_streaming(
        &self,
        changes: &[Change],
        event_tx: Option<&mpsc::Sender<ParallelEvent>>,
        iteration: u32,
    ) -> Result<Vec<ParallelGroup>> {
        let analyzer = ParallelizationAnalyzer::new(
            self.ai_runner.clone(),
            self.config.clone(),
            self.repo_root.clone(),
        );

        if let Some(tx) = event_tx {
            let tx = tx.clone();
            analyzer
                .analyze_groups_with_callback(changes, move |output| {
                    let _ = tx.try_send(ParallelEvent::AnalysisOutput {
                        output: output.clone(),
                        iteration,
                    });
                })
                .await
        } else {
            analyzer.analyze_groups(changes).await
        }
    }
}

/// Split changes into those committed to the repository and those skipped.
///
/// A change is eligible only when it exists in the HEAD tree and has no
/// uncommitted or untracked files under `openspec/changes/<change_id>/`. Shared
/// with the executor so post-checkpoint explicit-target resolution applies the
/// same eligibility rule as start-time preparation.
pub(crate) async fn filter_committed_changes_at(
    repo_root: &std::path::Path,
    changes: Vec<Change>,
) -> Result<(Vec<Change>, Vec<String>)> {
    let committed_change_ids: HashSet<String> =
        match crate::vcs::git::commands::list_changes_in_head(repo_root).await {
            Ok(ids) => ids.into_iter().collect(),
            Err(err) => {
                warn!(
                    "Failed to load committed change snapshot; assuming all changes are committed: {}",
                    err
                );
                return Ok((changes, Vec::new()));
            }
        };

    // Get changes with uncommitted files under openspec/changes/<change_id>/
    let uncommitted_file_change_ids: HashSet<String> =
        match crate::vcs::git::commands::list_changes_with_uncommitted_files(repo_root).await {
            Ok(ids) => ids.into_iter().collect(),
            Err(err) => {
                warn!(
                    "Failed to detect uncommitted files in changes; assuming no uncommitted files: {}",
                    err
                );
                HashSet::new()
            }
        };

    let mut committed = Vec::new();
    let mut skipped = Vec::new();

    for change in changes {
        // Exclude if:
        // 1. Not in HEAD commit tree, OR
        // 2. Has uncommitted/untracked files under openspec/changes/<change_id>/
        if !committed_change_ids.contains(&change.id)
            || uncommitted_file_change_ids.contains(&change.id)
        {
            skipped.push(change.id);
        } else {
            committed.push(change);
        }
    }

    skipped.sort();
    Ok((committed, skipped))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::openspec::ProposalMetadata;
    use tempfile::TempDir;
    use tokio::process::Command;

    fn create_test_change(id: &str, dependencies: Vec<&str>) -> Change {
        Change {
            id: id.to_string(),
            completed_tasks: 0,
            total_tasks: 5,
            last_modified: "1m ago".to_string(),
            dependencies: dependencies.into_iter().map(String::from).collect(),
            metadata: ProposalMetadata::default(),
        }
    }

    fn create_test_config() -> OrchestratorConfig {
        OrchestratorConfig {
            apply_command: Some("echo apply {change_id}".to_string()),
            archive_command: Some("echo archive {change_id}".to_string()),
            analyze_command: Some("echo '{\"order\":[\"route\",\"policy\"],\"dependencies\":{\"route\":[\"ghost\"]}}'".to_string()),
            acceptance_command: Some("echo acceptance".to_string()),
            resolve_command: Some("echo resolve".to_string()),
            ..Default::default()
        }
    }

    /// Collects tracing records so tests can assert on the tracing observability
    /// surface alongside the runtime event stream.
    #[derive(Clone, Default)]
    struct CaptureLayer(std::sync::Arc<std::sync::Mutex<Vec<(tracing::Level, String)>>>);

    impl CaptureLayer {
        fn records(&self) -> Vec<(tracing::Level, String)> {
            self.0.lock().expect("capture layer mutex").clone()
        }

        fn warnings_containing(&self, needle: &str) -> usize {
            self.records()
                .iter()
                .filter(|(level, fields)| *level == tracing::Level::WARN && fields.contains(needle))
                .count()
        }
    }

    impl<S> tracing_subscriber::Layer<S> for CaptureLayer
    where
        S: tracing::Subscriber,
    {
        fn on_event(
            &self,
            event: &tracing::Event<'_>,
            _ctx: tracing_subscriber::layer::Context<'_, S>,
        ) {
            struct Visitor {
                fields: String,
            }

            impl tracing::field::Visit for Visitor {
                fn record_debug(
                    &mut self,
                    field: &tracing::field::Field,
                    value: &dyn std::fmt::Debug,
                ) {
                    self.fields
                        .push_str(&format!("{}={:?};", field.name(), value));
                }
            }

            let mut visitor = Visitor {
                fields: String::new(),
            };
            event.record(&mut visitor);
            self.0
                .lock()
                .expect("capture layer mutex")
                .push((*event.metadata().level(), visitor.fields));
        }
    }

    /// Everything that must stay alive while a capture is being collected.
    ///
    /// The exclusion guard drops last, after the subscriber guard, so this
    /// capture's WARN-only max-level hint is gone before another capture test
    /// is allowed to start.
    struct TracingCapture {
        _exclusive: tokio::sync::MutexGuard<'static, ()>,
        _subscriber: tracing::subscriber::DefaultGuard,
    }

    /// Install a WARN-scoped capturing subscriber for the current test.
    ///
    /// Serialized against every other capturing test in the crate. That hint is
    /// process-global, not thread-local: while it is held, `info!` callsites are
    /// disabled everywhere, which would blank out the records an overlapping
    /// INFO-scoped capture test is asserting on. See
    /// [`crate::test_support::tracing_capture_lock`].
    async fn capture_tracing() -> (CaptureLayer, TracingCapture) {
        use tracing_subscriber::filter::LevelFilter;
        use tracing_subscriber::layer::SubscriberExt;
        use tracing_subscriber::Layer;

        let exclusive = crate::test_support::tracing_capture_lock().lock().await;
        let capture = CaptureLayer::default();
        // WARN and above is everything these assertions inspect. Keeping the filter
        // tight also keeps tracing's process-wide max-level hint from opening up to
        // TRACE while the guard is held, which would slow unrelated concurrent tests.
        let subscriber =
            tracing_subscriber::registry().with(capture.clone().with_filter(LevelFilter::WARN));
        // `set_default` (not `with_default`) keeps the subscriber installed across
        // `.await` points inside single-threaded `#[tokio::test]` runtimes.
        let subscriber = tracing::subscriber::set_default(subscriber);
        crate::test_support::refresh_tracing_interest();
        (
            capture,
            TracingCapture {
                _exclusive: exclusive,
                _subscriber: subscriber,
            },
        )
    }

    /// Minimal config satisfying the executor's required command fields.
    fn upstream_test_config() -> OrchestratorConfig {
        OrchestratorConfig {
            apply_command: Some("echo apply {change_id}".to_string()),
            archive_command: Some("echo archive {change_id}".to_string()),
            resolve_command: Some("echo resolve".to_string()),
            ..Default::default()
        }
    }

    #[test]
    fn upstream_integration_is_absent_by_default() {
        let service = ParallelRunService::new(PathBuf::from("/tmp"), upstream_test_config());
        assert!(service.upstream_integration().is_none());

        // A disabled service leaves the executor untouched, so no upstream
        // object is constructed on the default-off path.
        let mut executor = crate::parallel::ParallelExecutor::new(
            PathBuf::from("/tmp"),
            upstream_test_config(),
            None,
        );
        service.install_upstream_integration(&mut executor);
        assert!(!executor.has_upstream_integration());
    }

    #[test]
    fn upstream_integration_propagates_to_service_and_executor() {
        let runtime = crate::upstream::UpstreamRuntime {
            config: crate::upstream::UpstreamIntegrationConfig::new("upstream", "cargo test"),
            branch: "develop".to_string(),
        };
        let mut service = ParallelRunService::new(PathBuf::from("/tmp"), upstream_test_config());
        service.set_upstream_integration(runtime.clone());

        let stored = service.upstream_integration().expect("runtime installed");
        assert_eq!(stored.config.remote, "upstream");
        assert_eq!(stored.config.verify_command, "cargo test");
        assert_eq!(stored.branch, "develop");

        let mut executor = crate::parallel::ParallelExecutor::new(
            PathBuf::from("/tmp"),
            upstream_test_config(),
            None,
        );
        service.install_upstream_integration(&mut executor);
        assert!(executor.has_upstream_integration());
    }

    #[test]
    fn post_archive_action_propagates_to_service() {
        let temp_dir = TempDir::new().unwrap();
        let mut service =
            ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        service.set_post_archive_action(PostArchiveAction::PushToRemote {
            remote: "upstream".to_string(),
        });
        assert_eq!(
            service.post_archive_action(),
            &PostArchiveAction::PushToRemote {
                remote: "upstream".to_string()
            }
        );
    }

    /// Build an executor through the shared path with nothing optional supplied,
    /// which is exactly the shape a headless CLI run asks for.
    fn headless_executor(service: &ParallelRunService) -> crate::parallel::ParallelExecutor {
        service.create_executor_with_queue_state(None, None, None, None, None, None)
    }

    /// Regression: the CLI used to assemble its own executor and never copied
    /// `post_archive_action` onto it, so a headless `--push` run silently
    /// degraded to the default merge action. The shared builder is now the only
    /// construction path, so the configured action has to arrive on the
    /// executor itself — not merely on the service that configured it.
    #[test]
    fn create_executor_with_queue_state_carries_configured_post_archive_action() {
        let temp_dir = TempDir::new().unwrap();
        let mut service =
            ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        service.set_post_archive_action(PostArchiveAction::PushToRemote {
            remote: "origin".to_string(),
        });

        let executor = headless_executor(&service);

        assert_eq!(
            executor.post_archive_action_for_test(),
            &PostArchiveAction::PushToRemote {
                remote: "origin".to_string()
            },
            "a headless run's configured push action must reach the executor"
        );
    }

    /// The default stays the default: an unconfigured service must not acquire
    /// a push action just because construction moved.
    #[test]
    fn create_executor_with_queue_state_defaults_to_merge_to_base() {
        let temp_dir = TempDir::new().unwrap();
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());

        assert_eq!(
            headless_executor(&service).post_archive_action_for_test(),
            &PostArchiveAction::MergeToBase
        );
    }

    /// Regression: the duplicated CLI assembly also dropped the run owner's
    /// graceful-stop request. The scheduler loop only honours a stop it can read
    /// through the *same* flag run control writes, so identity — not value — is
    /// the contract.
    #[test]
    fn create_executor_with_queue_state_binds_run_owner_graceful_stop_flag() {
        let temp_dir = TempDir::new().unwrap();
        let mut service =
            ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let graceful_stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
        service.set_graceful_stop_flag(graceful_stop.clone());

        let executor = headless_executor(&service);

        let bound = executor
            .graceful_stop_flag_for_test()
            .expect("a bound graceful-stop request must reach the executor");
        assert!(
            Arc::ptr_eq(bound, &graceful_stop),
            "the executor must observe the owner's own flag, not a copy"
        );
    }

    /// A service no owner bound a flag to builds an executor that observes none.
    #[test]
    fn create_executor_with_queue_state_binds_no_graceful_stop_flag_by_default() {
        let temp_dir = TempDir::new().unwrap();
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());

        assert!(headless_executor(&service)
            .graceful_stop_flag_for_test()
            .is_none());
    }

    /// Routing the CLI through the shared builder must not make a finite run
    /// persistent: only a dynamic queue — which a CLI run never supplies —
    /// keeps the scheduler alive when idle.
    #[test]
    fn create_executor_with_queue_state_keeps_finite_lifetime_without_dynamic_queue() {
        use crate::parallel::SchedulerLifetime;

        let temp_dir = TempDir::new().unwrap();
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());

        assert_eq!(
            headless_executor(&service).scheduler_lifetime_for_test(),
            SchedulerLifetime::Finite,
            "a CLI run supplies no dynamic queue and must stay finite"
        );

        let loop_based = service.create_executor_with_queue_state(
            None,
            None,
            None,
            Some(Arc::new(crate::tui::queue::DynamicQueue::new())),
            None,
            None,
        );
        assert_eq!(
            loop_based.scheduler_lifetime_for_test(),
            SchedulerLifetime::Persistent,
            "a loop-based frontend still gets the persistent lifetime"
        );
    }

    async fn init_git_repo(temp_dir: &TempDir) -> bool {
        let init_result = Command::new("git")
            .args(["init"])
            .current_dir(temp_dir.path())
            .output()
            .await;

        let init_ok = init_result
            .as_ref()
            .map(|output| output.status.success())
            .unwrap_or(false);
        if !init_ok {
            return false;
        }

        let _ = Command::new("git")
            .args(["config", "user.email", "test@example.com"])
            .current_dir(temp_dir.path())
            .output()
            .await;
        let _ = Command::new("git")
            .args(["config", "user.name", "Test User"])
            .current_dir(temp_dir.path())
            .output()
            .await;

        true
    }

    #[tokio::test]
    async fn test_filter_committed_changes_skips_uncommitted() {
        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        let base_dir = temp_dir.path().join("openspec/changes");
        std::fs::create_dir_all(base_dir.join("change-a")).unwrap();
        std::fs::write(base_dir.join("change-a/proposal.md"), "test").unwrap();

        let _ = Command::new("git")
            .args(["add", "."])
            .current_dir(temp_dir.path())
            .output()
            .await;
        let _ = Command::new("git")
            .args(["commit", "-m", "add change-a"])
            .current_dir(temp_dir.path())
            .output()
            .await;

        std::fs::create_dir_all(base_dir.join("change-b")).unwrap();
        std::fs::write(base_dir.join("change-b/proposal.md"), "test").unwrap();

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = vec![
            create_test_change("change-a", vec![]),
            create_test_change("change-b", vec![]),
        ];

        let (committed, skipped) = service
            .filter_committed_changes(changes)
            .await
            .expect("filter changes");

        let committed_ids: Vec<String> = committed.into_iter().map(|change| change.id).collect();
        assert_eq!(committed_ids, vec!["change-a".to_string()]);
        assert_eq!(skipped, vec!["change-b".to_string()]);
    }

    #[tokio::test]
    async fn test_filter_committed_changes_skips_partially_uncommitted() {
        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        let base_dir = temp_dir.path().join("openspec/changes");

        // Create and commit change-a
        std::fs::create_dir_all(base_dir.join("change-a")).unwrap();
        std::fs::write(base_dir.join("change-a/proposal.md"), "test").unwrap();

        // Create and commit change-b
        std::fs::create_dir_all(base_dir.join("change-b")).unwrap();
        std::fs::write(base_dir.join("change-b/proposal.md"), "test").unwrap();

        let _ = Command::new("git")
            .args(["add", "."])
            .current_dir(temp_dir.path())
            .output()
            .await;
        let _ = Command::new("git")
            .args(["commit", "-m", "add changes"])
            .current_dir(temp_dir.path())
            .output()
            .await;

        // Add uncommitted file to change-a
        std::fs::write(base_dir.join("change-a/tasks.md"), "new task").unwrap();

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = vec![
            create_test_change("change-a", vec![]),
            create_test_change("change-b", vec![]),
        ];

        let (committed, skipped) = service
            .filter_committed_changes(changes)
            .await
            .expect("filter changes");

        let committed_ids: Vec<String> = committed.into_iter().map(|change| change.id).collect();
        // change-a should be skipped due to uncommitted files
        assert_eq!(committed_ids, vec!["change-b".to_string()]);
        assert_eq!(skipped, vec!["change-a".to_string()]);
    }

    /// Regression test: prepare_parallel_execution must emit a ParallelStartRejected event
    /// for each batch of rejected changes so that callers can reconcile user-visible state.
    #[tokio::test]
    async fn test_prepare_parallel_execution_emits_rejection_event() {
        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        // Commit change-a, leave change-b uncommitted.
        let base_dir = temp_dir.path().join("openspec/changes");
        std::fs::create_dir_all(base_dir.join("change-a")).unwrap();
        std::fs::write(base_dir.join("change-a/proposal.md"), "test").unwrap();
        let _ = Command::new("git")
            .args(["add", "."])
            .current_dir(temp_dir.path())
            .output()
            .await;
        let _ = Command::new("git")
            .args(["commit", "-m", "add change-a"])
            .current_dir(temp_dir.path())
            .output()
            .await;
        // change-b is not committed (only on disk).
        std::fs::create_dir_all(base_dir.join("change-b")).unwrap();
        std::fs::write(base_dir.join("change-b/proposal.md"), "test").unwrap();

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = vec![
            create_test_change("change-a", vec![]),
            create_test_change("change-b", vec![]),
        ];

        let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<ParallelEvent>(32);

        let result = service
            .prepare_parallel_execution(changes, &event_tx, false)
            .await
            .expect("prepare_parallel_execution");

        // change-a should pass; change-b should be rejected.
        assert!(result.is_some(), "change-a should still be eligible");
        let committed = result.unwrap();
        assert_eq!(committed.len(), 1);
        assert_eq!(committed[0].id, "change-a");

        drop(event_tx);

        let mut got_rejection_event = false;
        while let Some(event) = event_rx.recv().await {
            if let ParallelEvent::ParallelStartRejected { change_ids, .. } = event {
                assert!(
                    change_ids.contains(&"change-b".to_string()),
                    "rejection event should include change-b"
                );
                got_rejection_event = true;
            }
        }
        assert!(
            got_rejection_event,
            "expected a ParallelStartRejected event for the uncommitted change"
        );
    }

    /// Regression: when ALL requested changes are rejected, prepare_parallel_execution must
    /// still emit the rejection event (and return None).
    #[tokio::test]
    async fn test_prepare_parallel_execution_all_rejected_emits_rejection_event() {
        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        // Make an initial commit that contains `openspec/changes` but neither change-a nor
        // change-b. This ensures `git ls-tree HEAD:openspec/changes` succeeds and returns an
        // empty list so that both requested changes are correctly identified as "not in HEAD".
        let base_dir = temp_dir.path().join("openspec/changes");
        let placeholder = base_dir.join("placeholder");
        std::fs::create_dir_all(&placeholder).unwrap();
        std::fs::write(placeholder.join("proposal.md"), "placeholder").unwrap();
        let _ = Command::new("git")
            .args(["add", "."])
            .current_dir(temp_dir.path())
            .output()
            .await;
        let _ = Command::new("git")
            .args(["commit", "-m", "initial commit"])
            .current_dir(temp_dir.path())
            .output()
            .await;

        // Add both changes as uncommitted (not in the initial commit).
        for id in &["change-a", "change-b"] {
            std::fs::create_dir_all(base_dir.join(id)).unwrap();
            std::fs::write(base_dir.join(id).join("proposal.md"), "test").unwrap();
        }

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = vec![
            create_test_change("change-a", vec![]),
            create_test_change("change-b", vec![]),
        ];

        let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<ParallelEvent>(32);

        let result = service
            .prepare_parallel_execution(changes, &event_tx, false)
            .await
            .expect("prepare_parallel_execution");

        assert!(
            result.is_none(),
            "all changes were uncommitted so result should be None"
        );

        drop(event_tx);

        let mut got_rejection_event = false;
        let mut rejected_ids: Vec<String> = Vec::new();
        while let Some(event) = event_rx.recv().await {
            if let ParallelEvent::ParallelStartRejected { change_ids, .. } = event {
                rejected_ids = change_ids;
                got_rejection_event = true;
            }
        }
        assert!(
            got_rejection_event,
            "expected a ParallelStartRejected event even when all changes are rejected"
        );
        rejected_ids.sort();
        assert_eq!(rejected_ids, vec!["change-a", "change-b"]);
    }

    /// Regression: when ALL requested changes are rejected, `run_parallel` (the callback-based
    /// public API used by CLI) must forward the ParallelStartRejected event to the caller
    /// even on the early-return path where no forwarding task has been spawned yet.
    ///
    /// Before the fix, `run_parallel` returned `Ok(())` immediately after
    /// `prepare_parallel_execution` returned `None`, silently dropping the buffered events.
    #[tokio::test]
    async fn test_prepare_parallel_execution_allows_empty_when_resolve_wait_requested() {
        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = Vec::new();
        let (event_tx, _event_rx) = tokio::sync::mpsc::channel::<ParallelEvent>(32);

        let result = service
            .prepare_parallel_execution(changes, &event_tx, true)
            .await
            .expect("prepare_parallel_execution");

        assert!(
            result.is_some(),
            "empty startup should continue when reducer-owned ResolveWait exists"
        );
        assert!(
            result.expect("checked is_some").is_empty(),
            "no committed changes should still produce an empty queue"
        );
    }

    #[tokio::test]
    async fn test_prepare_parallel_execution_empty_parallel_without_resolve_wait_is_noop() {
        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = Vec::new();
        let (event_tx, _event_rx) = tokio::sync::mpsc::channel::<ParallelEvent>(32);

        let result = service
            .prepare_parallel_execution(changes, &event_tx, false)
            .await
            .expect("prepare_parallel_execution");

        assert!(
            result.is_none(),
            "empty startup without reducer-owned ResolveWait must remain a safe no-op"
        );
    }

    #[tokio::test]
    async fn test_analyze_order_fallback_preserves_metadata_dependencies_when_llm_disabled() {
        let temp_dir = TempDir::new().expect("tempdir");
        let mut config = create_test_config();
        config.use_llm_analysis = Some(false);
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), config);
        let changes = vec![
            create_test_change("route", vec!["policy"]),
            create_test_change("policy", vec![]),
        ];

        let outcome = service
            .analyze_order_with_sender(&changes, &[], None, 1)
            .await;

        assert_eq!(
            outcome.result.order,
            vec!["route".to_string(), "policy".to_string()]
        );
        assert_eq!(
            outcome.result.dependencies.get("route"),
            Some(&vec!["policy".to_string()])
        );
        assert_eq!(
            outcome.provenance,
            crate::analyzer::AnalysisProvenance::IntentionalMetadataOnly,
            "configured metadata-only analysis is the intended result, not a failure fallback"
        );
        assert!(
            !outcome.provenance.is_degraded(),
            "intentional metadata-only analysis must not be suppressed on a degraded interval"
        );
    }

    /// A shadow dependency judge is invisible to analysis provenance.
    ///
    /// Provenance is what decides whether the scheduler treats a result as
    /// healthy or degraded, so a judge that fails, or one that answers
    /// perfectly, must both leave it exactly where the conventional analyzer put
    /// it.
    #[cfg(unix)]
    #[tokio::test]
    async fn judge_outcomes_never_change_analysis_provenance() {
        use std::os::unix::fs::PermissionsExt;

        const MODEL: &str = "test-judge-1.0.0";
        let temp_dir = TempDir::new().expect("tempdir");
        let repo = temp_dir.path();
        for change_id in ["route", "policy"] {
            let dir = repo.join("openspec/changes").join(change_id);
            std::fs::create_dir_all(&dir).expect("create change dir");
            std::fs::write(dir.join("proposal.md"), "proposal body").expect("write proposal");
        }

        // A judge that answers validly, then publishes a marker the conventional
        // analyze command waits for: "the judge completed first" is causal here,
        // not timing-dependent.
        let judge_path = repo.join("fake-judge");
        std::fs::write(
            &judge_path,
            format!(
                "#!/bin/sh\ncat > /dev/null\nprintf '%s' '{{\"model\":\"{MODEL}\",\"answers\":{{\"q_0000_0001\":{{\"type\":\"noul\",\"noul\":0.99}},\"q_0001_0000\":{{\"type\":\"noul\",\"noul\":0.99}}}},\"usage\":{{\"input_tokens\":1,\"output_tokens\":1}}}}'\n: > '{}'\n",
                repo.join("judge-done").display()
            ),
        )
        .expect("write fake judge");
        std::fs::set_permissions(&judge_path, std::fs::Permissions::from_mode(0o755))
            .expect("chmod fake judge");

        let judge_entry = |command: Vec<String>| crate::config::JudgeCommandsConfig {
            parallel_dependency: Some(crate::config::ParallelDependencyJudgeConfig {
                command,
                model: MODEL.to_string(),
                timeout_ms: Some(300_000),
                max_input_bytes: None,
                max_output_bytes: None,
                yes_threshold: None,
                mode: None,
                evaluation: None,
            }),
        };

        let cases = vec![
            (
                "valid judge",
                judge_entry(vec![judge_path.display().to_string()]),
                format!(
                    "while [ ! -f '{}' ]; do sleep 0.01; done; printf '%s' '{{\"order\":[\"route\",\"policy\"],\"dependencies\":{{}}}}'",
                    repo.join("judge-done").display()
                ),
            ),
            (
                "unavailable judge",
                judge_entry(vec![repo.join("not-installed").display().to_string()]),
                r#"printf '%s' '{"order":["route","policy"],"dependencies":{}}'"#.to_string(),
            ),
        ];

        for (label, judge_commands, analyze_command) in cases {
            let config = OrchestratorConfig {
                use_llm_analysis: Some(true),
                analyze_command: Some(analyze_command),
                judge_commands: Some(judge_commands),
                command_queue_stagger_delay_ms: Some(0),
                command_queue_max_retries: Some(0),
                ..create_test_config()
            };
            let service = ParallelRunService::new(repo.to_path_buf(), config);
            let changes = vec![
                create_test_change("route", vec![]),
                create_test_change("policy", vec![]),
            ];

            let outcome = tokio::time::timeout(
                std::time::Duration::from_secs(60),
                service.analyze_order_with_sender(&changes, &[], None, 1),
            )
            .await
            .expect("analysis must not wait for the judge's own timeout");

            assert_eq!(
                outcome.provenance,
                crate::analyzer::AnalysisProvenance::HealthyLlm,
                "{label} must leave the conventional analyzer's provenance untouched"
            );
            assert!(
                !outcome.provenance.is_degraded(),
                "{label} must not degrade the analysis"
            );
            assert_eq!(
                outcome.result.order,
                vec!["route".to_string(), "policy".to_string()],
                "{label} must not reorder the authoritative result"
            );
            assert!(
                outcome.result.dependencies.is_empty(),
                "{label}: a judge `yes` never adds an authoritative dependency"
            );
        }
    }

    #[tokio::test]
    async fn test_analyze_order_recoverable_fallback_preserves_metadata_dependencies() {
        let temp_dir = TempDir::new().expect("tempdir");
        let mut config = create_test_config();
        config.use_llm_analysis = Some(true);
        config.analyze_command = Some("printf 'not json'".to_string());
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), config);
        let changes = vec![
            create_test_change("route", vec!["policy"]),
            create_test_change("policy", vec![]),
        ];

        let outcome = service
            .analyze_order_with_sender(&changes, &[], None, 1)
            .await;

        assert_eq!(
            outcome.result.order,
            vec!["route".to_string(), "policy".to_string()]
        );
        assert_eq!(
            outcome.result.dependencies.get("route"),
            Some(&vec!["policy".to_string()])
        );
        assert!(
            !outcome.result.dependencies.is_empty(),
            "recoverable fallback must not degrade to dependency-free analysis"
        );
        assert_eq!(
            outcome.provenance,
            crate::analyzer::AnalysisProvenance::RecoverableFailureFallback,
            "a failed LLM command must be distinguishable from a configured metadata-only result"
        );
        assert!(
            outcome.provenance.is_degraded(),
            "recoverable-failure fallback must use bounded suppression so one retry stays possible"
        );
    }

    #[tokio::test]
    async fn test_recoverable_analysis_fallback_emits_warning_without_terminal_error() {
        let temp_dir = TempDir::new().expect("tempdir");
        let mut config = create_test_config();
        config.use_llm_analysis = Some(true);
        // Reproduce the production case: the LLM omits a queued change ID, so the
        // response is rejected and metadata fallback takes over.
        config.analyze_command =
            Some("echo '{\"order\":[\"route\",\"policy\"],\"dependencies\":{}}'".to_string());
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), config);
        let changes = vec![
            create_test_change("route", vec!["policy"]),
            create_test_change("policy", vec![]),
            create_test_change("gateway", vec![]),
        ];
        let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<ParallelEvent>(32);

        let outcome = service
            .analyze_order_with_sender(&changes, &[], Some(&event_tx), 1)
            .await;

        assert_eq!(
            outcome.result.order,
            vec![
                "route".to_string(),
                "policy".to_string(),
                "gateway".to_string()
            ],
            "fallback must represent every queued change exactly once"
        );
        assert_eq!(
            outcome.result.dependencies.get("route"),
            Some(&vec!["policy".to_string()]),
            "declared metadata dependencies must survive fallback"
        );
        assert_eq!(
            outcome.provenance,
            crate::analyzer::AnalysisProvenance::RecoverableFailureFallback,
            "a rejected LLM response is a recoverable failure, not healthy analyzer output"
        );

        drop(event_tx);
        let mut warnings = Vec::new();
        let mut errors = Vec::new();
        while let Some(event) = event_rx.recv().await {
            match event {
                ParallelEvent::Log(entry) if entry.level == crate::events::LogLevel::Warn => {
                    warnings.push(entry.message)
                }
                ParallelEvent::Error { message } => errors.push(message),
                _ => {}
            }
        }

        assert!(
            errors.is_empty(),
            "successful metadata fallback must not emit a terminal error event: {errors:?}"
        );
        assert_eq!(
            warnings.len(),
            1,
            "successful metadata fallback should emit exactly one warning: {warnings:?}"
        );
        assert!(
            warnings[0].contains("metadata-dependency-only"),
            "warning must name the fallback mode: {}",
            warnings[0]
        );
        assert!(
            warnings[0].contains("Missing change IDs in response"),
            "warning must preserve the original analysis rejection reason: {}",
            warnings[0]
        );
    }

    #[test]
    fn test_recoverable_analysis_fallback_diagnostic_message_names_fallback_mode() {
        let changes = vec![
            create_test_change("route", vec!["policy"]),
            create_test_change("policy", vec![]),
        ];

        let (key, message) = ParallelRunService::recoverable_analysis_fallback_diagnostic(
            &changes,
            &["beta".to_string(), "alpha".to_string()],
            "  Missing change IDs in response: [\"gateway\"]  ",
        );

        assert!(
            matches!(
                key,
                DiagnosticDeduplicationKey::AnalysisFailure {
                    ref queued_ids,
                    ref in_flight_ids,
                    ref error,
                } if queued_ids == &["policy".to_string(), "route".to_string()]
                    && in_flight_ids == &["alpha".to_string(), "beta".to_string()]
                    && error == "Missing change IDs in response: [\"gateway\"]"
            ),
            "dedup identity must stay stable and order-independent: {key:?}"
        );
        assert!(
            message.contains("metadata-dependency-only"),
            "message must name the fallback mode: {message}"
        );
        assert!(
            message.contains("Missing change IDs in response"),
            "message must preserve the original reason: {message}"
        );
        assert!(
            !message.contains("Dependency analysis failed"),
            "recoverable fallback must not be phrased as a terminal failure: {message}"
        );
    }

    #[tokio::test]
    async fn test_recoverable_analysis_fallback_diagnostic_dedupes_by_signature() {
        let temp_dir = TempDir::new().expect("tempdir");
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = vec![create_test_change("route", vec!["policy"])];
        let other_changes = vec![
            create_test_change("route", vec!["policy"]),
            create_test_change("gateway", vec![]),
        ];
        let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<ParallelEvent>(32);
        let (capture, _tracing_guard) = capture_tracing().await;

        for _ in 0..2 {
            service
                .emit_recoverable_analysis_fallback_diagnostic_once(
                    &changes,
                    &[],
                    Some(&event_tx),
                    "Missing change IDs in response: [\"gateway\"]",
                )
                .await;
        }
        // A materially different error must remain independently visible.
        service
            .emit_recoverable_analysis_fallback_diagnostic_once(
                &changes,
                &[],
                Some(&event_tx),
                "Duplicate change ID in order: route",
            )
            .await;
        // So must a changed queued/in-flight context for the same error.
        service
            .emit_recoverable_analysis_fallback_diagnostic_once(
                &other_changes,
                &["alpha".to_string()],
                Some(&event_tx),
                "Missing change IDs in response: [\"gateway\"]",
            )
            .await;

        drop(event_tx);
        let mut warnings = Vec::new();
        let mut errors = Vec::new();
        while let Some(event) = event_rx.recv().await {
            match event {
                ParallelEvent::Log(entry) if entry.level == crate::events::LogLevel::Warn => {
                    warnings.push(entry.message)
                }
                ParallelEvent::Error { message } => errors.push(message),
                _ => {}
            }
        }

        assert!(
            errors.is_empty(),
            "fallback diagnostics must never emit terminal error events: {errors:?}"
        );
        assert_eq!(
            warnings.len(),
            3,
            "equivalent diagnostics collapse to one while distinct contexts stay visible: {warnings:?}"
        );
        assert_eq!(
            warnings
                .iter()
                .filter(|message| message.contains("Duplicate change ID in order: route"))
                .count(),
            1,
            "a different error must remain visible: {warnings:?}"
        );
        assert_eq!(
            warnings
                .iter()
                .filter(|message| message.contains("\"gateway\", \"route\""))
                .count(),
            1,
            "a different queued set must remain visible: {warnings:?}"
        );

        // The tracing surface must follow the same deduplication decision as the
        // runtime event stream: one record per distinct signature, never more.
        let records = capture.records();
        assert!(
            records
                .iter()
                .all(|(level, _)| *level != tracing::Level::ERROR),
            "recoverable fallback must not emit ERROR-level tracing records: {records:?}"
        );
        assert_eq!(
            records
                .iter()
                .filter(|(level, _)| *level == tracing::Level::WARN)
                .count(),
            3,
            "tracing records must collapse equivalent signatures exactly like runtime events: {records:?}"
        );
        assert_eq!(
            capture.warnings_containing("Missing change IDs in response"),
            2,
            "two repeats of one signature plus one changed queued context: {records:?}"
        );
        assert_eq!(
            capture.warnings_containing("Duplicate change ID in order: route"),
            1,
            "a different rejection reason must emit its own tracing record: {records:?}"
        );
    }

    #[tokio::test]
    async fn test_recoverable_analysis_fallback_dedupes_tracing_without_event_sender() {
        let temp_dir = TempDir::new().expect("tempdir");
        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = vec![create_test_change("route", vec!["policy"])];
        let (capture, _tracing_guard) = capture_tracing().await;

        for _ in 0..3 {
            service
                .emit_recoverable_analysis_fallback_diagnostic_once(
                    &changes,
                    &[],
                    None,
                    "Missing change IDs in response: [\"gateway\"]",
                )
                .await;
        }

        let records = capture.records();
        assert_eq!(
            records
                .iter()
                .filter(|(level, _)| *level == tracing::Level::WARN)
                .count(),
            1,
            "tracing-only callers must still get exactly one record per signature: {records:?}"
        );
    }

    // `async` only so the capture can take the process-wide tracing exclusion
    // lock; the assertions themselves are synchronous.
    #[tokio::test]
    async fn test_recoverable_fallback_log_uses_warn_level_only() {
        use tracing::Level;

        let (capture, guard) = capture_tracing().await;
        ParallelRunService::log_recoverable_analysis_fallback(&"invalid dependency graph");
        drop(guard);

        let events = capture.records();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].0, Level::WARN);
        assert!(
            events[0]
                .1
                .contains("falling back to metadata-dependency-only analysis"),
            "fallback diagnostic should remain operator-visible"
        );
        assert!(
            events[0].1.contains("invalid dependency graph"),
            "original LLM analysis failure should remain visible as warning context"
        );
        assert!(
            events.iter().all(|(level, _)| *level != Level::ERROR),
            "recoverable fallback must not emit ERROR-level records"
        );
    }

    #[tokio::test]
    async fn test_order_based_empty_resolve_wait_shared_state_enters_scheduler_path() {
        use crate::orchestration::state::{
            OrchestratorState, ReducerCommand, WorkspaceObservation,
        };
        use crate::parallel::ParallelExecutor;
        use std::sync::Arc;
        use tokio::sync::RwLock;

        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        let _ = Command::new("git")
            .args(["commit", "--allow-empty", "-m", "initial commit"])
            .current_dir(temp_dir.path())
            .output()
            .await;

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let (event_tx, mut event_rx) = tokio::sync::mpsc::channel::<ParallelEvent>(32);
        let shared = Arc::new(RwLock::new(OrchestratorState::new(
            vec!["alpha".to_string()],
            3,
        )));
        {
            let mut state = shared.write().await;
            state.apply_observation("alpha", WorkspaceObservation::WorkspaceArchived);
            state.apply_command(ReducerCommand::ResolveMerge("alpha".to_string()));
        }

        let mut executor = ParallelExecutor::new(
            temp_dir.path().to_path_buf(),
            create_test_config(),
            Some(event_tx.clone()),
        );
        executor.set_shared_orchestrator_state(shared.clone());

        service
            .run_parallel_order_based_with_executor(executor, Vec::new(), event_tx.clone())
            .await
            .expect("empty ResolveWait scheduler path should run");

        drop(event_tx);
        let mut rejected_empty_start = false;
        while let Some(event) = event_rx.recv().await {
            if matches!(event, ParallelEvent::ParallelStartRejected { .. }) {
                rejected_empty_start = true;
            }
        }
        assert!(
            !rejected_empty_start,
            "empty ResolveWait startup must not be treated as a zero-change start rejection"
        );
    }

    #[tokio::test]
    async fn test_run_parallel_all_rejected_forwards_event_to_callback() {
        let temp_dir = TempDir::new().expect("tempdir");
        if !init_git_repo(&temp_dir).await {
            return;
        }

        // Make an initial commit that does not contain change-a or change-b.
        let base_dir = temp_dir.path().join("openspec/changes");
        let placeholder = base_dir.join("placeholder");
        std::fs::create_dir_all(&placeholder).unwrap();
        std::fs::write(placeholder.join("proposal.md"), "placeholder").unwrap();
        let _ = Command::new("git")
            .args(["add", "."])
            .current_dir(temp_dir.path())
            .output()
            .await;
        let _ = Command::new("git")
            .args(["commit", "-m", "initial commit"])
            .current_dir(temp_dir.path())
            .output()
            .await;

        // Add both changes as uncommitted (not in HEAD).
        for id in &["change-a", "change-b"] {
            std::fs::create_dir_all(base_dir.join(id)).unwrap();
            std::fs::write(base_dir.join(id).join("proposal.md"), "test").unwrap();
        }

        let service = ParallelRunService::new(temp_dir.path().to_path_buf(), create_test_config());
        let changes = vec![
            create_test_change("change-a", vec![]),
            create_test_change("change-b", vec![]),
        ];

        let collected_events: std::sync::Arc<std::sync::Mutex<Vec<ParallelEvent>>> =
            std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
        let collected_events_clone = collected_events.clone();

        service
            .run_parallel(changes, None, move |event| {
                collected_events_clone.lock().unwrap().push(event);
            })
            .await
            .expect("run_parallel should succeed even when all changes are rejected");

        let events = collected_events.lock().unwrap();
        let got_rejection = events
            .iter()
            .any(|e| matches!(e, ParallelEvent::ParallelStartRejected { .. }));
        assert!(
            got_rejection,
            "ParallelStartRejected must be forwarded to the callback when all changes are rejected at start time"
        );
    }

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

    fn per_change_upstream_git(cwd: &std::path::Path, args: &[&str]) -> Option<String> {
        let output = std::process::Command::new("git")
            .args(args)
            .current_dir(cwd)
            .output()
            .ok()?;
        output
            .status
            .success()
            .then(|| String::from_utf8_lossy(&output.stdout).trim().to_string())
    }

    /// A repository on `main` with a local bare remote, an archived change, and
    /// a cumulative-base integration that records required publication but was
    /// never pushed.
    fn per_change_upstream_unpublished_repo() -> Option<(TempDir, PathBuf)> {
        let dir = TempDir::new().ok()?;
        let root = dir.path().join("repo");
        let remote = dir.path().join("remote.git");
        std::fs::create_dir_all(&root).ok()?;
        per_change_upstream_git(dir.path(), &["init", "--bare", "-b", "main", "remote.git"])?;
        per_change_upstream_git(&root, &["init", "-b", "main"])?;
        per_change_upstream_git(&root, &["config", "user.email", "test@example.com"])?;
        per_change_upstream_git(&root, &["config", "user.name", "Test User"])?;
        per_change_upstream_git(&root, &["config", "commit.gpgsign", "false"])?;
        std::fs::write(root.join("README.md"), "# base\n").ok()?;
        per_change_upstream_git(&root, &["add", "-A"])?;
        per_change_upstream_git(&root, &["commit", "-m", "Initial commit"])?;
        per_change_upstream_git(&root, &["remote", "add", "origin", remote.to_str()?])?;
        per_change_upstream_git(&root, &["push", "-u", "origin", "main"])?;

        let archive = root.join("openspec/changes/archive/alpha");
        std::fs::create_dir_all(&archive).ok()?;
        std::fs::write(archive.join("proposal.md"), "# archived alpha\n").ok()?;
        per_change_upstream_git(&root, &["add", "-A"])?;
        per_change_upstream_git(&root, &["commit", "-m", "Archive: alpha"])?;
        let marker = crate::upstream::publication::format_publication_marker_message(
            "alpha", "origin", "main",
        );
        per_change_upstream_git(&root, &["commit", "--allow-empty", "-m", &marker])?;
        Some((dir, root))
    }

    /// Run one finite opted-in run over a repository whose only outstanding work
    /// is publication, and return the events it emitted in order.
    async fn per_change_upstream_finite_run(
        root: &std::path::Path,
        verify_command: &str,
    ) -> Vec<ParallelEvent> {
        let mut service = ParallelRunService::new(root.to_path_buf(), upstream_test_config());
        service.set_upstream_integration(crate::upstream::UpstreamRuntime {
            config: crate::upstream::UpstreamIntegrationConfig::new("origin", verify_command),
            branch: "main".to_string(),
        });

        // No dynamic queue: this is a finite run, so the scheduler drains and
        // the run must decide completion for itself.
        let (event_tx, mut event_rx) = mpsc::channel::<ParallelEvent>(256);
        let collector = tokio::spawn(async move {
            let mut events = Vec::new();
            while let Some(event) = event_rx.recv().await {
                events.push(event);
            }
            events
        });

        service
            .run_parallel_with_channel_and_queue_state(
                Vec::new(),
                event_tx,
                None,
                None,
                None,
                None,
                None,
                true,
            )
            .await
            .expect("finite opted-in run");
        collector.await.expect("event collector")
    }

    /// A finite run may only report completion after the remote has confirmed
    /// every targeted change. Ordering — not just presence — is the contract.
    #[tokio::test]
    async fn per_change_upstream_finite_run_completes_only_after_confirmation() {
        // The run takes the process-global merge lock for its base lane, so it
        // must hold the base-lane test mutex (see `crate::parallel`) or another
        // base-lane test's lock makes this one observe a busy lane.
        let _serialize = crate::parallel::merge_lock_test_mutex().lock().await;
        let Some((_dir, root)) = per_change_upstream_unpublished_repo() else {
            println!("Skipping test: git not available");
            return;
        };
        let head = per_change_upstream_git(&root, &["rev-parse", "HEAD"]).expect("head");

        let events = per_change_upstream_finite_run(&root, "exit 0").await;

        let pushed = events
            .iter()
            .position(|event| {
                matches!(event, ParallelEvent::PushCompleted { change_id, .. } if change_id == "alpha")
            })
            .expect("the targeted change must reach confirmed publication");
        let completed = events
            .iter()
            .position(|event| matches!(event, ParallelEvent::AllCompleted))
            .expect("a successful finite run reports completion");
        assert!(
            pushed < completed,
            "AllCompleted must follow remote confirmation, never precede it"
        );
        assert_eq!(
            per_change_upstream_git(&root, &["ls-remote", "origin", "refs/heads/main"])
                .expect("ls-remote")
                .split_whitespace()
                .next()
                .expect("remote head"),
            head,
            "completion is reported against a remotely observed cumulative HEAD"
        );
    }

    /// A blocked publication is not a slow success: the run reports no
    /// completion at all, and the change stays resumable.
    #[tokio::test]
    async fn per_change_upstream_finite_run_withholds_completion_when_publication_fails() {
        let _serialize = crate::parallel::merge_lock_test_mutex().lock().await;
        let Some((_dir, root)) = per_change_upstream_unpublished_repo() else {
            println!("Skipping test: git not available");
            return;
        };
        let remote_before =
            per_change_upstream_git(&root, &["ls-remote", "origin", "refs/heads/main"])
                .expect("ls-remote");

        let events = per_change_upstream_finite_run(&root, "exit 1").await;

        assert!(
            !events
                .iter()
                .any(|event| matches!(event, ParallelEvent::AllCompleted)),
            "a finite run with an unpublished change must not report completion"
        );
        assert!(
            !events.iter().any(|event| matches!(
                event,
                ParallelEvent::PushCompleted { change_id, .. } if change_id == "alpha"
            )),
            "failed verification must suppress confirmed publication"
        );
        assert_eq!(
            per_change_upstream_git(&root, &["ls-remote", "origin", "refs/heads/main"])
                .expect("ls-remote"),
            remote_before,
            "nothing may reach the remote when verification fails"
        );
    }
}