cflx 0.6.327

//! TUI runner and main event loop
//!
//! Contains run_tui and run_tui_loop functions.

use crate::config::OrchestratorConfig;
use crate::error::Result;
use crate::lifecycle_integration::LifecycleHandle;
use crate::openspec::Change;
use crate::parallel::PostArchiveAction;
use crate::vcs::{GitWorkspaceManager, WorkspaceManager};
use crossterm::event::{self, Event, KeyEventKind, MouseEventKind};
use ratatui::DefaultTerminal;
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicBool;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, info, warn};

use super::command_handlers::{handle_tui_command, TuiCommandContext};
use super::events::{LogEntry, OrchestratorEvent, TuiCommand, TuiRefreshObservation};
use super::key_handlers::{handle_key_event, KeyEventContext};
use super::lifecycle::TuiLifecycleSnapshot;
use super::log_deduplicator;
// orchestrator functions now called from command_handlers
use super::queue::DynamicQueue;
use super::render::{render, SPINNER_CHARS};
use super::state::{AppState, AUTO_REFRESH_INTERVAL_SECS};
// AppExecutionMode/StopMode are used in handlers
use super::terminal::restore_terminal;
use super::worktrees::load_worktrees_with_conflict_check;

fn is_refresh_root_usable(repo_root: &Path) -> bool {
    repo_root.is_dir()
}

fn should_skip_local_refresh(repo_root: &Path, stale_refresh_root_warned: &mut bool) -> bool {
    let refresh_root_usable = is_refresh_root_usable(repo_root);
    if !refresh_root_usable {
        if !*stale_refresh_root_warned {
            *stale_refresh_root_warned = true;
            warn!(
                repo_root = %repo_root.display(),
                "Skipping local TUI auto-refresh: stale or missing refresh root"
            );
        }
        return true;
    }

    *stale_refresh_root_warned = false;
    false
}

/// Observe workspace dirty state for one auto-refresh tick.
///
/// The predicate is the shared read-only one
/// ([`crate::vcs::git::commands::has_uncommitted_changes`]), so the header
/// counts staged, unstaged, and untracked work and excludes ignored files under
/// the same explicit untracked/ignored modes every other dirty-state consumer
/// gets — repository or user `status.showUntrackedFiles` configuration cannot
/// make untracked work report as clean here either.
///
/// A failed read returns `None` rather than a value. There is no third answer to
/// publish: the caller keeps its last successful observation instead of turning
/// an unobservable workspace into a clean one. The warning is bounded the same
/// way the stale-root warning above is — once per failure episode, cleared by
/// the next success — so a persistently unreadable repository cannot fill the
/// log at the refresh cadence.
async fn observe_workspace_dirty(
    repo_root: &Path,
    dirty_observation_warned: &mut bool,
) -> Option<TuiRefreshObservation> {
    match crate::vcs::git::commands::has_uncommitted_changes(repo_root).await {
        Ok((dirty, _status)) => {
            *dirty_observation_warned = false;
            Some(TuiRefreshObservation::WorkspaceDirty { dirty })
        }
        Err(err) => {
            if !*dirty_observation_warned {
                *dirty_observation_warned = true;
                warn!(
                    repo_root = %repo_root.display(),
                    error = %err,
                    "Failed to observe workspace dirty state: keeping the last successful observation"
                );
            }
            None
        }
    }
}

fn refresh_local_changes(repo_root: &Path) -> Result<(Vec<Change>, Vec<Change>)> {
    let active_changes = crate::openspec::list_changes_native_from(repo_root)?;
    let rejected_changes = crate::openspec::list_rejected_changes_native_from(repo_root)?;
    Ok((active_changes, rejected_changes))
}

/// Whether this event can change the reducer-derived display caches the TUI
/// renders from.
///
/// It no longer decides whether the reducer is *written* — the dispatch owner
/// does that, exactly once, before the event reaches this frontend. It only
/// decides whether re-reading the reducer afterwards could show anything new,
/// so a chatty output event does not cost a lock and two map rebuilds per chunk.
fn should_apply_event_to_tui_reducer(event: &crate::events::ExecutionEvent) -> bool {
    use crate::events::ExecutionEvent;

    match event {
        // Reducer-visible lifecycle and workspace observations that derive TUI display status,
        // queue intent, active counts, wait states, or terminal state.
        ExecutionEvent::ProcessingStarted(_)
        | ExecutionEvent::ProcessingError { .. }
        | ExecutionEvent::ApplyStarted { .. }
        | ExecutionEvent::ApplyCompleted { .. }
        | ExecutionEvent::ApplyFailed { .. }
        | ExecutionEvent::ArchiveStarted { .. }
        | ExecutionEvent::ArchiveResumed { .. }
        | ExecutionEvent::ArchiveRetryScheduled { .. }
        | ExecutionEvent::ChangeArchived(_)
        | ExecutionEvent::ArchiveFailed { .. }
        | ExecutionEvent::AcceptanceStarted { .. }
        | ExecutionEvent::AcceptanceCompleted { .. }
        | ExecutionEvent::AcceptanceFailed { .. }
        | ExecutionEvent::ChangeRejected { .. }
        | ExecutionEvent::RejectionReviewCompleted { .. }
        | ExecutionEvent::RejectionReviewFailed { .. }
        | ExecutionEvent::WorkspaceStatusUpdated { .. }
        | ExecutionEvent::WorkspacePreparationStarted { .. }
        | ExecutionEvent::WorkspacePreparationEnded { .. }
        | ExecutionEvent::PushStarted { .. }
        | ExecutionEvent::PushCompleted { .. }
        | ExecutionEvent::PushFailed { .. }
        | ExecutionEvent::MergeCompleted { .. }
        | ExecutionEvent::MergeDeferred { .. }
        | ExecutionEvent::ResolveStarted { .. }
        | ExecutionEvent::ResolveCompleted { .. }
        | ExecutionEvent::ResolveFailed { .. }
        | ExecutionEvent::DependencyBlocked { .. }
        | ExecutionEvent::DependencyResolved { .. }
        | ExecutionEvent::AcceptanceGated { .. }
        | ExecutionEvent::ExecutionBlocked { .. }
        | ExecutionEvent::ChangeDequeued { .. }
        | ExecutionEvent::ChangeStopped { .. }
        | ExecutionEvent::ChangesRefreshed { .. } => true,

        // A process-level stop is a reducer-owned run boundary: it returns every
        // interrupted row to `not queued` and guards it against late lifecycle
        // events. The TUI reads that one transition instead of maintaining a
        // second string-matching reset that a new status variant could miss.
        ExecutionEvent::Stopped => true,

        // An accepted operator command commits reducer queue intent, retry
        // edges, and resolve intent before it publishes this event, so the row
        // statuses it produced have to be re-read here. Not re-reading them is
        // exactly how a remote queue or retry command used to reach `/api/v2`
        // without ever reaching the next TUI render.
        ExecutionEvent::OperatorCommandApplied { .. } => true,

        // The reducer holds the ephemeral commit subphase, so the TUI must
        // re-read it to switch a row's rendered operation between apply and
        // commit. It never changes the canonical `applying` display status.
        ExecutionEvent::ApplyCommitPhase { .. } => true,

        // Presentation-only or unrelated TUI events do not affect reducer display state.
        ExecutionEvent::ApplyCommitOutput { .. }
        | ExecutionEvent::ApplyOutput { .. }
        | ExecutionEvent::ArchiveOutput { .. }
        | ExecutionEvent::AcceptanceOutput { .. }
        | ExecutionEvent::ProgressUpdated { .. }
        | ExecutionEvent::WorkspaceCreated { .. }
        | ExecutionEvent::WorkspaceResumed { .. }
        | ExecutionEvent::WorkspacePreserved { .. }
        | ExecutionEvent::CleanupStarted { .. }
        | ExecutionEvent::CleanupCompleted { .. }
        | ExecutionEvent::MergeStarted { .. }
        | ExecutionEvent::MergeConflict { .. }
        | ExecutionEvent::ConflictResolutionStarted
        | ExecutionEvent::ConflictResolutionCompleted
        | ExecutionEvent::ConflictResolutionFailed { .. }
        | ExecutionEvent::ChangeSkipped { .. }
        | ExecutionEvent::AnalysisStarted { .. }
        | ExecutionEvent::AnalysisOutput { .. }
        | ExecutionEvent::AnalysisCompleted { .. }
        | ExecutionEvent::ResolveOutput { .. }
        | ExecutionEvent::HookStarted { .. }
        | ExecutionEvent::HookCompleted { .. }
        | ExecutionEvent::HookFailed { .. }
        | ExecutionEvent::Warning { .. }
        | ExecutionEvent::ParallelStartRejected { .. }
        | ExecutionEvent::Log(_)
        | ExecutionEvent::Stopping
        | ExecutionEvent::AllCompleted
        // The idle transition owns no reducer mutation at all: rows keep their
        // status, blockers, queue intent, and worktree evidence across it, so
        // re-reading the reducer here could only cost a lock for no new fact.
        | ExecutionEvent::PersistentSchedulerIdle
        | ExecutionEvent::Error { .. }
        | ExecutionEvent::WorktreesRefreshed { .. }
        | ExecutionEvent::BranchMergeStarted { .. }
        | ExecutionEvent::BranchMergeCompleted { .. }
        | ExecutionEvent::BranchMergeFailed { .. }
        | ExecutionEvent::ChangeStopFailed { .. } => false,
    }
}

/// Apply the mark writes an operator interaction requested, then mirror the rows.
///
/// The write is deferred out of key handling for one reason: it has to take the
/// same async operator mutation guard the authoritative dispatcher takes for
/// mark reconciliation. Applying it here — one requested change at a time —
/// keeps an interaction from ever replacing the shared store from this
/// frontend's cached row set, so a mark a concurrent event revoked stays
/// revoked and a mark another frontend set stays set.
pub(crate) async fn apply_pending_mark_writes(
    app: &mut AppState,
    service: &Arc<crate::orchestration::operator_command::OperatorCommandService>,
) {
    let pending = app.take_pending_mark_writes();
    if pending.is_empty() {
        return;
    }
    for (change_id, marked) in pending {
        service.apply_execution_mark(&change_id, marked).await;
    }
    app.sync_execution_marks_from_store();
}

