zeph 0.22.2

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

#[cfg(feature = "tui")]
use std::time::Duration;

#[cfg(feature = "tui")]
use crate::bootstrap::warmup_provider;
#[cfg(feature = "tui")]
use crate::channel::TuiHandle;
#[cfg(feature = "tui")]
use zeph_core::channel::Channel;
#[cfg(feature = "tui")]
use zeph_llm::any::AnyProvider;

#[cfg(feature = "tui")]
pub(crate) struct TuiRunParams<'a> {
    pub(crate) tui_handle: TuiHandle,
    pub(crate) config: &'a zeph_core::config::Config,
    pub(crate) status_rx: Option<tokio::sync::mpsc::UnboundedReceiver<String>>,
    pub(crate) tool_rx: Option<tokio::sync::mpsc::Receiver<zeph_tools::ToolEvent>>,
    pub(crate) metrics_rx:
        Option<tokio::sync::watch::Receiver<zeph_core::metrics::MetricsSnapshot>>,
    pub(crate) warmup_provider: AnyProvider,
    pub(crate) index_progress_rx: Option<tokio::sync::watch::Receiver<zeph_index::IndexProgress>>,
    /// Whether --tafc CLI flag was passed (overrides config).
    pub(crate) cli_tafc: bool,
    /// Set when TUI rendering was started early via `start_tui_early`.
    /// When `Some`, `run_tui_agent` skips creating a new TUI task and uses the existing one.
    pub(crate) early_tui: Option<EarlyTuiHandle>,
    /// Watch receiver for embed backfill progress.
    /// `None` = idle/done; `Some(p)` = backfill running with progress `p`.
    pub(crate) backfill_rx:
        tokio::sync::watch::Receiver<Option<zeph_memory::semantic::BackfillProgress>>,
    /// Optional supervisor passed to the TUI task registry panel (#2962).
    pub(crate) task_supervisor: Option<zeph_common::task_supervisor::TaskSupervisor>,
    /// Fleet session ID to mark completed/failed when the TUI session exits.
    pub(crate) fleet_session_id: String,
    /// URI that triggered this session via `url-open` deep-link dispatch.
    ///
    /// When `Some`, a one-shot status notification is emitted in the TUI status area within
    /// 1 s of launch per spec §9 (TASK-8).
    #[cfg(feature = "deep-link")]
    pub(crate) deep_link_uri: Option<String>,
    /// Pre-formatted "Resuming session" banner text (spec-068 §13.5), sent as
    /// `AgentEvent::ResumeBanner` once the TUI's `agent_tx` is available. `None` for a fresh
    /// conversation — no banner is sent in that case (AC-16).
    pub(crate) resume_banner: Option<String>,
}

/// Phase-1 TUI handle: TUI is rendering but the agent hasn't started yet.
#[cfg(feature = "tui")]
pub(crate) struct EarlyTuiHandle {
    /// Oneshot receiver that fires when the TUI thread finishes.
    pub(crate) tui_done: tokio::sync::oneshot::Receiver<anyhow::Result<()>>,
    /// Send status/event updates to the TUI during setup.
    pub(crate) agent_tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
}

/// Resolve the TUI [`Theme`](zeph_tui::theme::Theme) from the config section `[tui.theme]`.
///
/// Returns the resolved theme, the palette name (for cycle tracking), and the effective colour
/// mode (for re-derivation on runtime swap without re-running OS detection). Falls back to the
/// default theme on any error so startup always succeeds.
#[cfg(feature = "tui")]
fn build_tui_theme(
    config: &zeph_core::config::Config,
) -> (
    zeph_tui::theme::Theme,
    String,
    zeph_tui::theme::EffectiveColorMode,
) {
    use zeph_tui::theme::{EffectiveColorMode, Theme, resolve_color_mode, resolve_palette};
    let theme_cfg = &config.tui.theme;
    let mode = resolve_color_mode(theme_cfg.color_mode);
    match resolve_palette(&theme_cfg.name) {
        Ok(p) => (
            Theme::from_palette_with_mode(&p, mode),
            theme_cfg.name.clone(),
            mode,
        ),
        Err(e) => {
            tracing::warn!("TUI theme '{}' could not be loaded: {e}", theme_cfg.name);
            (
                Theme::default(),
                "zephyr".to_owned(),
                EffectiveColorMode::Truecolor,
            )
        }
    }
}

/// Start TUI rendering immediately (Phase 1).
///
/// Extracts `agent_rx` from `tui_handle`, creates the TUI `App`, spawns the rendering task,
/// and sends an initial "Starting up..." status message. The caller continues agent setup and
/// calls `run_tui_agent` (Phase 2) once ready.
///
/// Index progress forwarding is wired separately via `forward_index_progress_to_tui` once
/// `index_progress_rx` becomes available (after `apply_code_indexer`).
///
/// # Panics
///
/// Panics if `tui_handle.agent_rx` is `None` (already taken by a previous call).
#[cfg(feature = "tui")]
pub(crate) fn start_tui_early(
    tui_handle: &mut TuiHandle,
    config: &zeph_core::config::Config,
) -> EarlyTuiHandle {
    let (event_tx, event_rx) = tokio::sync::mpsc::channel(256);
    let reader = zeph_tui::EventReader::new(event_tx, Duration::from_millis(100));
    std::thread::spawn(move || reader.run());

    let agent_rx = tui_handle
        .agent_rx
        .take()
        .expect("agent_rx already taken by start_tui_early");
    let (tui_theme, tui_theme_name, tui_color_mode) = build_tui_theme(config);
    let mut tui_app = zeph_tui::App::new(tui_handle.user_tx.clone(), agent_rx)
        .with_command_tx(tui_handle.command_tx.clone())
        .with_tool_density(config.tui.tool_density)
        .with_theme(tui_theme)
        .with_theme_name(tui_theme_name)
        .with_effective_color_mode(tui_color_mode)
        .with_motion(config.tui.motion)
        .with_delights(config.tui.delights.clone())
        .with_mouse(config.tui.mouse);
    tui_app.set_show_source_labels(config.tui.show_source_labels);
    tui_app.set_show_balance(config.cocoon.show_balance);

    let agent_tx = tui_handle.agent_tx.clone();

    // Send initial loading status directly — channel is empty at this point (capacity 256).
    let _ = agent_tx.try_send(zeph_tui::AgentEvent::Status("Starting up...".into()));

    let (done_tx, done_rx) = tokio::sync::oneshot::channel::<anyhow::Result<()>>();
    std::thread::Builder::new()
        .name("zeph-tui".into())
        .spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("tui runtime");
            let result = rt.block_on(async move {
                zeph_tui::run_tui(tui_app, event_rx).await?;
                Ok(())
            });
            let _ = done_tx.send(result);
        })
        .expect("spawn tui thread");

    EarlyTuiHandle {
        tui_done: done_rx,
        agent_tx,
    }
}

/// Warms up the provider, signals readiness, then shows embed backfill status until done.
///
/// After warmup completes and the "model ready" message clears, this task monitors
/// `backfill_rx` and keeps the TUI status bar showing "Backfilling embeddings..." until
/// the backfill finishes. This avoids the status being overwritten by subsequent init steps
/// during the startup sequence.
#[cfg(feature = "tui")]
async fn spawn_warmup_with_backfill_status(
    provider: AnyProvider,
    mut backfill_rx: tokio::sync::watch::Receiver<Option<zeph_memory::semantic::BackfillProgress>>,
    warmup_tx: tokio::sync::watch::Sender<bool>,
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
) {
    let _ = tx
        .send(zeph_tui::AgentEvent::Status("warming up model...".into()))
        .await;
    warmup_provider(&provider).await;
    let _ = tx
        .send(zeph_tui::AgentEvent::Status("model ready".into()))
        .await;
    let _ = warmup_tx.send(true);
    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
    let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;
    // After init status clears, show backfill progress until it finishes.
    loop {
        let progress = *backfill_rx.borrow_and_update();
        if let Some(p) = progress {
            let pct = (p.done * 100).checked_div(p.total).unwrap_or(0);
            let _ = tx
                .send(zeph_tui::AgentEvent::Status(format!(
                    "Backfilling embeddings: {}/{} ({}%)",
                    p.done, p.total, pct
                )))
                .await;
        } else {
            let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;
            break;
        }
        if backfill_rx.changed().await.is_err() {
            break;
        }
    }
}