/// Refresh the TUI's reducer-derived display caches after one event.
///
/// The reducer is **read**, never written. By the time an event reaches this
/// frontend its dispatch owner has already applied it, so a write here would be
/// the second transition for one internal event: an apply count that advances
/// twice, a change that leaves the queue twice, a terminal state reached twice.
///
/// Status, blocker view, and error detail come from the same snapshot so a row's
/// `blocked`/`stalled` word, its blocker kind, and its final diagnostic can never
/// disagree.
pub(crate) async fn sync_reducer_display_caches(
    app: &mut AppState,
    shared_state: &Arc<tokio::sync::RwLock<crate::orchestration::state::OrchestratorState>>,
    event: &crate::events::ExecutionEvent,
) {
    if !should_apply_event_to_tui_reducer(event) {
        return;
    }
    let (display_map, archived, blocker_views, error_details, apply_operations) = {
        let state = shared_state.read().await;
        (
            state.all_display_statuses(),
            // Same read as the statuses: after `ChangeArchived` the reducer keeps
            // the archive milestone here while truthfully advancing the row to
            // `resolving` or `resolve pending`, so presentation needs both facts
            // from one instant to hide the checkbox without inferring it from the
            // status string.
            state.archived_changes().clone(),
            state.all_blocker_views(),
            state.all_error_details(),
            state.all_apply_operation_labels(),
        )
    };
    app.apply_display_statuses_from_reducer(&display_map);
    app.apply_archived_changes_from_reducer(&archived);
    app.apply_blocker_views_from_reducer(&blocker_views);
    // Rendering-only refresh: the reducer's ephemeral commit subphase decides
    // whether the Apply lane prints `[apply]` or `[commit]`.
    app.apply_operation_labels_from_reducer(&apply_operations);
    // Error rows adopt the reducer's retained diagnostic even when this frontend
    // never observed the failure event itself, so the row can still name its
    // failure after the matching log entry has been evicted.
    app.apply_error_details_from_reducer(&error_details);
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LocalOrchestratorShutdownOutcome {
    NoTask,
    AlreadyFinished,
    Graceful,
    AbortedAfterTimeout,
}

/// How long local TUI quit waits for the orchestrator before escalating.
///
/// It must not undercut the scheduler's own cancellation boundary: the run
/// needs that whole window to finish command cleanup and pending merge/base-lane
/// handling. The extra margin covers the scheduler's own return path after its
/// barrier completes, so a run that used its full budget still gets to report
/// gracefully instead of being aborted at the exact same instant.
pub const LOCAL_ORCHESTRATOR_SHUTDOWN_GRACE: Duration = Duration::from_secs(
    crate::tui::orchestrator::PARALLEL_CANCELLATION_CLEANUP_DEADLINE.as_secs() + 5,
);

/// Bounded budget for the supervisor's own forceful cleanup after a timeout.
const LOCAL_SHUTDOWN_FORCE_CLEANUP_BUDGET: Duration = Duration::from_secs(10);

pub async fn shutdown_local_orchestrator_task(
    orchestrator_handle: Option<tokio::task::JoinHandle<Result<()>>>,
    orchestrator_cancel: Option<CancellationToken>,
    run_command_scope: Option<crate::ai_command_runner::RunCommandScope>,
    grace_period: Duration,
) -> LocalOrchestratorShutdownOutcome {
    // Admission closes at cancellation time, not after the grace period: a
    // command must not be able to start while the TUI is already leaving.
    if let Some(scope) = &run_command_scope {
        scope.close();
    }

    if let Some(cancel) = orchestrator_cancel {
        info!(
            grace_ms = grace_period.as_millis(),
            "Cancelling local TUI orchestrator during shutdown"
        );
        cancel.cancel();
    }

    let Some(handle) = orchestrator_handle else {
        debug!("No local TUI orchestrator task to shut down");
        return LocalOrchestratorShutdownOutcome::NoTask;
    };

    if handle.is_finished() {
        let _ = handle.await;
        info!("Local TUI orchestrator task was already finished during shutdown");
        return LocalOrchestratorShutdownOutcome::AlreadyFinished;
    }

    tokio::pin!(handle);
    tokio::select! {
        join_result = &mut handle => {
            match join_result {
                Ok(Ok(())) => info!("Local TUI orchestrator task finished gracefully during shutdown"),
                Ok(Err(err)) => warn!(error = %err, "Local TUI orchestrator task exited with error during shutdown"),
                Err(err) => warn!(error = %err, "Local TUI orchestrator task join failed during shutdown"),
            }
            LocalOrchestratorShutdownOutcome::Graceful
        }
        _ = tokio::time::sleep(grace_period) => {
            warn!(
                grace_ms = grace_period.as_millis(),
                "Local TUI orchestrator did not finish before shutdown grace period; aborting task to prevent detached local work"
            );
            // Task abort is not child-process cleanup evidence, and it destroys
            // the only in-task path to the run's PGIDs. The retained scope is
            // consumed here, before the abort, so deterministic groups are
            // force-cleaned and verified while their identities still exist.
            if let Some(scope) = &run_command_scope {
                let cleanup = scope
                    .force_cleanup_retained(LOCAL_SHUTDOWN_FORCE_CLEANUP_BUDGET)
                    .await;
                if cleanup.is_quiescent() {
                    info!(
                        escalated = cleanup.escalated,
                        "Forced cleanup of retained run-owned process groups before aborting the local TUI orchestrator"
                    );
                } else {
                    warn!(
                        "Local TUI shutdown could not prove run-owned process cleanup: {}",
                        cleanup.diagnostics()
                    );
                }
            }
            handle.as_ref().abort_handle().abort();
            match tokio::time::timeout(Duration::from_secs(1), &mut handle).await {
                Ok(Ok(_)) => info!("Aborted local TUI orchestrator task joined after abort"),
                Ok(Err(err)) if err.is_cancelled() => {
                    info!("Local TUI orchestrator task aborted successfully")
                }
                Ok(Err(err)) => {
                    warn!(error = %err, "Local TUI orchestrator task join failed after abort")
                }
                Err(_) => warn!("Timed out while joining aborted local TUI orchestrator task"),
            }
            LocalOrchestratorShutdownOutcome::AbortedAfterTimeout
        }
    }
}

/// Run the local TUI application.
#[allow(clippy::too_many_arguments)]
pub async fn run_tui(
    initial_changes: Vec<Change>,
    config: OrchestratorConfig,
    web_url: Option<String>,
    #[cfg(feature = "web-monitoring")] web_state: Option<Arc<crate::web::WebState>>,
    post_archive_action: PostArchiveAction,
    upstream_runtime: Option<crate::upstream::UpstreamRuntime>,
    lifecycle: LifecycleHandle,
) -> Result<()> {
    // Set up panic hook to restore terminal on panic
    let original_hook = std::panic::take_hook();
    std::panic::set_hook(Box::new(move |panic_info| {
        restore_terminal();
        original_hook(panic_info);
    }));

    let mut terminal = ratatui::init();

    // Mouse capture disabled due to terminal compatibility issues
    // Use PageUp/PageDown or k/j keys for scrolling instead
    // execute!(std::io::stdout(), EnableMouseCapture)?;

    let result = run_tui_loop(
        &mut terminal,
        initial_changes,
        config,
        web_url,
        #[cfg(feature = "web-monitoring")]
        web_state,
        post_archive_action,
        upstream_runtime,
        lifecycle,
    )
    .await;

    // Restore terminal state
    restore_terminal();

    result
}

/// Main TUI event loop
#[allow(clippy::too_many_arguments)]
async fn run_tui_loop(
    terminal: &mut DefaultTerminal,
    initial_changes: Vec<Change>,
    config: OrchestratorConfig,
    web_url: Option<String>,
    #[cfg(feature = "web-monitoring")] web_state: Option<Arc<crate::web::WebState>>,
    post_archive_action: PostArchiveAction,
    upstream_runtime: Option<crate::upstream::UpstreamRuntime>,
    lifecycle: LifecycleHandle,
) -> Result<()> {
    let repo_root = std::env::current_dir()?;

    let (committed_change_ids, uncommitted_file_change_ids): (HashSet<String>, HashSet<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: {}", err);
                    initial_changes
                        .iter()
                        .map(|change| change.id.clone())
                        .collect()
                }
            };

        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: {}", err);
                    HashSet::new()
                }
            };

        (committed_change_ids, uncommitted_file_change_ids)
    };
    let worktree_base_dir = config
        .get_workspace_base_dir()
        .map(PathBuf::from)
        .unwrap_or_else(|| crate::config::defaults::default_workspace_base_dir(Some(&repo_root)));
    let worktree_manager = GitWorkspaceManager::new(
        worktree_base_dir.clone(),
        repo_root.clone(),
        config.get_max_concurrent_workspaces(),
        config.clone(),
    );
    let worktree_change_ids: HashSet<String> =
        match worktree_manager.list_worktree_change_ids().await {
            Ok(ids) => ids,
            Err(err) => {
                warn!("Failed to load worktree snapshot: {}", err);
                HashSet::new()
            }
        };

    // Collect initial worktree paths for all changes
    let mut initial_worktree_paths = std::collections::HashMap::new();
    for change in &initial_changes {
        match crate::vcs::git::get_worktree_path_for_change(&repo_root, &change.id).await {
            Ok(Some(wt_path)) => {
                initial_worktree_paths.insert(change.id.clone(), wt_path);
            }
            Ok(None) => {
                // No worktree for this change
            }
            Err(e) => {
                debug!("Failed to get worktree path for {}: {}", change.id, e);
            }
        }
    }

    // Create shared orchestration state for unified tracking across TUI and Web
    let change_ids: Vec<String> = initial_changes.iter().map(|c| c.id.clone()).collect();
    let max_iterations = config.get_max_iterations();
    let shared_state = std::sync::Arc::new(tokio::sync::RwLock::new(
        crate::orchestration::state::OrchestratorState::new(change_ids, max_iterations),
    ));

    let tui_config = crate::tui::config::TuiConfig::load_user_config()?;
    let mut app = AppState::new(initial_changes);
    app.set_tui_config(tui_config);
    app.worktree_paths = initial_worktree_paths;
    // Inject shared state reference into TUI for unified tracking
    app.set_shared_state(shared_state.clone());
    // Startup already refused a workspace without a usable Git repository, so
    // worktree execution is available here by construction.
    app.max_concurrent = config.get_max_concurrent_workspaces();
    app.vcs_backend = config.get_vcs_backend().to_string();
    // Header project identity, captured once. The same `repo_root` backs the
    // refresh task and the dirty predicate, so a later process `chdir` moves
    // none of the three.
    app.set_project_path(repo_root.clone());
    // The branch that identity is read against, resolved once here and never
    // again. A detached HEAD or a failed lookup leaves it unset: the header
    // simply omits the segment, because an operator reading a guessed branch is
    // worse off than one reading none, and neither outcome may stop startup.
    app.set_base_branch(
        match crate::vcs::git::commands::get_current_branch(&repo_root).await {
            Ok(branch) => branch,
            Err(err) => {
                debug!("Failed to resolve the startup base branch for the header: {err}");
                None
            }
        },
    );
    app.publish_parallel_runtime();
    app.apply_parallel_eligibility(&committed_change_ids, &uncommitted_file_change_ids);
    app.apply_worktree_status(&worktree_change_ids);
    app.web_url = web_url;

    // Create shared stagger state for all AI commands (worktree, apply, archive, acceptance)
    use crate::ai_command_runner::{AiCommandRunner, SharedStaggerState};
    let shared_stagger_state: SharedStaggerState = Arc::new(tokio::sync::Mutex::new(None));
    let ai_runner =
        AiCommandRunner::from_orchestrator_config(&config, shared_stagger_state.clone());

    // Two directions, deliberately not one channel.
    //
    // `tx` is the producer side every TUI-local emitter already holds. `rx` is
    // therefore a *producer* boundary: this loop is the dispatch owner for what
    // arrives there, and applies it to the reducer once before rendering it.
    //
    // `frontend_rx` is the delivery side of the orchestration boundary's own
    // dispatch owner. Those events were already applied to the reducer and
    // already projected to `/api/v2`; re-applying them here is exactly the
    // double transition — a doubled apply count, a doubled event sequence —
    // that this frontend must not cause.
    let (tx, mut rx) = mpsc::channel::<OrchestratorEvent>(100);
    let (frontend_tx, mut frontend_rx) = mpsc::channel::<OrchestratorEvent>(100);
    let (cmd_tx, mut cmd_rx) = mpsc::channel::<TuiCommand>(100);

    // Inject shared state into WebState if web monitoring is enabled.
    //
    // The execution-mark store and the repository root go with it: without them
    // the v2 snapshot could report neither operator intent nor a redacted
    // change-to-worktree relation, and a remote frontend would have to infer
    // both.
    #[cfg(feature = "web-monitoring")]
    if let Some(ref ws) = web_state {
        ws.set_shared_state(shared_state.clone()).await;
        ws.set_execution_marks(app.execution_marks()).await;
        ws.set_parallel_runtime(app.parallel_runtime()).await;
        // Bound later, once the supervisor exists: see `managed_termination`.
        ws.set_repo_root(repo_root.clone()).await;
    }

    // Dynamic queue for runtime change additions
    let dynamic_queue = DynamicQueue::new();

    // Manual resolve counter for tracking active manual resolves
    // This allows manual resolves to consume parallel execution slots
    let manual_resolve_counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));

    // Shared flag for graceful stop (signaling orchestrator to stop after current change)
    let graceful_stop_flag = Arc::new(AtomicBool::new(false));

    // One reconciler for this process: the shared mark store plus the shared
    // operator mutation guard. Every production dispatch path binds *this* value,
    // so a mark revoked by an event and a mark set by a command are the same fact
    // in the TUI, in `/api/v2`, and in Start target resolution.
    let mark_reconciler = crate::orchestration::mark_reconciliation::ExecutionMarkReconciler::new(
        app.execution_marks(),
        app.parallel_runtime(),
    );

    // The one ephemeral process lifecycle mode. TUI `execution_mode` and Web
    // `app_mode` are projections of it; command admission is validated against
    // it rather than against whichever frontend happened to submit the command.
    // It is process-local and starts fresh here, so a restart recomputes the next
    // action from the workspace alone.
    let core_mode = Arc::new(crate::orchestration::operator_coordinator::CoreMode::new());

    // The one process-local execution-facts store. The dispatch owner feeds it,
    // `/api/v2/execution-status` reads it, and stop settlement consults it, so
    // "which phase was running" has one answer for every consumer. It is an
    // observability output only: it starts empty on every restart and no
    // workflow decision reads it.
    let execution_facts =
        Arc::new(crate::orchestration::execution_facts::ExecutionFactsStore::new());

    // *The* dispatch owner for this process. Runner-local producers, accepted
    // operator-command outcomes, and every orchestration run publish through it,
    // which is what makes one internal event produce exactly one reducer
    // transition, one core mode transition, one mark reconciliation, and one
    // delivery per frontend — no matter which producer raised it.
    //
    // Its TUI sink is `frontend_tx`, the delivery side. Wiring the producer
    // channel here instead would route every dispatched event back into the
    // producer boundary and apply it to the reducer a second time.
    let dispatcher = Arc::new(
        crate::events::EventDispatcher::new(
            shared_state.clone(),
            crate::tui::orchestrator::process_event_sinks(
                &frontend_tx,
                #[cfg(feature = "web-monitoring")]
                web_state.as_ref(),
            ),
        )
        .with_mark_reconciler(Some(mark_reconciler.clone()))
        .with_core_mode(Some(core_mode.clone()))
        .with_execution_facts(Some(execution_facts.clone())),
    );

    // TUI-local producers (key handlers, hooks, worktree operations, the
    // auto-refresh) reach that owner through this forwarder rather than through
    // the render loop. Forwarding off the loop is what stops a burst of local
    // events from filling the delivery channel while the only task that drains
    // it is blocked publishing into it.
    {
        let producer_dispatcher = dispatcher.clone();
        tokio::spawn(async move {
            while let Some(event) = rx.recv().await {
                producer_dispatcher.dispatch(event).await;
            }
        });
    }

    // One run supervisor owns the local orchestrator task for this invocation.
    // The TUI adapter and the `/api/v2` adapter drive it through the same
    // run-control service, so neither can start or cancel a run the other
    // cannot see.
    let supervisor = Arc::new(crate::tui::run_supervisor::TuiRunSupervisor::new(
        repo_root.clone(),
        config.clone(),
        dispatcher.clone(),
        dynamic_queue.clone(),
        shared_state.clone(),
        manual_resolve_counter.clone(),
        post_archive_action.clone(),
        upstream_runtime.clone(),
        graceful_stop_flag.clone(),
    ));

    // External SIGINT/SIGTERM become a quit *request* the event loop drains, so
    // they leave through the same bounded shutdown boundary as a keypress
    // instead of terminating the process on top of live agent descendants.
    let external_shutdown = crate::tui::run_supervisor::ExternalShutdownRequest::new();
    external_shutdown.install();

    // The one managed-process authority for this process. The supervisor holds
    // whichever run is live, so reading its scope per call is what keeps a
    // targeted force-stop bound to the run that owns the processes rather than
    // to a scope captured once at startup and reaped several runs ago.
    let managed_termination: Arc<
        dyn crate::orchestration::operator_command::ManagedProcessTermination,
    > = {
        let scope_owner = supervisor.clone();
        Arc::new(
            crate::orchestration::operator_command::LiveRunManagedProcesses::new(move || {
                scope_owner.run_command_scope()
            }),
        )
    };

    // The single process-local application services every frontend commands
    // through. They are built once here, before the first keypress and before v2
    // is bound, so a remote command and a keypress cannot reach different
    // instances of the lifecycle matrix.
    let operator_service = {
        use crate::orchestration::operator_command::{
            HookRunnerQueueHooks, OperatorCommandService,
        };
        // The same two inputs the Acceptance boundary itself derives from: where
        // a change's managed worktree lives, and which embedded reviewer skill
        // the manifest binds.
        let acceptance_worktree_base_dir = worktree_base_dir.clone();
        let accept_skill_name = config.get_accept_skill().to_string();
        let hook_runner = crate::hooks::HookRunner::with_event_tx(
            config.get_hooks(),
            repo_root.clone(),
            tx.clone(),
        );
        Arc::new(
            OperatorCommandService::new(
                shared_state.clone(),
                Arc::new(dynamic_queue.clone()),
                Arc::new(HookRunnerQueueHooks::new(hook_runner)),
                app.execution_marks(),
            )
            .with_parallel(app.parallel_runtime())
            // Read-only, and only at stop settlement: the phase a settled
            // command reports comes from the same store the status resource
            // publishes.
            .with_execution_facts(execution_facts.clone())
            // The managed ownership graph a targeted force-stop signals
            // through. It re-reads the *live* run's command scope on every
            // call, so a kill always reaches the run that is actually
            // executing and reports "owns nothing" between runs.
            .with_managed_termination(managed_termination.clone())
            // Refuses an unchanged retry before anything is dispatched, from
            // two facts that must both hold: a hold *this owner* recorded, and
            // a fingerprint recomputed from the managed worktree right now. A
            // restarted owner remembers nothing and refuses nothing; a
            // repository change moves the fingerprint and restores eligibility.
            // The path is resolved per call, and a change whose worktree is gone
            // simply has nothing to compare against.
            .with_acceptance_admission(Arc::new(
                crate::orchestration::acceptance::execution_manifest::WorkspaceAcceptanceAdmission::new(
                    {
                        let base = acceptance_worktree_base_dir.clone();
                        move |change_id: &str| {
                            let path = base.join(change_id.replace(['/', '\\', ' '], "-"));
                            path.is_dir().then_some(path)
                        }
                    },
                    accept_skill_name.clone(),
                    config.get_state_base_dir().map(str::to_string),
                ),
            )),
        )
    };
    // One store, three readers: the start guard, the operator service that
    // mutates the toggle, and the TUI that observes it.
    let start_eligibility = app.parallel_runtime();
    let run_control = Arc::new(crate::orchestration::run_control::RunControlService::new(
        shared_state.clone(),
        operator_service.clone(),
        supervisor.clone(),
        app.resolve_reservations(),
        start_eligibility.clone(),
    ));

    // The one application transaction. Both adapters submit typed intent to it,
    // so equivalent TUI and `/api/v2` intent cannot take different paths through
    // validation, commit ordering, outcome dispatch, or scheduler activation.
    #[allow(unused_mut)]
    let mut application = crate::orchestration::operator_coordinator::OperatorApplication::new(
        core_mode.clone(),
        run_control.clone(),
        dispatcher.clone(),
    )
    // Explanatory Git evidence for stop settlement, read from the server's own
    // change-to-worktree mapping while the terminated worktree is quiescent.
    .with_apply_commit_evidence(Some(Arc::new(
        crate::orchestration::apply_commit_evidence::GitApplyCommitEvidence::new(repo_root.clone()),
    )));
    #[cfg(feature = "web-monitoring")]
    if let Some(ref ws) = web_state {
        // Where an outcome dispatch's exact revision is read back from. Without
        // it the coordinator still runs the same transaction; it just has no
        // command record to bind a revision to.
        let revisions: Arc<dyn crate::events::OutcomeRevisions> = ws.clone();
        application = application.with_revisions(Some(revisions));
        // The same port the shared transaction kills through, so the
        // `actions.force_stop_change` a client reads and the admission a
        // command hits are one fact rather than two.
        ws.set_managed_termination(managed_termination.clone())
            .await;
    }
    let application = Arc::new(application);
    // Mark settlement becomes possible only now: the transaction that admits its
    // additions has to exist before a deadline may be armed against it. Before
    // this line every mark write in this process is mark-only.
    crate::orchestration::operator_coordinator::bind_mark_settlement(&application);

    // The TUI's ordered submission path. One worker drains it, so keypresses
    // reach the coordinator in the order the operator made them, while the event
    // loop below never awaits the application gate itself.
    let (submission_tx, mut submission_rx) =
        mpsc::channel::<crate::tui::command_handlers::Submission>(256);
    let (feedback_tx, mut feedback_rx) =
        mpsc::channel::<crate::tui::command_handlers::CommandFeedback>(256);
    {
        let worker_application = application.clone();
        let worker_feedback = feedback_tx.clone();
        tokio::spawn(async move {
            while let Some(submission) = submission_rx.recv().await {
                let settled = submission.run(&worker_application).await;
                if worker_feedback.send(settled).await.is_err() {
                    break;
                }
            }
        });
    }
    // One worktree service for this repository, for the same reason: it owns the
    // repository mutation guard, and two instances would be two guards that
    // cannot see each other's in-flight deletion.
    let worktree_service =
        crate::tui::command_handlers::build_worktree_service(&repo_root, &config, &tx);

    // Bind `/api/v2` command delegation to the shared application services.
    //
    // The web server started before this point (it owns the URL shown in the
    // TUI), so v2 refuses commands until the same services the TUI uses exist.
    // Binding them here is what makes a remote command and a keypress take
    // identical paths through lifecycle validation and side effects.
    #[cfg(feature = "web-monitoring")]
    if let Some(ref ws) = web_state {
        // The remote worktree port is built once and bound to both halves of v2:
        // the read routes and the command executor must agree about which
        // worktrees exist and which opaque IDs address them. It is wired to the
        // *same* service the TUI commands through, so a remote delete and a
        // keypress contend for one guard rather than racing through two.
        let worktree_port: Arc<dyn crate::web::remote_control_api::worktrees::WorktreeOperations> =
            Arc::new(
                crate::web::remote_control_api::worktrees::RemoteWorktreeOperations::new(
                    worktree_service.clone(),
                    Arc::new(crate::web::remote_control_api::worktrees::WorktreeRegistry::new()),
                    repo_root.clone(),
                ),
            );

        let runtime = ws.remote_control();
        runtime
            .bind(Arc::new(
                crate::web::remote_control_api::executor::SharedServiceExecutor::new(
                    application.clone(),
                    ws.clone(),
                )
                .with_worktrees(worktree_port.clone()),
            ))
            .await;
        runtime.bind_worktrees(worktree_port).await;
        // The same gate a keypress takes. Binding it is what turns `/api/v2`
        // admission from "atomic for the record" into "serialized through
        // settlement", so two new commands cannot consume one revision.
        runtime.bind_gate(application.gate()).await;
        // The same liveness authority the execution-status resource reports as
        // `scheduler_running`, kept distinct from `has_active_work` so a parked
        // persistent scheduler is never mistaken for admitted work.
        runtime.bind_run_boundary(supervisor.clone());
        // The store the dispatch owner already feeds, so a v2 client reads the
        // same phases the coordinator settles against.
        ws.set_execution_facts(execution_facts.clone()).await;
    }

    // Cancellation token for graceful shutdown
    let cancel_token = CancellationToken::new();

    // Start auto-refresh task
    let refresh_tx = tx.clone();
    let refresh_cancel = cancel_token.clone();
    let refresh_repo_root = repo_root.clone();
    let refresh_worktree_base_dir = worktree_base_dir.clone();
    let refresh_config = config.clone();
    // Presentation-only observations from the same refresh tick. They travel on
    // their own channel rather than as an `ExecutionEvent` precisely because
    // they are not workflow facts: no reducer transition, no operator snapshot
    // revision, and no `/api/v2` event is owed for them.
    let (refresh_observation_tx, mut refresh_observation_rx) =
        mpsc::channel::<TuiRefreshObservation>(16);
    let refresh_handle = tokio::spawn(async move {
        let worktree_manager = GitWorkspaceManager::new(
            refresh_worktree_base_dir,
            refresh_repo_root.clone(),
            refresh_config.get_max_concurrent_workspaces(),
            refresh_config,
        );
        let mut stale_refresh_root_warned = false;
        let mut dirty_observation_warned = false;
        let mut interval = tokio::time::interval(Duration::from_secs(AUTO_REFRESH_INTERVAL_SECS));
        loop {
            tokio::select! {
                _ = refresh_cancel.cancelled() => {
                    break;
                }
                _ = interval.tick() => {
                    if should_skip_local_refresh(
                        &refresh_repo_root,
                        &mut stale_refresh_root_warned,
                    ) {
                        continue;
                    }

                    // The captured startup root, not the process current
                    // directory: a `cd` elsewhere in this process must not
                    // silently retarget what the header reports.
                    if let Some(observation) = observe_workspace_dirty(
                        &refresh_repo_root,
                        &mut dirty_observation_warned,
                    )
                    .await
                    {
                        // A closed receiver means the render loop is gone, so
                        // the badge has no consumer left; the refresh task keeps
                        // serving its orchestration duties regardless.
                        let _ = refresh_observation_tx.send(observation).await;
                    }

                    match refresh_local_changes(&refresh_repo_root) {
                        Ok((mut changes, rejected_changes)) => {
                            let committed_change_ids: HashSet<String> =
                                match crate::vcs::git::commands::list_changes_in_head(&refresh_repo_root).await {
                                    Ok(ids) => ids.into_iter().collect(),
                                    Err(err) => {
                                        warn!("Failed to refresh committed change snapshot: {}", err);
                                        changes.iter().map(|change| change.id.clone()).collect()
                                    }
                                };
                            let uncommitted_file_change_ids: HashSet<String> =
                                match crate::vcs::git::commands::list_changes_with_uncommitted_files(&refresh_repo_root).await {
                                    Ok(ids) => ids.into_iter().collect(),
                                    Err(err) => {
                                        warn!("Failed to refresh uncommitted files snapshot: {}", err);
                                        HashSet::new()
                                    }
                                };
                            let worktree_change_ids: HashSet<String> =
                                match worktree_manager.list_worktree_change_ids().await {
                                    Ok(ids) => ids,
                                    Err(err) => {
                                        warn!("Failed to refresh worktree snapshot: {}", err);
                                        HashSet::new()
                                    }
                                };

                            // Collect worktree paths for all changes
                            let mut worktree_paths = std::collections::HashMap::new();

                            // Enrich progress from worktrees (uncommitted task artifact)
                            for change in &mut changes {
                                match crate::vcs::git::get_worktree_path_for_change(
                                    &refresh_repo_root,
                                    &change.id
                                ).await {
                                    Ok(Some(wt_path)) => {
                                        // Store the worktree path for this change
                                        worktree_paths.insert(change.id.clone(), wt_path.clone());
                                        // Use unified fallback helper: worktree → archive → base
                                        match crate::task_parser::parse_progress_with_fallback(
                                            &change.id,
                                            Some(&wt_path)
                                        ) {
                                            Ok(progress) => {
                                                if progress.total > 0 {
                                                    change.completed_tasks = progress.completed;
                                                    change.total_tasks = progress.total;
                                                } else {
                                                    // Keep existing progress if 0/0
                                                    debug!("Keeping existing progress for {} (parsed: 0/0)", change.id);
                                                }
                                            }
                                            Err(e) => {
                                                debug!("Failed to read progress for {}: {}", change.id, e);
                                                // Keep existing progress (from base tree)
                                            }
                                        }
                                    }
                                    Ok(None) => {
                                        // No worktree exists, use progress from base tree
                                    }
                                    Err(e) => {
                                        warn!("Failed to get worktree path for {}: {}", change.id, e);
                                        // Keep existing progress
                                    }
                                }
                            }

                            // Check which worktrees are not ahead of base (for MergeWait auto-clear)
                            let mut worktree_not_ahead_ids = std::collections::HashSet::new();
                            // Check which worktrees are archived but not merged (for MergeWait restoration)
                            let mut merge_wait_ids = std::collections::HashSet::new();

                            // Get base branch (current branch in main repo)
                            if let Ok(Some(base_branch)) = crate::vcs::git::commands::get_current_branch(&refresh_repo_root).await {
                                // For each change with a worktree, check if worktree branch is ahead of base
                                for (change_id, wt_path) in &worktree_paths {
                                    // Get the branch name for this worktree
                                    if let Ok(Some(worktree_branch)) = crate::vcs::git::commands::get_current_branch(wt_path).await {
                                        // Count commits ahead
                                        match crate::vcs::git::commands::count_commits_ahead(
                                            &refresh_repo_root,
                                            &base_branch,
                                            &worktree_branch
                                        ).await {
                                            Ok(0) => {
                                                // Worktree is not ahead (0 commits), mark for auto-clear
                                                worktree_not_ahead_ids.insert(change_id.clone());
                                            }
                                            Ok(_) => {
                                                // Worktree is ahead, keep MergeWait if present
                                            }
                                            Err(e) => {
                                                debug!("Failed to count commits ahead for {}: {}", change_id, e);
                                                // On error, don't auto-clear (safe default)
                                            }
                                        }
                                    }

                                    // Detect WorkspaceState::Archived for MergeWait restoration
                                    match crate::execution::state::detect_workspace_state(change_id, wt_path, &base_branch).await {
                                        Ok(crate::execution::state::WorkspaceState::Archived) => {
                                            // Worktree is archived but not merged, restore MergeWait
                                            merge_wait_ids.insert(change_id.clone());
                                            debug!("Detected MergeWait for '{}': archive complete, waiting for merge", change_id);
                                        }
                                        Ok(_) => {
                                            // Other states, do nothing
                                        }
                                        Err(e) => {
                                            debug!("Failed to detect workspace state for {}: {}", change_id, e);
                                            // On error, skip detection (safe default)
                                        }
                                    }
                                }
                            }

                            if refresh_tx
                                .send(OrchestratorEvent::ChangesRefreshed {
                                    changes,
                                    rejected_changes,
                                    committed_change_ids,
                                    uncommitted_file_change_ids,
                                    worktree_change_ids,
                                    worktree_paths,
                                    worktree_not_ahead_ids,
                                    merge_wait_ids,
                                })
                                .await
                                .is_err()
                            {
                                break;
                            }
                        }
                        Err(e) => {
                            let _ = refresh_tx
                                .send(OrchestratorEvent::Log(LogEntry::error(format!(
                                    "Refresh failed: {}",
                                    e
                                ))))
                                .await;
                        }
                    }

                    // Refresh worktrees with conflict check (if in Worktrees view)
                    // We do this in the background without blocking
                    let wt_refresh_tx = refresh_tx.clone();
                    let wt_refresh_repo_root = refresh_repo_root.clone();
                    tokio::spawn(async move {
                        match load_worktrees_with_conflict_check(&wt_refresh_repo_root).await {
                            Ok(worktrees) => {
                                let _ = wt_refresh_tx
                                    .send(OrchestratorEvent::WorktreesRefreshed { worktrees })
                                    .await;
                            }
                            Err(e) => {
                                debug!("Failed to refresh worktrees: {}", e);
                                // Don't spam logs on refresh failures
                            }
                        }
                    });

                    log_deduplicator::maybe_log_summary();
                }
            }
        }
    });

    // External lifecycle reporting is derived from typed TUI state, never from
    // rendered screen contents. Unchanged states are deduplicated by the
    // dispatcher, so publishing once per frame is cheap and non-blocking.
    let lifecycle_workspace = repo_root.display().to_string();
    let publish_lifecycle_state = |app: &AppState| {
        if !lifecycle.is_enabled() {
            return;
        }
        let snapshot = TuiLifecycleSnapshot::from_app(app);
        lifecycle.publish_state(
            snapshot.lifecycle_state(),
            snapshot.lifecycle_context(&lifecycle_workspace),
        );
    };

    publish_lifecycle_state(&app);

    loop {
        // Increment spinner frame for animation (updates every 100ms)
        app.spinner_frame = (app.spinner_frame + 1) % SPINNER_CHARS.len();

        // Draw the UI
        terminal.draw(|frame| render(frame, &mut app))?;

        // Handle events with timeout
        if event::poll(Duration::from_millis(100))? {
            match event::read()? {
                Event::Key(key) if key.kind == KeyEventKind::Press => {
                    // Create context for key event handling
                    let mut key_ctx = KeyEventContext {
                        app: &mut app,
                        terminal,
                        repo_root: &repo_root,
                        config: &config,
                        worktree_base_dir: &worktree_base_dir,
                        tx: &tx,
                        cmd_tx: &cmd_tx,
                        ai_runner: &ai_runner,
                        supervisor: &supervisor,
                    };

                    // Handle key event using helper
                    match handle_key_event(key, &mut key_ctx).await {
                        Ok(Some(cmd)) => {
                            // Send command to command channel for processing
                            let _ = cmd_tx.send(cmd).await;
                        }
                        Ok(None) => {
                            // No command to execute
                        }
                        Err(e) => {
                            app.add_log(LogEntry::error(format!("Key handling error: {}", e)));
                        }
                    }

                    // The interaction expressed intent; the shared service owns
                    // the write. Draining here — target-scoped, under the same
                    // guard event reconciliation takes — is what stops a stale
                    // cached row from resurrecting a mark a concurrent event
                    // already revoked.
                    apply_pending_mark_writes(&mut app, &operator_service).await;

                    // Check if app should quit (set by Ctrl+C)
                    if app.should_quit {
                        break;
                    }
                }
                Event::Mouse(mouse) => {
                    match mouse.kind {
                        MouseEventKind::ScrollUp => {
                            // Scroll logs up (show older entries) - 3 lines at a time
                            app.scroll_logs_up(3);
                        }
                        MouseEventKind::ScrollDown => {
                            // Scroll logs down (show newer entries) - 3 lines at a time
                            app.scroll_logs_down(3);
                        }
                        _ => {}
                    }
                }
                _ => {}
            }
        }

        // Authoritative deliveries from the process-lifetime dispatch owner —
        // runner-local producers, accepted command outcomes, and orchestration
        // runs alike. The reducer transition, the core mode transition, and the
        // `/api/v2` projection already happened; this frontend reads the result
        // and renders it.
        //
        // Painting only: an orchestrator event never produces a command. Every
        // transition an event implies is dispatched by the scheduler from the
        // reducer-owned intent the same event already recorded.
        while let Ok(event) = frontend_rx.try_recv() {
            sync_reducer_display_caches(&mut app, &shared_state, &event).await;
            app.handle_orchestrator_event(event);
        }

        // Presentation-only refresh observations. They are drained on their own
        // because they own no reducer transition: nothing above needs to run for
        // them, and nothing below may change because of them.
        while let Ok(observation) = refresh_observation_rx.try_recv() {
            app.adopt_workspace_dirty_observation(observation);
        }

        // TUI command submissions. The coordinator work runs off this loop, so
        // neither the application gate nor a termination waiter is ever awaited
        // inside event processing or rendering; accepted outcomes return through
        // the authoritative dispatch above.
        while let Ok(cmd) = cmd_rx.try_recv() {
            let mut cmd_ctx = TuiCommandContext {
                app: &mut app,
                tx: &tx,
                application: &application,
                submissions: &submission_tx,
                feedback: &feedback_tx,
                worktree_service: &worktree_service,
            };

            if let Err(e) = handle_tui_command(cmd, &mut cmd_ctx, &shared_state).await {
                app.add_log(LogEntry::error(format!("Command handling error: {}", e)));
            }
        }

        // Wording for commands the worker settled since the last frame. The
        // accepted state itself already arrived through the dispatch above.
        while let Ok(settled) = feedback_rx.try_recv() {
            crate::tui::command_handlers::apply_command_feedback(&mut app, settled);
        }

        // The mode every frontend projects, read back once per frame. Commands
        // and lifecycle events both move it inside the dispatch boundary, so the
        // TUI adopts it rather than maintaining a second opinion about whether
        // this process is running.
        app.adopt_core_mode(core_mode.get());

        // The eligibility set is a TUI observation, so it is republished once
        // per frame instead of at every place the TUI can change it.
        app.publish_parallel_runtime();

        publish_lifecycle_state(&app);

        // An external SIGINT/SIGTERM is drained here as an ordinary quit
        // request, so it reaches the same cleanup below that a keypress does.
        if external_shutdown.is_requested() {
            app.should_quit = true;
        }

        if app.should_quit {
            break;
        }
    }

    // Cleanup: cancel all TUI-scoped tasks and force-stop local orchestration launched by this TUI.
    cancel_token.cancel();

    // Wait for tasks to finish gracefully. Remote mode has no local orchestrator handle here;
    // remote server-side work is stopped only by explicit Stop/ForceStop commands.
    refresh_handle.abort();
    // Remote mode has no local run, so the supervisor holds no handle and no
    // scope and this is a no-op: closing a remote TUI client still sends no stop.
    //
    // The outcome is not discarded. Cleanup that could not prove the owned
    // process groups empty means detached descendants may still be mutating the
    // managed worktree, and reporting a clean exit there would tell the operator
    // the opposite of what happened.
    supervisor
        .shutdown_run(LOCAL_ORCHESTRATOR_SHUTDOWN_GRACE)
        .await
        .into_exit_result()
}