/// Spawn the TUI render thread (legacy path: no `EarlyTuiHandle`).
///
/// Creates the [`zeph_tui::App`], wires optional receivers, and spawns the TUI on a
/// dedicated OS thread with its own `current_thread` tokio runtime so that
/// `terminal.draw()` never blocks a shared tokio worker.
///
/// Returns a oneshot receiver that fires when the thread exits.
#[cfg(feature = "tui")]
// Cannot split: all arguments configure the App before the thread is spawned and there is
// no natural grouping that would not create an ad-hoc builder used only here.
#[allow(clippy::too_many_arguments)]
fn spawn_tui_thread(
    user_tx: tokio::sync::mpsc::Sender<String>,
    agent_rx: tokio::sync::mpsc::Receiver<zeph_tui::AgentEvent>,
    command_tx: tokio::sync::mpsc::Sender<zeph_tui::TuiCommand>,
    cancel_signal: std::sync::Arc<tokio::sync::Notify>,
    show_source_labels: bool,
    show_balance: bool,
    tool_density: zeph_config::ToolDensity,
    motion: zeph_config::Motion,
    delights: zeph_config::DelightsConfig,
    mouse: bool,
    theme: zeph_tui::theme::Theme,
    theme_name: String,
    effective_color_mode: zeph_tui::theme::EffectiveColorMode,
    metrics_rx: Option<tokio::sync::watch::Receiver<zeph_core::metrics::MetricsSnapshot>>,
    task_supervisor: Option<zeph_common::task_supervisor::TaskSupervisor>,
    index_progress_rx: Option<tokio::sync::watch::Receiver<zeph_index::IndexProgress>>,
    agent_tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
) -> tokio::sync::oneshot::Receiver<anyhow::Result<()>> {
    let (event_tx, event_rx) = tokio::sync::mpsc::channel(256);
    let reader = zeph_tui::EventReader::new(event_tx, Duration::from_millis(100));
    std::thread::spawn(move || reader.run());

    let mut tui_app = zeph_tui::App::new(user_tx, agent_rx)
        .with_cancel_signal(cancel_signal)
        .with_command_tx(command_tx)
        .with_tool_density(tool_density)
        .with_motion(motion)
        .with_delights(delights)
        .with_mouse(mouse)
        .with_theme(theme)
        .with_theme_name(theme_name)
        .with_effective_color_mode(effective_color_mode);
    tui_app.set_show_source_labels(show_source_labels);
    tui_app.set_show_balance(show_balance);

    if let Some(rx) = metrics_rx {
        tui_app = tui_app.with_metrics_rx(rx);
    }

    if let Some(supervisor) = task_supervisor {
        tui_app = tui_app.with_task_supervisor(supervisor);
    }

    if let Some(progress_rx) = index_progress_rx {
        tokio::spawn(forward_index_progress_to_tui(progress_rx, agent_tx)); // EXEMPT(#5143): self-terminating forwarder inside TUI thread LocalSet context
    }

    let (done_tx, done_rx) = tokio::sync::oneshot::channel::<anyhow::Result<()>>();
    std::thread::Builder::new()
        .name("zeph-tui".into())
        .spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("tui runtime");
            let result = rt.block_on(async move {
                zeph_tui::run_tui(tui_app, event_rx)
                    .await
                    .map_err(anyhow::Error::from)
            });
            let _ = done_tx.send(result);
        })
        .expect("spawn tui thread");
    done_rx
}

#[cfg(feature = "tui")]
fn tui_command_context(config: &zeph_core::config::Config, cli_tafc: bool) -> TuiCommandContext {
    TuiCommandContext {
        provider: format!("{:?}", config.llm.effective_provider()),
        model: config.llm.effective_model().to_owned(),
        agent_name: config.agent.name.clone(),
        semantic_enabled: config.memory.semantic.enabled,
        autonomy_level: format!("{:?}", config.security.autonomy_level),
        max_tool_iterations: config.agent.max_tool_iterations,
        tafc_enabled: config.tools.tafc.enabled || cli_tafc,
        tafc_complexity_threshold: config.tools.tafc.complexity_threshold,
        sandbox_backend: if config.tools.sandbox.enabled {
            #[cfg(target_os = "macos")]
            {
                Some("macos-seatbelt".to_owned())
            }
            #[cfg(all(target_os = "linux", feature = "sandbox"))]
            {
                Some("linux-bwrap-landlock".to_owned())
            }
            #[cfg(not(any(target_os = "macos", all(target_os = "linux", feature = "sandbox"))))]
            {
                Some("noop".to_owned())
            }
        } else {
            None
        },
        sandbox_denied_domains_count: config.tools.sandbox.denied_domains.len(),
        sandbox_fail_if_unavailable: config.tools.sandbox.fail_if_unavailable,
    }
}