#[cfg(test)]
mod tests {
    use super::{
        is_refresh_root_usable, observe_workspace_dirty, refresh_local_changes,
        should_apply_event_to_tui_reducer, shutdown_local_orchestrator_task,
        LocalOrchestratorShutdownOutcome,
    };
    use super::{AppState, OrchestratorEvent, TuiRefreshObservation};
    use crate::events::{ExecutionEvent, RejectionOutcome, StalledBlocker};
    use crate::openspec::{Change, ProposalMetadata};
    use crate::vcs::WorkspaceStatus;
    use std::collections::{HashMap, HashSet};
    use std::path::{Path, PathBuf};
    use std::sync::Arc;
    use std::time::Duration;
    use tokio::sync::mpsc;
    use tokio_util::sync::CancellationToken;

    fn sample_change() -> Change {
        Change {
            id: "change-a".to_string(),
            completed_tasks: 0,
            total_tasks: 1,
            last_modified: "now".to_string(),
            dependencies: Vec::new(),
            metadata: ProposalMetadata::default(),
        }
    }

    fn empty_changes_refreshed_event() -> ExecutionEvent {
        ExecutionEvent::ChangesRefreshed {
            changes: vec![sample_change()],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::new(),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::new(),
            worktree_paths: HashMap::<String, PathBuf>::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::new(),
        }
    }