#[cfg(feature = "tui")]
#[allow(clippy::too_many_lines)]
pub(crate) async fn run_tui_agent<C: Channel + 'static>(
    agent: zeph_core::agent::Agent<C>,
    mut params: TuiRunParams<'_>,
) -> anyhow::Result<()> {
    // Destructure handle fields needed regardless of path.
    let TuiHandle {
        user_tx,
        agent_tx: handle_agent_tx,
        agent_rx,
        command_tx,
        command_rx,
    } = params.tui_handle;

    // Determine TUI done-signal: reuse early-started thread or spawn a new one.
    let (tui_done, agent_tx) = if let Some(early) = params.early_tui {
        // Phase-2 path: TUI is already rendering. Wire cancel signal and metrics
        // into the running App via AgentEvent so Ctrl+C and metrics panel work correctly.
        drop(user_tx);
        drop(agent_rx);
        drop(command_tx);
        let _ = early
            .agent_tx
            .try_send(zeph_tui::AgentEvent::SetCancelSignal(agent.cancel_signal()));
        if let Some(metrics_rx) = params.metrics_rx {
            let _ = early
                .agent_tx
                .try_send(zeph_tui::AgentEvent::SetMetricsRx(metrics_rx));
        }
        if let Some(task_supervisor) = params.task_supervisor.take() {
            let _ = early
                .agent_tx
                .try_send(zeph_tui::AgentEvent::SetTaskSupervisor(task_supervisor));
        }
        (early.tui_done, early.agent_tx)
    } else {
        // Legacy path: TUI hasn't started yet, create App and spawn its thread now.
        let (legacy_theme, legacy_theme_name, legacy_color_mode) = build_tui_theme(params.config);
        let done_rx = spawn_tui_thread(
            user_tx,
            agent_rx.expect("agent_rx not set in TuiHandle"),
            command_tx,
            agent.cancel_signal(),
            params.config.tui.show_source_labels,
            params.config.cocoon.show_balance,
            params.config.tui.tool_density,
            params.config.tui.motion,
            params.config.tui.delights.clone(),
            params.config.tui.mouse,
            legacy_theme,
            legacy_theme_name,
            legacy_color_mode,
            params.metrics_rx.take(),
            params.task_supervisor.take(),
            params.index_progress_rx.take(),
            handle_agent_tx.clone(),
        );
        (done_rx, handle_agent_tx)
    };

    if let Some(banner) = params.resume_banner.take() {
        let _ = agent_tx.try_send(zeph_tui::AgentEvent::ResumeBanner(banner));
    }

    // Track all forwarding tasks so we can abort them when the agent exits,
    // ensuring the agent_event channel closes and the TUI thread quits.
    let mut forwarders = tokio::task::JoinSet::new();

    if let Some(rx) = params.status_rx {
        forwarders.spawn(forward_status_to_tui(rx, agent_tx.clone()));
    }
    // else: early forwarder already owns status_rx and is draining it
    forwarders.spawn(forward_tui_commands(
        command_rx,
        agent_tx.clone(),
        tui_command_context(params.config, params.cli_tafc),
    ));

    {
        let fleet_cfg = params.config.tui.fleet;
        let db_path = params.config.memory.sqlite_path.clone();
        forwarders.spawn(fleet_poll_task(db_path, fleet_cfg, agent_tx.clone()));
    }

    if params.config.durable.enabled {
        let durable_cfg = params.config.durable.clone();
        let durable_url = crate::commands::durable::resolve_durable_db_url(params.config);
        forwarders.spawn(durable_poll_task(
            durable_url,
            durable_cfg,
            agent_tx.clone(),
        ));
    }

    if let Some(tool_rx) = params.tool_rx {
        forwarders.spawn(forward_tool_events_to_tui(tool_rx, agent_tx.clone()));
    }

    let (warmup_tx, warmup_rx) = tokio::sync::watch::channel(false);
    forwarders.spawn(spawn_warmup_with_backfill_status(
        params.warmup_provider,
        params.backfill_rx,
        warmup_tx,
        agent_tx.clone(),
    ));

    // TASK-8: emit a one-shot deep-link notification within 1 s of launch.
    // Tracked in `forwarders` so it is aborted cleanly when the TUI or agent exits.
    #[cfg(all(feature = "deep-link", feature = "tui"))]
    if let Some(uri) = params.deep_link_uri.take() {
        forwarders.spawn(deep_link_notification_task(agent_tx.clone(), uri));
    }

    let mut agent = agent.with_warmup_ready(warmup_rx);
    let agent_future = agent.run();

    let run_result: anyhow::Result<()> = tokio::select! {
        result = tui_done => {
            forwarders.abort_all();
            agent.shutdown().await;
            result.map_err(|_| anyhow::anyhow!("TUI thread exited without sending result"))?
        }
        result = agent_future => {
            // Abort all forwarding tasks first, then drop our agent_tx clone.
            // Once all senders are gone the agent_event_rx channel closes, which
            // causes poll_agent_event to return None and the TUI thread to quit.
            forwarders.abort_all();
            drop(agent_tx);
            agent.shutdown().await;
            result.map_err(anyhow::Error::from)
        }
    };

    let db_path = params.config.memory.sqlite_path.clone();
    match zeph_memory::store::SqliteStore::new(&db_path).await {
        Ok(store) => {
            crate::fleet_session::end_session(&store, &params.fleet_session_id, &run_result).await;
        }
        Err(e) => {
            tracing::warn!(error = %e, "fleet: failed to open DB for end_session on TUI exit");
        }
    }

    run_result
}

/// Emits a deep-link launch notification in the TUI status bar, then clears it after 3 s.
#[cfg(all(feature = "deep-link", feature = "tui"))]
async fn deep_link_notification_task(
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
    uri: String,
) {
    tokio::time::sleep(std::time::Duration::from_millis(800)).await;
    let _ = tx
        .send(zeph_tui::AgentEvent::Status(format!(
            "Opened via deep link: {uri}"
        )))
        .await;
    tokio::time::sleep(std::time::Duration::from_secs(3)).await;
    let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;
}

pub(crate) async fn forward_status_to_stderr(mut rx: tokio::sync::mpsc::UnboundedReceiver<String>) {
    while let Some(msg) = rx.recv().await {
        eprintln!("[status] {msg}");
    }
}

// SECURITY: non-secret fields only
#[cfg(feature = "tui")]
pub(crate) struct TuiCommandContext {
    pub(crate) provider: String,
    pub(crate) model: String,
    pub(crate) agent_name: String,
    pub(crate) semantic_enabled: bool,
    pub(crate) autonomy_level: String,
    pub(crate) max_tool_iterations: usize,
    pub(crate) tafc_enabled: bool,
    pub(crate) tafc_complexity_threshold: f64,
    /// Active sandbox backend name (e.g. `"macos-seatbelt"`, `"linux-bwrap-landlock"`, `"noop"`).
    /// `None` when sandbox is disabled.
    pub(crate) sandbox_backend: Option<String>,
    /// Number of entries in `[tools.sandbox].denied_domains`.
    pub(crate) sandbox_denied_domains_count: usize,
    /// Whether `fail_if_unavailable` is set in config.
    pub(crate) sandbox_fail_if_unavailable: bool,
}

#[cfg(feature = "tui")]
pub(crate) async fn forward_tui_commands(
    mut rx: tokio::sync::mpsc::Receiver<zeph_tui::TuiCommand>,
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
    ctx: TuiCommandContext,
) {
    while let Some(cmd) = rx.recv().await {
        let (command_id, output) = match cmd {
            zeph_tui::TuiCommand::ViewConfig => {
                let text = format!(
                    "Active configuration:\n  Provider: {}\n  Model: {}\n  Agent name: {}\n  Semantic enabled: {}",
                    ctx.provider, ctx.model, ctx.agent_name, ctx.semantic_enabled,
                );
                ("view:config".to_owned(), text)
            }
            zeph_tui::TuiCommand::ViewAutonomy => {
                let text = format!(
                    "Autonomy level: {}\n  Max tool iterations: {}",
                    ctx.autonomy_level, ctx.max_tool_iterations,
                );
                ("view:autonomy".to_owned(), text)
            }
            zeph_tui::TuiCommand::TafcStatus => {
                let text = if ctx.tafc_enabled {
                    format!(
                        "TAFC (Think-Augmented Function Calling): enabled\n  \
                         Complexity threshold: {:.2}\n  \
                         Note: changing TAFC settings mid-session causes a prompt cache miss.",
                        ctx.tafc_complexity_threshold,
                    )
                } else {
                    "TAFC (Think-Augmented Function Calling): disabled\n  \
                     Enable with --tafc CLI flag or [tools.tafc] enabled = true in config."
                        .to_owned()
                };
                ("tafc:status".to_owned(), text)
            }
            zeph_tui::TuiCommand::SandboxStatus => {
                let text = match &ctx.sandbox_backend {
                    None => "Sandbox: disabled\n  Set [tools.sandbox] enabled = true to enable."
                        .to_owned(),
                    Some(backend) => {
                        let egress = if ctx.sandbox_denied_domains_count == 0 {
                            "no denied domains configured".to_owned()
                        } else {
                            format!("{} denied domain(s)", ctx.sandbox_denied_domains_count)
                        };
                        let fail_str = if ctx.sandbox_fail_if_unavailable {
                            "yes"
                        } else {
                            "no"
                        };
                        format!(
                            "Sandbox: enabled\n  Backend: {backend}\n  \
                             Egress filter: {egress}\n  \
                             fail_if_unavailable: {fail_str}"
                        )
                    }
                };
                ("sandbox:status".to_owned(), text)
            }
            _ => continue,
        };
        if tx
            .send(zeph_tui::AgentEvent::CommandResult { command_id, output })
            .await
            .is_err()
        {
            break;
        }
    }
}

#[cfg(feature = "tui")]
pub(crate) async fn forward_status_to_tui(
    mut rx: tokio::sync::mpsc::UnboundedReceiver<String>,
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
) {
    while let Some(msg) = rx.recv().await {
        if tx.send(zeph_tui::AgentEvent::Status(msg)).await.is_err() {
            break;
        }
    }
}

#[cfg(feature = "tui")]
pub(crate) async fn forward_tool_events_to_tui(
    mut rx: tokio::sync::mpsc::Receiver<zeph_tools::ToolEvent>,
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
) {
    // Only forward streaming chunks. ToolStart and ToolOutput are already sent via
    // TuiChannel::send_tool_start / send_tool_output from the Channel trait — forwarding
    // Started and Completed here would duplicate those events in the TUI.
    while let Some(event) = rx.recv().await {
        let agent_event = match event {
            zeph_tools::ToolEvent::Started { .. } | zeph_tools::ToolEvent::Completed { .. } => {
                continue;
            }
            zeph_tools::ToolEvent::OutputChunk {
                tool_name,
                command,
                chunk,
                tool_call_id,
                ..
            } => zeph_tui::AgentEvent::ToolOutputChunk {
                tool_name,
                command,
                chunk: zeph_tools::strip_ansi(&chunk),
                tool_call_id,
            },
            zeph_tools::ToolEvent::Rollback {
                restored_count,
                deleted_count,
                ..
            } => zeph_tui::AgentEvent::Status(format!(
                "Rolled back {restored_count} file(s), deleted {deleted_count} new file(s)"
            )),
            _ => continue,
        };
        if tx.send(agent_event).await.is_err() {
            break;
        }
    }
}