    fn stalled_blocker() -> StalledBlocker {
        StalledBlocker::acceptance_external(
            "pending_verification",
            "managed verification job still running",
        )
    }

    /// A boundary run's events reach this frontend already applied.
    ///
    /// The regression this pins: the loop used to write the reducer for every
    /// event it received, including the ones the orchestration boundary had
    /// already applied. One `ApplyCompleted` then advanced the apply count
    /// twice — once per frontend path — which is exactly what an authoritative
    /// remote snapshot cannot survive.
    #[tokio::test]
    async fn frontend_delivery_does_not_reapply_the_reducer() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;
        use crate::tui::events::TuiEventSink;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let (frontend_tx, mut frontend_rx) = mpsc::channel::<OrchestratorEvent>(16);
        let sinks: Vec<Arc<dyn EventSink>> = vec![Arc::new(TuiEventSink::new(frontend_tx))];

        for event in [
            ExecutionEvent::ApplyStarted {
                change_id: "change-a".to_string(),
                command: "apply".to_string(),
            },
            ExecutionEvent::ApplyCompleted {
                change_id: "change-a".to_string(),
                revision: String::new(),
            },
        ] {
            crate::events::dispatch_event(&shared_state, &sinks, event).await;
        }
        assert_eq!(shared_state.read().await.apply_count("change-a"), 1);