#[cfg(all(test, feature = "tui"))]
mod tests {
    use super::*;

    #[tokio::test]
    async fn forward_status_to_tui_delivers_messages() {
        let (status_tx, status_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let (agent_tx, mut agent_rx) = tokio::sync::mpsc::channel::<zeph_tui::AgentEvent>(16);

        tokio::spawn(forward_status_to_tui(status_rx, agent_tx)); // EXEMPT(#5143): test-only spawn

        status_tx.send("Connecting tools...".into()).unwrap();
        status_tx.send("Memory ready".into()).unwrap();
        drop(status_tx);

        let mut received = Vec::new();
        while let Some(ev) = agent_rx.recv().await {
            received.push(ev);
        }

        assert_eq!(received.len(), 2);
        assert!(
            matches!(&received[0], zeph_tui::AgentEvent::Status(s) if s == "Connecting tools...")
        );
        assert!(matches!(&received[1], zeph_tui::AgentEvent::Status(s) if s == "Memory ready"));
    }

    #[tokio::test]
    async fn forward_status_to_tui_stops_when_agent_rx_dropped() {
        let (status_tx, status_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
        let (agent_tx, agent_rx) = tokio::sync::mpsc::channel::<zeph_tui::AgentEvent>(1);

        let handle = tokio::spawn(forward_status_to_tui(status_rx, agent_tx)); // EXEMPT(#5143): test-only spawn

        // Drop receiver — forwarder must exit cleanly when send fails.
        drop(agent_rx);

        status_tx.send("some status".into()).unwrap();
        // Give the forwarder a chance to detect the closed channel.
        handle.await.expect("forwarder panicked");
    }

    #[tokio::test]
    async fn forward_tool_events_skips_started_and_completed() {
        let (tool_tx, tool_rx) = tokio::sync::mpsc::channel::<zeph_tools::ToolEvent>(64);
        let (agent_tx, mut agent_rx) = tokio::sync::mpsc::channel::<zeph_tui::AgentEvent>(16);

        tokio::spawn(forward_tool_events_to_tui(tool_rx, agent_tx)); // EXEMPT(#5143): test-only spawn

        tool_tx
            .send(zeph_tools::ToolEvent::Started {
                tool_name: "shell".into(),
                command: "ls".into(),
                sandbox_profile: None,
                resolved_cwd: None,
                execution_env: None,
            })
            .await
            .unwrap();
        tool_tx
            .send(zeph_tools::ToolEvent::OutputChunk {
                tool_call_id: String::new(),
                tool_name: "shell".into(),
                command: "ls".into(),
                chunk: "file.txt\n".into(),
                skill_name: None,
            })
            .await
            .unwrap();
        tool_tx
            .send(zeph_tools::ToolEvent::Completed {
                tool_name: "shell".into(),
                command: "ls".into(),
                output: "file.txt\n".into(),
                success: true,
                diff: None,
                filter_stats: None,
                run_id: None,
            })
            .await
            .unwrap();
        drop(tool_tx);

        let mut received = Vec::new();
        while let Some(ev) = agent_rx.recv().await {
            received.push(ev);
        }

        assert_eq!(
            received.len(),
            1,
            "expected exactly one event (OutputChunk)"
        );
        assert!(
            matches!(received[0], zeph_tui::AgentEvent::ToolOutputChunk { .. }),
            "expected ToolOutputChunk, got {:?}",
            received[0]
        );
    }

    /// Regression for #6041: an `encryption_gate` (INV-8) rejection must produce a
    /// `DurableStatus::GateRejected` snapshot, not the same `Unavailable` status used for an
    /// ordinary failed-to-open journal — the two failure modes must stay distinguishable in the
    /// TUI. Uses a `postgres://`-scheme URL with `encrypt_payload = false` (default `true`) so
    /// `enforce_encryption_gate` rejects deterministically without needing a real Postgres server;
    /// the gate check runs before any backend connection is attempted.
    #[tokio::test]
    async fn durable_poll_task_sends_gate_rejected_on_encryption_gate_failure() {
        use zeph_tui::widgets::durable::DurableStatus;

        let (tx, mut rx) = tokio::sync::mpsc::channel::<zeph_tui::AgentEvent>(4);
        let cfg = zeph_config::DurableConfig {
            encrypt_payload: false,
            key_id: 7,
            previous_key_id: Some(6),
            ..zeph_config::DurableConfig::default()
        };

        durable_poll_task("postgres://user@host/db".to_owned(), cfg, tx).await;

        let event = rx
            .recv()
            .await
            .expect("expected a DurableSnapshot event on gate rejection");
        match event {
            zeph_tui::AgentEvent::DurableSnapshot(snapshot) => {
                assert_eq!(
                    snapshot.status,
                    DurableStatus::GateRejected,
                    "gate rejection must be distinguishable from a plain Unavailable snapshot"
                );
                assert!(snapshot.executions.is_empty());
                // Regression for #6450: key_id/previous_key_id must reach the panel even on the
                // early-return gate-rejected branch, not just the happy-path Available branch.
                assert_eq!(snapshot.key_id, 7);
                assert_eq!(snapshot.previous_key_id, Some(6));
            }
            other => panic!("expected DurableSnapshot event, got {other:?}"),
        }
        assert!(
            rx.recv().await.is_none(),
            "durable_poll_task must return immediately after a gate rejection, not loop"
        );
    }

    /// Regression for #6450: the `Unavailable` early-return branch (journal file/backend could
    /// not be opened) must also carry `key_id`/`previous_key_id` through — a third, easy-to-miss
    /// construction site alongside `GateRejected` and the happy-path `Available` snapshot.
    #[tokio::test]
    async fn durable_poll_task_sends_key_ids_on_unavailable_backend_open_failure() {
        use zeph_tui::widgets::durable::DurableStatus;

        let (tx, mut rx) = tokio::sync::mpsc::channel::<zeph_tui::AgentEvent>(4);
        let cfg = zeph_config::DurableConfig {
            key_id: 9,
            previous_key_id: Some(8),
            ..zeph_config::DurableConfig::default()
        };

        // A bare absolute path (no `sqlite://` scheme — `LocalBackend::open`/`connect_sqlite`
        // takes a raw filesystem path, not a URL, and prepends its own `sqlite:` prefix) under a
        // root-owned directory that a non-root test process cannot create. This fails
        // `LocalBackend::open` via a clean permission-denied on `create_dir_all`, without needing
        // a real Postgres/shared-DB setup and without leaving stray files behind. (A prior version
        // of this test used a `sqlite://`-prefixed string here, which `connect_sqlite` treated as
        // a literal relative path — `PathBuf::from("sqlite:///nonexistent/...")` — and created a
        // real `sqlite:/nonexistent/` directory in the process cwd before the resulting
        // double-prefixed URL failed to parse.)
        durable_poll_task("/nonexistent/durable-poll-test.db".to_owned(), cfg, tx).await;

        let event = rx
            .recv()
            .await
            .expect("expected a DurableSnapshot event on backend open failure");
        match event {
            zeph_tui::AgentEvent::DurableSnapshot(snapshot) => {
                assert_eq!(snapshot.status, DurableStatus::Unavailable);
                assert!(snapshot.executions.is_empty());
                assert_eq!(snapshot.key_id, 9);
                assert_eq!(snapshot.previous_key_id, Some(8));
            }
            other => panic!("expected DurableSnapshot event, got {other:?}"),
        }
    }
}

#[cfg(feature = "tui")]
pub(crate) async fn forward_index_progress_to_tui(
    mut rx: tokio::sync::watch::Receiver<zeph_index::IndexProgress>,
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
) {
    let mut indexing_completed = false;
    while rx.changed().await.is_ok() {
        let p = rx.borrow_and_update().clone();
        if p.files_total == 0 {
            continue;
        }
        let msg = if p.files_done >= p.files_total {
            indexing_completed = true;
            format!(
                "Index ready ({} files, {} chunks)",
                p.files_total, p.chunks_created
            )
        } else {
            let pct = p.files_done * 100 / p.files_total;
            format!(
                "Indexing codebase... {}/{} files ({}%)",
                p.files_done, p.files_total, pct
            )
        };
        if tx.send(zeph_tui::AgentEvent::Status(msg)).await.is_err() {
            break;
        }
    }
    // Keep the final message visible briefly so the user can read it, then clear.
    // Use a shorter delay when indexing finished normally vs. when the sender was
    // dropped unexpectedly (e.g. error path) so the status bar does not stall.
    let delay = if indexing_completed {
        Duration::from_secs(1)
    } else {
        Duration::from_millis(200)
    };
    tokio::time::sleep(delay).await;
    let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;
}

/// Periodically poll the database for fleet session data and forward snapshots to the TUI.
///
/// Runs until the `agent_tx` channel closes (agent exited). The interval is controlled by
/// [`zeph_config::FleetConfig::refresh_interval_secs`].
#[cfg(feature = "tui")]
pub(crate) async fn fleet_poll_task(
    db_path: String,
    cfg: zeph_config::FleetConfig,
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
) {
    let interval_secs = cfg.refresh_interval_secs.max(1);
    let limit = cfg.max_sessions;

    let store = match zeph_memory::store::SqliteStore::new(&db_path).await {
        Ok(s) => s,
        Err(e) => {
            tracing::warn!(error = %e, "fleet poll: failed to open DB; fleet panel will be empty");
            return;
        }
    };

    let mut ticker = tokio::time::interval(Duration::from_secs(interval_secs));
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

    loop {
        ticker.tick().await;

        if tx
            .send(zeph_tui::AgentEvent::Status(
                "Refreshing fleet...".to_owned(),
            ))
            .await
            .is_err()
        {
            break;
        }

        let sessions = match store.list_agent_sessions(limit, None).await {
            Ok(s) => s,
            Err(e) => {
                tracing::debug!(error = %e, "fleet poll: list_agent_sessions failed");
                let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;
                continue;
            }
        };

        let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;

        let snapshot = zeph_tui::widgets::fleet::FleetSnapshot { sessions };
        if tx
            .send(zeph_tui::AgentEvent::FleetSnapshot(snapshot))
            .await
            .is_err()
        {
            break;
        }
    }
}

/// Refresh interval for the TUI durable executions panel, in seconds.
#[cfg(feature = "tui")]
const DURABLE_REFRESH_SECS: u64 = 5;

/// Periodically poll the durable journal and forward snapshots to the TUI (spec-064, #4949).
///
/// Read-only: it never mutates the journal. Runs until the `tx` channel closes (agent exited). Spawned
/// only when `[durable] enabled = true`; otherwise the panel keeps its default state and renders the
/// "non-durable mode" message.
///
/// First evaluates the INV-8 `encryption_gate` security policy (same as the `zeph durable` CLI's
/// `open_backend`, `src/commands/durable.rs`): a declared/detected shared database combined with
/// `encrypt_payload = false` must be rejected here too, or the TUI panel would happily render a
/// journal the CLI refuses to open — a cross-mode divergence (#5996). Unlike the CLI, this is a
/// best-effort background poller with no user waiting on an error message, so a gate rejection
/// degrades gracefully to a `GateRejected` snapshot (distinct from a failed backend open, #6041)
/// rather than panicking or looping forever.
#[cfg(feature = "tui")]
pub(crate) async fn durable_poll_task(
    db_url: String,
    cfg: zeph_config::DurableConfig,
    tx: tokio::sync::mpsc::Sender<zeph_tui::AgentEvent>,
) {
    use zeph_tui::widgets::durable::{DurableRow, DurableSnapshot, DurableStatus};

    if let Err(e) = crate::commands::durable::enforce_encryption_gate(&cfg, &db_url) {
        tracing::warn!(
            error = %e,
            "durable poll: encryption policy rejected this deployment; panel shows gate-rejected status"
        );
        let _ = tx
            .send(zeph_tui::AgentEvent::DurableSnapshot(DurableSnapshot {
                status: DurableStatus::GateRejected,
                executions: Vec::new(),
                key_id: cfg.key_id,
                previous_key_id: cfg.previous_key_id,
            }))
            .await;
        return;
    }

    let backend = match zeph_durable::LocalBackend::open(&db_url, cfg.max_payload_bytes).await {
        Ok(b) => b,
        Err(e) => {
            tracing::warn!(error = %e, "durable poll: failed to open journal; panel shows non-durable mode");
            let _ = tx
                .send(zeph_tui::AgentEvent::DurableSnapshot(DurableSnapshot {
                    status: DurableStatus::Unavailable,
                    executions: Vec::new(),
                    key_id: cfg.key_id,
                    previous_key_id: cfg.previous_key_id,
                }))
                .await;
            return;
        }
    };
    if let Err(e) = backend.init().await {
        tracing::warn!(error = %e, "durable poll: schema init failed");
    }

    let mut ticker = tokio::time::interval(Duration::from_secs(DURABLE_REFRESH_SECS));
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

    loop {
        ticker.tick().await;

        if tx
            .send(zeph_tui::AgentEvent::Status(
                "Refreshing durable journal...".to_owned(),
            ))
            .await
            .is_err()
        {
            break;
        }

        let rows = match backend.list_executions(None, None, 200).await {
            Ok(r) => r,
            Err(e) => {
                tracing::debug!(error = %e, "durable poll: list_executions failed");
                let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;
                continue;
            }
        };

        let _ = tx.send(zeph_tui::AgentEvent::Status(String::new())).await;

        let now_ms = chrono::Utc::now().timestamp_millis();
        let executions: Vec<DurableRow> = rows
            .into_iter()
            .map(|r| DurableRow {
                id_short: r
                    .execution_id
                    .as_uuid()
                    .to_string()
                    .chars()
                    .take(8)
                    .collect(),
                kind: r.kind,
                status: r.status.as_str().to_owned(),
                step_count: r.step_count,
                age_secs: ((now_ms - r.created_at_ms).max(0) / 1000).cast_unsigned(),
            })
            .collect();

        let snapshot = DurableSnapshot {
            status: DurableStatus::Available,
            executions,
            key_id: cfg.key_id,
            previous_key_id: cfg.previous_key_id,
        };
        if tx
            .send(zeph_tui::AgentEvent::DurableSnapshot(snapshot))
            .await
            .is_err()
        {
            break;
        }
    }
}