        let mut app = AppState::new(vec![sample_change()]);
        app.set_shared_state(shared_state.clone());
        while let Ok(event) = frontend_rx.try_recv() {
            super::sync_reducer_display_caches(&mut app, &shared_state, &event).await;
        }

        assert_eq!(
            shared_state.read().await.apply_count("change-a"),
            1,
            "the frontend applied a delivered event a second time"
        );
    }

    /// A TUI-local producer's event is applied once, by this loop, and reaches
    /// the other frontends through the same authoritative dispatch.
    #[cfg(feature = "web-monitoring")]
    #[tokio::test]
    async fn tui_local_producer_events_are_dispatched_once_to_every_frontend() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;
        use crate::web::state::{WebEventSink, WebState};

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let web_state = Arc::new(WebState::new(&[]));
        web_state.set_shared_state(shared_state.clone()).await;
        let projection = web_state.remote_control().projection();
        let local_sinks: Vec<Arc<dyn EventSink>> =
            vec![Arc::new(WebEventSink::new(web_state.clone()))];

        let event = ExecutionEvent::ApplyCompleted {
            change_id: "change-a".to_string(),
            revision: String::new(),
        };
        crate::events::dispatch_event(&shared_state, &local_sinks, event.clone()).await;

        let mut app = AppState::new(vec![sample_change()]);
        app.set_shared_state(shared_state.clone());
        super::sync_reducer_display_caches(&mut app, &shared_state, &event).await;

        assert_eq!(
            shared_state.read().await.apply_count("change-a"),
            1,
            "a locally produced event must be applied exactly once"
        );
        let (_, _, sequence) = projection.snapshot();
        assert_eq!(
            sequence, 1,
            "a locally produced event must reach the v2 stream exactly once"
        );
    }

    /// An acceptance blocker whose external prerequisite claim is complete, so
    /// the reducer classifies it as canonical `blocked`.
    fn external_blocker() -> StalledBlocker {
        StalledBlocker::acceptance_external(
            "pending_verification",
            "docker image pull failed: lookup registry-1.docker.io i/o timeout",
        )
    }

    /// A reported hold with no verifiable unblock condition, which the reducer
    /// keeps on the `stalled` path instead of promoting to external `blocked`.
    fn unvalidated_blocker() -> StalledBlocker {
        StalledBlocker {
            category: "no_progress".to_string(),
            phase: "acceptance".to_string(),
            gate: "acceptance_review".to_string(),
            error_summary: "no semantic progress across the retry budget".to_string(),
            evidence: vec!["tasks.md unchanged across two iterations".to_string()],
            unblock_condition: None,
            prerequisite_owner: None,
            next_action: "operator review".to_string(),
            resumable: true,
            worktree_preserved: true,
        }
    }

    /// A persistent TUI holds `Running` while every remaining change waits, and
    /// external lifecycle reporting must say `blocked` rather than `working`.
    ///
    /// The path is deliberately traversed end to end — event dispatch, reducer
    /// blocker classification, display-cache synchronization, snapshot
    /// projection — because the bug lived in the seam between the reducer's
    /// canonical row status and the frontend snapshot, not in either alone.
    #[tokio::test]
    async fn running_tui_projects_blocked_for_reducer_blocked_and_stalled_rows() {
        use crate::events::EventSink;
        use crate::lifecycle_integration::LifecycleState;
        use crate::orchestration::state::OrchestratorState;
        use crate::tui::lifecycle::TuiLifecycleSnapshot;
        use crate::tui::types::AppExecutionMode;

        for (blocker, expected_status) in [
            (external_blocker(), "blocked"),
            (unvalidated_blocker(), "stalled"),
        ] {
            let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
                vec!["change-a".to_string()],
                10,
            )));
            let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
            let mut app = AppState::new(vec![sample_change()]);
            app.set_shared_state(shared_state.clone());
            app.execution_mode = AppExecutionMode::Running;

            for event in [
                ExecutionEvent::AcceptanceStarted {
                    change_id: "change-a".to_string(),
                    command: "accept".to_string(),
                },
                ExecutionEvent::AcceptanceGated {
                    change_id: "change-a".to_string(),
                    blocker: blocker.clone(),
                },
            ] {
                crate::events::dispatch_event(&shared_state, &sinks, event.clone()).await;
                super::sync_reducer_display_caches(&mut app, &shared_state, &event).await;
            }

            assert_eq!(
                app.changes[0].display_status_cache, expected_status,
                "reducer classification must reach the TUI row for {blocker:?}"
            );
            assert_eq!(
                app.execution_mode,
                AppExecutionMode::Running,
                "the persistent process stays Running while it waits"
            );
            assert_eq!(
                TuiLifecycleSnapshot::from_app(&app).lifecycle_state(),
                LifecycleState::Blocked,
                "a {expected_status}-only wait must report blocked"
            );
        }
    }

    /// Once work is active or queued again, the same still-`Running` TUI reports
    /// `working` even while a blocked row remains on screen.
    #[tokio::test]
    async fn active_or_queued_reducer_row_restores_working_lifecycle() {
        use crate::events::EventSink;
        use crate::lifecycle_integration::LifecycleState;
        use crate::orchestration::state::{OrchestratorState, ReducerCommand};
        use crate::tui::lifecycle::TuiLifecycleSnapshot;
        use crate::tui::types::AppExecutionMode;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string(), "change-b".to_string()],
            10,
        )));
        let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
        let mut change_b = sample_change();
        change_b.id = "change-b".to_string();
        let mut app = AppState::new(vec![sample_change(), change_b]);
        app.set_shared_state(shared_state.clone());
        app.execution_mode = AppExecutionMode::Running;

        let gated = ExecutionEvent::AcceptanceGated {
            change_id: "change-a".to_string(),
            blocker: external_blocker(),
        };
        crate::events::dispatch_event(&shared_state, &sinks, gated.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &gated).await;
        assert_eq!(
            TuiLifecycleSnapshot::from_app(&app).lifecycle_state(),
            LifecycleState::Blocked
        );

        // A queued sibling keeps work admitted for dispatch.
        shared_state
            .write()
            .await
            .apply_command(ReducerCommand::AddToQueue("change-b".to_string()));
        let refresh = empty_changes_refreshed_event();
        crate::events::dispatch_event(&shared_state, &sinks, refresh.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &refresh).await;
        assert_eq!(app.changes[0].display_status_cache, "blocked");
        assert_eq!(app.changes[1].display_status_cache, "queued");
        assert_eq!(
            TuiLifecycleSnapshot::from_app(&app).lifecycle_state(),
            LifecycleState::Working,
            "a queued row alongside a blocked row must report working"
        );

        // And so does an actively executing sibling.
        let started = ExecutionEvent::ApplyStarted {
            change_id: "change-b".to_string(),
            command: "apply".to_string(),
        };
        crate::events::dispatch_event(&shared_state, &sinks, started.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &started).await;
        assert_eq!(app.changes[1].display_status_cache, "applying");
        assert_eq!(
            TuiLifecycleSnapshot::from_app(&app).lifecycle_state(),
            LifecycleState::Working,
            "an active row alongside a blocked row must report working"
        );
    }

    /// The production ordering that used to repaint a healthy Archive.
    ///
    /// `ArchiveStarted` → reducer-cache synchronization → a `ChangesRefreshed`
    /// whose `merge_wait_ids` carries the same change. The archive commit is
    /// real by then, so repository inspection legitimately reports an
    /// archived-but-not-integrated workspace — but Archive has not returned, and
    /// the row must not tell the operator that manual intervention is due. The
    /// reducer refuses the downgrade; the TUI's own refresh handling must too.
    #[tokio::test]
    async fn archive_refresh_preserves_reducer_owned_archiving_row() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
        let mut app = AppState::new(vec![sample_change()]);
        app.set_shared_state(shared_state.clone());
        app.changes[0].selected = true;
        app.publish_execution_marks();

        let archiving = ExecutionEvent::ArchiveStarted {
            change_id: "change-a".to_string(),
            command: "archive".to_string(),
        };
        crate::events::dispatch_event(&shared_state, &sinks, archiving.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &archiving).await;
        app.handle_orchestrator_event(archiving);
        assert_eq!(app.changes[0].display_status_cache, "archiving");

        let queued_before = shared_state.read().await.queued_change_ids();

        // The five-second scan sees the archive commit that Archive itself just
        // created, with the change not yet integrated into base.
        let refresh = ExecutionEvent::ChangesRefreshed {
            changes: vec![sample_change()],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::from(["change-a".to_string()]),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["change-a".to_string()]),
            worktree_paths: HashMap::<String, PathBuf>::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::from(["change-a".to_string()]),
        };
        crate::events::dispatch_event(&shared_state, &sinks, refresh.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &refresh).await;
        app.handle_orchestrator_event(refresh);

        assert_eq!(
            app.changes[0].display_status_cache, "archiving",
            "refresh merge-wait evidence repainted an in-flight Archive row"
        );
        assert_eq!(
            shared_state.read().await.display_status("change-a"),
            "archiving",
            "the reducer must still own the active archive lifecycle"
        );
        assert_eq!(
            shared_state.read().await.queued_change_ids(),
            queued_before,
            "a presentation-only refresh must not change queue intent"
        );
        assert!(
            app.changes[0].selected,
            "a presentation-only refresh must not clear the execution mark"
        );
    }

    // ========================================================================
    // Changes row layout: reducer archive-completion synchronization
    //
    // Integration evidence: a real `OrchestratorState`, real event dispatch, and
    // the real TUI cache-synchronization and refresh paths. The reducer is what
    // owns the archive milestone; these prove the TUI reads it rather than
    // inferring "post-archive" from the `resolving` display string.
    // ========================================================================

    /// The reducer records archive completion while truthfully advancing the row
    /// to a live post-archive status, and the TUI adopts both facts at once.
    #[tokio::test]
    async fn tui_change_row_layout_archive_sync_adopts_the_reducer_archive_record() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
        let mut app = AppState::new(vec![sample_change()]);
        app.set_shared_state(shared_state.clone());

        assert!(
            !app.changes[0].archive_complete_cache,
            "no archive is on record before the event"
        );

        let archived = ExecutionEvent::ChangeArchived("change-a".to_string());
        crate::events::dispatch_event(&shared_state, &sinks, archived.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &archived).await;

        // The reducer's own view: archived, and already moved on to post-archive
        // resolve handling. Neither fact is derivable from the other.
        assert!(shared_state.read().await.is_archived("change-a"));
        assert_eq!(
            app.changes[0].display_status_cache, "resolving",
            "the row must show the reducer's truthful post-archive status"
        );
        assert!(
            app.changes[0].archive_complete_cache,
            "the archive milestone must reach the row as presentation evidence"
        );
        assert_eq!(
            app.reducer_archived_changes(),
            &HashSet::from(["change-a".to_string()])
        );
    }

    /// One `ChangesRefreshed` event carrying `change-a` as an active change.
    fn archive_refresh_event() -> ExecutionEvent {
        ExecutionEvent::ChangesRefreshed {
            changes: vec![sample_change()],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::from(["change-a".to_string()]),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["change-a".to_string()]),
            worktree_paths: HashMap::<String, PathBuf>::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::from(["change-a".to_string()]),
        }
    }

    /// A row the refresh pass *constructs* adopts the retained archive evidence.
    ///
    /// The regression this pins: `ChangeState::from_change` knows nothing about
    /// the reducer, so a row built during refresh came back with no archive
    /// evidence and would render an empty `[ ]` for a change the reducer had
    /// already archived. The app-level snapshot is what closes that, and it has to
    /// be reapplied inside the refresh pass rather than one reducer sync later —
    /// refresh handling runs *after* cache synchronization in the run loop, so
    /// anything it leaves wrong is what the next frame draws.
    #[tokio::test]
    async fn tui_change_row_layout_archive_sync_survives_refresh_reconstruction() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
        // No rows yet: the refresh pass below is what materializes this one.
        let mut app = AppState::new(Vec::new());
        app.set_shared_state(shared_state.clone());

        let archived = ExecutionEvent::ChangeArchived("change-a".to_string());
        crate::events::dispatch_event(&shared_state, &sinks, archived.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &archived).await;
        assert!(app.changes.is_empty());
        assert_eq!(
            app.reducer_archived_changes(),
            &HashSet::from(["change-a".to_string()]),
            "the snapshot is retained even with no row to apply it to"
        );

        let refresh = archive_refresh_event();
        crate::events::dispatch_event(&shared_state, &sinks, refresh.clone()).await;
        app.handle_orchestrator_event(refresh.clone());

        assert_eq!(app.changes.len(), 1, "refresh constructed the row");
        assert!(
            app.changes[0].archive_complete_cache,
            "refresh reconstruction must reapply the retained archive evidence \
             in the same pass, not one reducer sync later"
        );

        // And the following reducer sync agrees rather than correcting it.
        super::sync_reducer_display_caches(&mut app, &shared_state, &refresh).await;
        assert!(app.changes[0].archive_complete_cache);
    }

    /// Refresh must not use a reactivated proposal directory to retire the
    /// archive evidence either.
    ///
    /// Refresh resets a row it observes as `archived` back to `not queued` when the
    /// proposal is present again. That reset is display-status work; the reducer's
    /// archive record is not refresh's to revoke, so the checkbox stays hidden.
    #[tokio::test]
    async fn tui_change_row_layout_archive_sync_refresh_does_not_revoke_the_archive_record() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
        let mut app = AppState::new(vec![sample_change()]);
        app.set_shared_state(shared_state.clone());

        let archived = ExecutionEvent::ChangeArchived("change-a".to_string());
        crate::events::dispatch_event(&shared_state, &sinks, archived.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &archived).await;

        // Post-archive merge readiness parks the row on a base-lane wait, which
        // is a live non-terminal status and not a terminal one.
        let deferred = ExecutionEvent::MergeDeferred {
            change_id: "change-a".to_string(),
            reason: "waiting for base".to_string(),
            auto_resumable: false,
        };
        crate::events::dispatch_event(&shared_state, &sinks, deferred.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &deferred).await;
        assert_eq!(app.changes[0].display_status_cache, "merge wait");
        assert!(app.changes[0].archive_complete_cache);

        // The row's own refresh reset path, reached by presenting it as archived
        // with the proposal directory present again and no merge-wait evidence to
        // restore the terminal status.
        app.changes[0].set_display_status_cache("archived");
        let refresh = ExecutionEvent::ChangesRefreshed {
            changes: vec![sample_change()],
            rejected_changes: Vec::new(),
            committed_change_ids: HashSet::from(["change-a".to_string()]),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["change-a".to_string()]),
            worktree_paths: HashMap::<String, PathBuf>::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::new(),
        };
        app.handle_orchestrator_event(refresh);

        assert_eq!(
            app.changes[0].display_status_cache, "not queued",
            "refresh still owns the display-status reset"
        );
        assert!(
            app.changes[0].archive_complete_cache,
            "but it must not clear the reducer's archive record"
        );
    }

    /// A `resolving` row with no reducer archive record stays a plain active row.
    ///
    /// This is the control for the whole design: a fresh process that resolves a
    /// retry shows `resolving` without having archived anything, and inferring
    /// post-archive state from that string would retire a live run candidate.
    #[tokio::test]
    async fn tui_change_row_layout_archive_sync_leaves_a_fresh_resolving_row_alone() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
        let mut app = AppState::new(vec![sample_change()]);
        app.set_shared_state(shared_state.clone());

        let resolving = ExecutionEvent::ResolveStarted {
            change_id: "change-a".to_string(),
            command: "resolve".to_string(),
        };
        crate::events::dispatch_event(&shared_state, &sinks, resolving.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &resolving).await;

        assert_eq!(app.changes[0].display_status_cache, "resolving");
        assert!(
            !shared_state.read().await.is_archived("change-a"),
            "no archive happened in this process"
        );
        assert!(
            !app.changes[0].archive_complete_cache,
            "an identical display status must not manufacture archive evidence"
        );
    }

    /// The exact observed force-stop order must leave `accepting` behind.
    ///
    /// `AcceptanceStarted` → authoritative `Stopped` dispatch → reducer-cache
    /// synchronization → local stopped handling → a later `ChangesRefreshed`.
    /// Before the reducer owned the stop, step three re-read an untouched
    /// `accepting`, and step five could restore it even after the agent process
    /// and the scheduler had already stopped.
    #[tokio::test]
    async fn stopped_reducer_sync_prevents_accepting_resurrection() {
        use crate::events::EventSink;
        use crate::orchestration::state::OrchestratorState;
        use crate::tui::types::AppExecutionMode;

        let shared_state = Arc::new(tokio::sync::RwLock::new(OrchestratorState::new(
            vec!["change-a".to_string()],
            10,
        )));
        let sinks: Vec<Arc<dyn EventSink>> = Vec::new();
        let mut app = AppState::new(vec![sample_change()]);
        app.set_shared_state(shared_state.clone());
        app.execution_mode = AppExecutionMode::Running;
        // The operator marked this change for execution; a stop must not clear
        // the mark, because F5 resume is what converts it back into intent.
        app.changes[0].selected = true;
        app.publish_execution_marks();

        let accepting = ExecutionEvent::AcceptanceStarted {
            change_id: "change-a".to_string(),
            command: "accept".to_string(),
        };
        crate::events::dispatch_event(&shared_state, &sinks, accepting.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &accepting).await;
        assert_eq!(app.changes[0].display_status_cache, "accepting");

        let stopped = ExecutionEvent::Stopped;
        crate::events::dispatch_event(&shared_state, &sinks, stopped.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &stopped).await;
        app.handle_orchestrator_event(stopped);

        assert_eq!(
            app.changes[0].display_status_cache, "not queued",
            "the reducer-derived stop did not reach the TUI row"
        );
        assert_eq!(app.execution_mode, AppExecutionMode::Stopped);

        // The refresh scan still sees the workspace the stopped run left behind.
        let refresh = ExecutionEvent::ChangesRefreshed {
            changes: vec![sample_change()],
            rejected_changes: Vec::new(),
            // Committed and clean, so the row stays parallel-eligible and the
            // separate ineligibility cleanup cannot be what clears its mark.
            committed_change_ids: HashSet::from(["change-a".to_string()]),
            uncommitted_file_change_ids: HashSet::new(),
            worktree_change_ids: HashSet::from(["change-a".to_string()]),
            worktree_paths: HashMap::<String, PathBuf>::new(),
            worktree_not_ahead_ids: HashSet::new(),
            merge_wait_ids: HashSet::from(["change-a".to_string()]),
        };
        crate::events::dispatch_event(&shared_state, &sinks, refresh.clone()).await;
        super::sync_reducer_display_caches(&mut app, &shared_state, &refresh).await;
        app.handle_orchestrator_event(refresh);

        assert_eq!(
            app.changes[0].display_status_cache, "not queued",
            "a later refresh restored an interrupted status after the run stopped"
        );
        assert_eq!(
            shared_state.read().await.display_status("change-a"),
            "not queued"
        );
        assert!(
            app.changes[0].selected,
            "the execution mark must survive so the row stays resumable"
        );
    }

    #[test]
    fn tui_reducer_sync_includes_running_lifecycle_display_events() {
        let reducer_visible_events = vec![
            ExecutionEvent::ProcessingStarted("change-a".to_string()),
            ExecutionEvent::ProcessingError {
                id: "change-a".to_string(),
                error: "boom".to_string(),
            },
            ExecutionEvent::ApplyStarted {
                change_id: "change-a".to_string(),
                command: "apply".to_string(),
            },
            ExecutionEvent::ApplyCompleted {
                change_id: "change-a".to_string(),
                revision: "rev-a".to_string(),
            },
            ExecutionEvent::ApplyFailed {
                change_id: "change-a".to_string(),
                error: "boom".to_string(),
            },
            ExecutionEvent::AcceptanceStarted {
                change_id: "change-a".to_string(),
                command: "accept".to_string(),
            },
            ExecutionEvent::AcceptanceCompleted {
                change_id: "change-a".to_string(),
            },
            ExecutionEvent::AcceptanceFailed {
                change_id: "change-a".to_string(),
                error: "boom".to_string(),
            },
            ExecutionEvent::ArchiveStarted {
                change_id: "change-a".to_string(),
                command: "archive".to_string(),
            },
            ExecutionEvent::ArchiveResumed {
                change_id: "change-a".to_string(),
                reason: Some("resume".to_string()),
                summary: Some("resume archive".to_string()),
            },
            ExecutionEvent::ArchiveRetryScheduled {
                change_id: "change-a".to_string(),
                attempt: 1,
                max_attempts: 2,
                reason: Some("retry".to_string()),
                summary: Some("retry archive".to_string()),
            },
            ExecutionEvent::ChangeArchived("change-a".to_string()),
            ExecutionEvent::ArchiveFailed {
                change_id: "change-a".to_string(),
                error: "boom".to_string(),
                reason: Some("failed".to_string()),
                summary: Some("archive failed".to_string()),
            },
            ExecutionEvent::MergeDeferred {
                change_id: "change-a".to_string(),
                reason: "dirty base".to_string(),
                auto_resumable: true,
            },
            ExecutionEvent::MergeCompleted {
                change_id: "change-a".to_string(),
                revision: "rev-a".to_string(),
            },
            ExecutionEvent::ResolveStarted {
                change_id: "change-a".to_string(),
                command: "resolve".to_string(),
            },
            ExecutionEvent::ResolveCompleted {
                change_id: "change-a".to_string(),
                worktree_change_ids: None,
            },
            ExecutionEvent::ResolveFailed {
                change_id: "change-a".to_string(),
                error: "boom".to_string(),
            },
            ExecutionEvent::WorkspaceStatusUpdated {
                change_id: "change-a".to_string(),
                workspace_name: "ws-a".to_string(),
                status: WorkspaceStatus::Applying,
            },
            ExecutionEvent::RejectionReviewCompleted {
                change_id: "change-a".to_string(),
                outcome: RejectionOutcome::Resume,
            },
            ExecutionEvent::RejectionReviewFailed {
                change_id: "change-a".to_string(),
                error: "boom".to_string(),
            },
            ExecutionEvent::DependencyBlocked {
                change_id: "change-a".to_string(),
                dependency_ids: vec!["dep".to_string()],
            },
            ExecutionEvent::DependencyResolved {
                change_id: "change-a".to_string(),
            },
            ExecutionEvent::AcceptanceGated {
                change_id: "change-a".to_string(),
                blocker: stalled_blocker(),
            },
            ExecutionEvent::ExecutionBlocked {
                change_id: "change-a".to_string(),
                blocker: stalled_blocker(),
            },
            ExecutionEvent::ChangeDequeued {
                change_id: "change-a".to_string(),
            },
            ExecutionEvent::ChangeStopped {
                change_id: "change-a".to_string(),
            },
            ExecutionEvent::Stopped,
            empty_changes_refreshed_event(),
        ];

        for event in reducer_visible_events {
            assert!(
                should_apply_event_to_tui_reducer(&event),
                "event should sync to TUI reducer before display snapshot: {event:?}"
            );
        }
    }

    #[test]
    fn tui_reducer_sync_excludes_presentation_only_events() {
        let presentation_events = vec![
            ExecutionEvent::ApplyOutput {
                change_id: "change-a".to_string(),
                output: "chunk".to_string(),
                iteration: Some(1),
            },
            ExecutionEvent::ProgressUpdated {
                change_id: "change-a".to_string(),
                completed: 1,
                total: 2,
            },
            ExecutionEvent::Log(crate::events::LogEntry::info("hello")),
            ExecutionEvent::WorktreesRefreshed { worktrees: vec![] },
        ];

        for event in presentation_events {
            assert!(
                !should_apply_event_to_tui_reducer(&event),
                "presentation-only event should not sync to TUI reducer: {event:?}"
            );
        }
    }

    #[test]
    fn refresh_root_usable_for_existing_directory() {
        let temp_dir = tempfile::tempdir().expect("tempdir should be created");
        assert!(is_refresh_root_usable(temp_dir.path()));
    }

    #[test]
    fn refresh_root_not_usable_for_missing_directory() {
        let temp_dir = tempfile::tempdir().expect("tempdir should be created");
        let missing_path = temp_dir.path().join("missing-root");
        assert!(!is_refresh_root_usable(&missing_path));
    }

    #[test]
    fn stale_refresh_root_sets_warned_once_and_resets_when_root_recovers() {
        let temp_dir = tempfile::tempdir().expect("tempdir should be created");
        let missing_path = temp_dir.path().join("missing-root");
        let mut warned = false;

        assert!(super::should_skip_local_refresh(&missing_path, &mut warned));
        assert!(warned, "first stale root check should set warned flag");

        assert!(super::should_skip_local_refresh(&missing_path, &mut warned));
        assert!(warned, "second stale root check keeps warned flag set");

        assert!(!super::should_skip_local_refresh(
            temp_dir.path(),
            &mut warned
        ));
        assert!(!warned, "usable root should reset warned flag");
    }

    #[test]
    fn local_refresh_not_skipped_for_existing_root() {
        let temp_dir = tempfile::tempdir().expect("tempdir should be created");
        let mut warned = false;

        assert!(!super::should_skip_local_refresh(
            temp_dir.path(),
            &mut warned
        ));
        assert!(
            !warned,
            "existing root should not trigger stale warning suppression"
        );
    }

    #[test]
    fn refresh_local_changes_uses_explicit_repo_root_for_active_and_rejected_rows() {
        let _lock = crate::test_support::cwd_lock().lock().unwrap();
        let repo_dir = tempfile::tempdir().expect("repo tempdir");
        let other_dir = tempfile::tempdir().expect("cwd tempdir");
        let changes_dir = repo_dir.path().join("openspec").join("changes");

        let active_dir = changes_dir.join("change-active");
        std::fs::create_dir_all(&active_dir).expect("active dir");
        std::fs::write(active_dir.join("proposal.md"), "# proposal").expect("active proposal");
        std::fs::write(active_dir.join("tasks.md"), "- [ ] task").expect("active tasks");

        let rejected_dir = changes_dir.join("change-rejected");
        std::fs::create_dir_all(&rejected_dir).expect("rejected dir");
        std::fs::write(rejected_dir.join("proposal.md"), "# proposal").expect("rejected proposal");
        std::fs::write(rejected_dir.join("tasks.md"), "- [ ] task").expect("rejected tasks");
        std::fs::write(rejected_dir.join("REJECTED.md"), "# REJECTED").expect("rejected marker");

        let original_dir = std::env::current_dir().expect("cwd");
        std::env::set_current_dir(other_dir.path()).expect("set cwd elsewhere");

        let (active, rejected) = refresh_local_changes(repo_dir.path()).expect("refresh succeeds");

        std::env::set_current_dir(original_dir).expect("restore cwd");

        assert_eq!(
            active.iter().map(|c| c.id.as_str()).collect::<Vec<_>>(),
            vec!["change-active"]
        );
        assert_eq!(
            rejected.iter().map(|c| c.id.as_str()).collect::<Vec<_>>(),
            vec!["change-rejected"]
        );
    }

    async fn run_git_ok(cwd: &Path, args: &[&str]) {
        let output = tokio::process::Command::new("git")
            .args(args)
            .current_dir(cwd)
            .output()
            .await
            .expect("git command should start");
        assert!(
            output.status.success(),
            "git {:?} failed: {}",
            args,
            String::from_utf8_lossy(&output.stderr)
        );
    }

    /// A committed repository with deterministic identity, so `git status` has a
    /// HEAD to compare against and inherits no developer configuration.
    async fn init_dirty_state_repo(root: &Path) {
        run_git_ok(root, &["init", "-b", "main"]).await;
        run_git_ok(root, &["config", "user.email", "test@example.com"]).await;
        run_git_ok(root, &["config", "user.name", "Test User"]).await;
        std::fs::write(root.join("tracked.txt"), "base\n").expect("tracked file");
        run_git_ok(root, &["add", "."]).await;
        run_git_ok(root, &["commit", "-m", "base"]).await;
    }

    /// One refresh tick's observation, reduced to the fact the header renders.
    async fn observed_dirty(root: &Path) -> Option<bool> {
        let mut warned = false;
        observe_workspace_dirty(root, &mut warned)
            .await
            .map(|observation| match observation {
                TuiRefreshObservation::WorkspaceDirty { dirty } => dirty,
            })
    }

    /// Every form of operator work the badge is supposed to catch, and the
    /// return to clean that removes it again — on one real repository.
    #[tokio::test]
    async fn refresh_workspace_dirty_observes_staged_unstaged_and_untracked_work() {
        let repo = tempfile::tempdir().expect("repo tempdir");
        init_dirty_state_repo(repo.path()).await;

        assert_eq!(
            observed_dirty(repo.path()).await,
            Some(false),
            "a freshly committed repository is clean"
        );

        // Staged.
        std::fs::write(repo.path().join("staged.txt"), "staged\n").expect("staged file");
        run_git_ok(repo.path(), &["add", "staged.txt"]).await;
        assert_eq!(
            observed_dirty(repo.path()).await,
            Some(true),
            "a staged change is dirty"
        );

        run_git_ok(repo.path(), &["reset"]).await;
        std::fs::remove_file(repo.path().join("staged.txt")).expect("remove staged file");
        assert_eq!(observed_dirty(repo.path()).await, Some(false));

        // Unstaged tracked-file change, then the clean-after-dirty transition.
        std::fs::write(repo.path().join("tracked.txt"), "edited\n").expect("edit tracked file");
        assert_eq!(
            observed_dirty(repo.path()).await,
            Some(true),
            "an unstaged tracked-file change is dirty"
        );
        run_git_ok(repo.path(), &["checkout", "--", "tracked.txt"]).await;
        assert_eq!(
            observed_dirty(repo.path()).await,
            Some(false),
            "a later successful observation reports the workspace clean again"
        );

        // Untracked, under configuration that would otherwise suppress it.
        run_git_ok(repo.path(), &["config", "status.showUntrackedFiles", "no"]).await;
        std::fs::write(repo.path().join("untracked.txt"), "new\n").expect("untracked file");
        assert_eq!(
            observed_dirty(repo.path()).await,
            Some(true),
            "status.showUntrackedFiles=no must not make untracked work report as clean"
        );
    }

    /// Generated content is not operator work, so it must not raise the badge.
    #[tokio::test]
    async fn refresh_workspace_dirty_excludes_ignored_files() {
        let repo = tempfile::tempdir().expect("repo tempdir");
        init_dirty_state_repo(repo.path()).await;
        std::fs::write(repo.path().join(".gitignore"), "generated/\n").expect("gitignore");
        run_git_ok(repo.path(), &["add", ".gitignore"]).await;
        run_git_ok(repo.path(), &["commit", "-m", "ignore generated"]).await;

        std::fs::create_dir_all(repo.path().join("generated")).expect("generated dir");
        std::fs::write(repo.path().join("generated").join("out.txt"), "built\n")
            .expect("generated file");

        assert_eq!(
            observed_dirty(repo.path()).await,
            Some(false),
            "ignored files alone are not dirty"
        );
    }

    /// The badge reports the root the TUI captured at startup. A later process
    /// current-directory change is not allowed to retarget it.
    // The guard must span the awaits: it is what keeps a concurrent test from
    // seeing the process current directory this test moves.
    #[allow(clippy::await_holding_lock)]
    #[tokio::test]
    async fn refresh_workspace_dirty_uses_the_captured_root_not_the_process_cwd() {
        let _lock = crate::test_support::cwd_lock()
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        let captured = tempfile::tempdir().expect("captured tempdir");
        let other = tempfile::tempdir().expect("other tempdir");
        init_dirty_state_repo(captured.path()).await;
        init_dirty_state_repo(other.path()).await;
        std::fs::write(captured.path().join("untracked.txt"), "work\n").expect("dirty file");

        let original_dir = std::env::current_dir().expect("cwd");
        std::env::set_current_dir(other.path()).expect("set cwd elsewhere");

        let from_captured = observed_dirty(captured.path()).await;
        let from_other = observed_dirty(other.path()).await;

        std::env::set_current_dir(original_dir).expect("restore cwd");

        assert_eq!(
            from_captured,
            Some(true),
            "the captured root is dirty and must be what the header reports"
        );
        assert_eq!(
            from_other,
            Some(false),
            "the process current directory is clean, which is not the observed fact"
        );
    }

    /// A failed read has no answer, so it publishes nothing and the last
    /// successful observation stands. The warning is bounded per failure episode.
    #[tokio::test]
    async fn refresh_workspace_dirty_failure_preserves_the_last_successful_observation() {
        let repo = tempfile::tempdir().expect("repo tempdir");
        init_dirty_state_repo(repo.path()).await;
        std::fs::write(repo.path().join("untracked.txt"), "work\n").expect("dirty file");

        let mut app = AppState::new(vec![sample_change()]);
        let mut warned = false;
        let observation = observe_workspace_dirty(repo.path(), &mut warned)
            .await
            .expect("a readable repository observes");
        app.adopt_workspace_dirty_observation(observation);
        assert_eq!(
            app.workspace_dirty(),
            crate::tui::types::WorkspaceDirtyState::Dirty
        );
        assert!(!warned, "a successful read never warns");

        // A directory that is not a Git repository at all: the read fails.
        let unreadable = tempfile::tempdir().expect("non-repo tempdir");
        assert!(
            observe_workspace_dirty(unreadable.path(), &mut warned)
                .await
                .is_none(),
            "a failed read must publish no observation"
        );
        assert!(warned, "the first failure of an episode warns");
        assert!(observe_workspace_dirty(unreadable.path(), &mut warned)
            .await
            .is_none());
        assert!(
            warned,
            "the repeated failure stays inside the same bounded warning"
        );
        assert_eq!(
            app.workspace_dirty(),
            crate::tui::types::WorkspaceDirtyState::Dirty,
            "the failed reads must not have replaced the observation with clean"
        );

        // Recovery re-arms the bound, so a new failure episode is reported.
        let recovered = observe_workspace_dirty(repo.path(), &mut warned)
            .await
            .expect("the readable repository observes again");
        app.adopt_workspace_dirty_observation(recovered);
        assert!(!warned, "a success closes the failure episode");
        assert_eq!(
            app.workspace_dirty(),
            crate::tui::types::WorkspaceDirtyState::Dirty
        );
    }

    #[tokio::test]
    async fn shutdown_local_orchestrator_cancels_and_aborts_non_finishing_task() {
        let token = CancellationToken::new();
        let task_token = token.clone();
        let (cancelled_tx, cancelled_rx) = tokio::sync::oneshot::channel();
        let (post_abort_tx, mut post_abort_rx) = tokio::sync::mpsc::channel::<()>(1);

        let handle = tokio::spawn(async move {
            task_token.cancelled().await;
            let _ = cancelled_tx.send(());
            tokio::time::sleep(Duration::from_secs(60)).await;
            loop {
                let _ = post_abort_tx.send(()).await;
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
            #[allow(unreachable_code)]
            Ok(())
        });

        let outcome = shutdown_local_orchestrator_task(
            Some(handle),
            Some(token),
            None,
            Duration::from_millis(10),
        )
        .await;

        assert_eq!(
            outcome,
            LocalOrchestratorShutdownOutcome::AbortedAfterTimeout
        );
        assert!(
            cancelled_rx.await.is_ok(),
            "shutdown should cancel orchestrator token"
        );
        tokio::task::yield_now().await;
        let post_cleanup_event =
            tokio::time::timeout(Duration::from_millis(80), post_abort_rx.recv()).await;
        assert!(
            !matches!(post_cleanup_event, Ok(Some(()))),
            "aborted local orchestrator must not keep sending events after cleanup"
        );
    }

    #[tokio::test]
    async fn shutdown_local_orchestrator_without_handle_is_noop() {
        let outcome =
            shutdown_local_orchestrator_task(None, None, None, Duration::from_millis(1)).await;

        assert_eq!(outcome, LocalOrchestratorShutdownOutcome::NoTask);
    }

    /// Local quit is a cancellation boundary for the run this TUI started.
    ///
    /// Three things had to hold together and previously did not: the command
    /// scope must close at cancellation time rather than after a grace period,
    /// that grace must not expire before the scheduler's own cancellation
    /// boundary has had its chance, and a timeout must force-clean the retained
    /// process identities *before* the task is aborted — because abort destroys
    /// the only in-task path to them and is not cleanup evidence by itself.
    ///
    /// The real process boundary is covered by
    /// `run_scope_tui_quit_cleans_process_group_after_timeout` in
    /// `tests/process_cleanup_test.rs`; this pins the control flow.
    #[tokio::test]
    async fn local_tui_shutdown_waits_for_run_command_scope() {
        use crate::ai_command_runner::RunCommandScope;
        use crate::tui::orchestrator::PARALLEL_CANCELLATION_CLEANUP_DEADLINE;

        // The grace must not undercut the scheduler's outer cancellation
        // boundary, or the TUI aborts the very cleanup it is waiting for.
        assert!(
            super::LOCAL_ORCHESTRATOR_SHUTDOWN_GRACE >= PARALLEL_CANCELLATION_CLEANUP_DEADLINE,
            "local shutdown grace {:?} undercuts the scheduler boundary {:?}",
            super::LOCAL_ORCHESTRATOR_SHUTDOWN_GRACE,
            PARALLEL_CANCELLATION_CLEANUP_DEADLINE
        );

        // Graceful completion: the scope closes at cancellation time, and the
        // task is joined rather than aborted.
        let scope = RunCommandScope::new();
        let token = CancellationToken::new();
        let task_token = token.clone();
        let handle = tokio::spawn(async move {
            task_token.cancelled().await;
            Ok(())
        });
        let outcome = shutdown_local_orchestrator_task(
            Some(handle),
            Some(token),
            Some(scope.clone()),
            Duration::from_secs(5),
        )
        .await;
        assert_eq!(outcome, LocalOrchestratorShutdownOutcome::Graceful);
        assert!(
            scope.is_closed() && scope.cancel_token().is_cancelled(),
            "quit closes run command admission immediately, not after the grace period"
        );

        // Timeout escalation: a retained owned identity is force-cleaned and
        // verified before the task is aborted.
        let dead = std::process::Command::new("sh")
            .arg("-c")
            .arg("exit 0")
            .spawn()
            .expect("spawn");
        let pid = dead.id();
        let mut dead = dead;
        let _ = dead.wait();

        let scope = RunCommandScope::new();
        scope
            .register_unproven_for_test("apply", Some("change-a"), pid)
            .release_confirmed();
        assert_eq!(scope.retained_process_ids(), vec![pid]);

        let token = CancellationToken::new();
        let handle = tokio::spawn(async move {
            std::future::pending::<()>().await;
            Ok(())
        });
        let outcome = shutdown_local_orchestrator_task(
            Some(handle),
            Some(token),
            Some(scope.clone()),
            Duration::from_millis(20),
        )
        .await;
        assert_eq!(
            outcome,
            LocalOrchestratorShutdownOutcome::AbortedAfterTimeout,
            "timeout stays distinguishable from graceful completion"
        );
        assert!(
            scope.retained_process_ids().is_empty(),
            "retained owned identities must be cleaned and verified before task abort"
        );

        // Remote mode: no local run means no handle and no scope, and nothing
        // here reaches the remote server.
        assert_eq!(
            shutdown_local_orchestrator_task(None, None, None, Duration::from_millis(1)).await,
            LocalOrchestratorShutdownOutcome::NoTask
        );
    }
}