#![cfg_attr(test, allow(clippy::expect_used, clippy::unwrap_used))]

mod cli;
mod rpc;

use anyhow::{Context, Result};
use capo_agent::{AppBuilder, UiEvent, UserMessage};
use clap::Parser;
use cli::{Cli, Mode};
use futures::StreamExt;
use tracing_subscriber::EnvFilter;

fn main() -> Result<()> {
    color_eyre::install().ok();

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .context("failed to start tokio runtime")?;
    rt.block_on(async_main())
}

fn resolve_permissions_path(
    agent_dir: &std::path::Path,
    cwd: &std::path::Path,
) -> std::path::PathBuf {
    let canonical = agent_dir.join("permissions.toml");
    if canonical.exists() {
        return canonical;
    }

    // Legacy locations from earlier milestones.
    let home_config = std::env::var("HOME")
        .ok()
        .map(|h| std::path::PathBuf::from(h).join(".config/capo/permissions.toml"));
    let project_legacy = cwd.join(".capo").join("permissions.toml");

    for legacy in [home_config.as_ref(), Some(&project_legacy)]
        .into_iter()
        .flatten()
    {
        if legacy.exists() {
            tracing::warn!(
                "permissions.toml found at {}; please mv to {}",
                legacy.display(),
                canonical.display()
            );
            return legacy.clone();
        }
    }

    // No file anywhere — return canonical; AppBuilder treats absent as "default empty policy".
    canonical
}

fn resolve_initial_permission_mode(
    cli: &Cli,
    settings: &capo_agent::Settings,
) -> capo_agent::PermissionMode {
    if let Some(mode) = cli.mode {
        mode
    } else if cli.dangerously_allow_all {
        capo_agent::PermissionMode::Bypass
    } else {
        settings
            .permissions
            .default_mode
            .parse::<capo_agent::PermissionMode>()
            .unwrap_or_else(|err| {
                tracing::warn!(
                    target: "permissions",
                    "settings.toml permissions.default_mode invalid: {err}; falling back to prompt"
                );
                capo_agent::PermissionMode::default()
            })
    }
}

async fn async_main() -> Result<()> {
    let cli = Cli::parse();
    let agent_dir = capo_agent::agent_dir();
    let _tracing_guard = init_tracing(cli.verbosity, &agent_dir);

    let cwd = std::env::current_dir().context("cannot read cwd")?;

    let overrides = capo_agent::CliOverrides {
        model: cli.model.clone(),
    };
    let mut settings = capo_agent::Settings::load(&overrides).context("failed to load settings")?;
    if let Some(provider) = &cli.provider {
        settings.model.provider = provider.clone();
    }
    let initial_permission_mode = resolve_initial_permission_mode(&cli, &settings);
    let auth = capo_agent::Auth::load(&agent_dir).context("failed to load auth.json")?;

    let skills = if cli.no_skills {
        Vec::new()
    } else {
        let result = capo_agent::skills::load_all(&cwd, &capo_agent::paths::agent_dir());
        for d in &result.diagnostics {
            tracing::warn!(target: "skills", path = %d.file_path.display(), "{}", d.message);
        }
        eprintln!("  skills: {} loaded", result.skills.len());
        result.skills
    };

    // Permissions config: try new location first, warn on legacy paths.
    let permissions_path = resolve_permissions_path(&agent_dir, &cwd);

    // M3 Phase A: per-cwd session bucket.
    let session_paths = capo_agent::SessionPaths::for_cwd(agent_dir.clone(), &cwd);

    let resume_session_id: Option<capo_agent::SessionId> = if let Some(s) = &cli.session {
        Some(
            capo_agent::SessionLookup::resolve(&session_paths, s)
                .await
                .context("--session resolution failed")?,
        )
    } else if cli.continue_session {
        Some(
            capo_agent::SessionLookup::most_recent(&session_paths)
                .await
                .context("--continue: no prior session in this directory")?,
        )
    } else {
        None
    };

    eprintln!(
        "  provider: {} | model: {}",
        settings.model.provider, settings.model.name
    );
    if let Some(id) = &resume_session_id {
        eprintln!("  resuming session: {}", id.as_str());
    }

    let (mcp_cfg, mcp_diags) =
        match capo_agent::mcp::load_config(&cwd, &capo_agent::paths::agent_dir(), &|k| {
            std::env::var(k).ok()
        }) {
            Ok(out) => out,
            Err(err) => {
                eprintln!("  mcp: config load failed: {err}");
                (capo_agent::mcp::McpConfig::default(), Vec::new())
            }
        };
    for d in &mcp_diags {
        eprintln!("  mcp: {d}");
    }

    let started = capo_agent::mcp::connect_all(&mcp_cfg, &capo_agent::paths::agent_dir()).await;
    eprintln!("  mcp: {} server(s) connected", started.len());

    let mcp_tools: Vec<std::sync::Arc<dyn motosan_agent_tool::Tool>> = {
        let mut acc = Vec::new();
        for s in &started {
            match motosan_agent_loop::mcp::McpToolAdapter::from_server(std::sync::Arc::clone(
                &s.server,
            ))
            .await
            {
                Ok(mut tools) => acc.append(&mut tools),
                Err(e) => {
                    tracing::error!(target: "mcp", server = %s.name, "list_tools failed: {e}")
                }
            }
        }
        acc
    };

    let mcp_pairs = capo_agent::mcp::into_pairs(started);

    // Best-effort: motosan FileSessionStore creates the dir on first write,
    // but pre-creating avoids a confusing error when listing an empty bucket.
    session_paths.ensure_bucket().ok();
    let store: std::sync::Arc<dyn motosan_agent_loop::SessionStore> = std::sync::Arc::new(
        motosan_agent_loop::FileSessionStore::new(session_paths.bucket_dir.clone()),
    );

    match cli.mode() {
        Mode::Print(msg) => {
            let mut builder = AppBuilder::new()
                .with_settings(settings)
                .with_auth(auth)
                .with_cwd(&cwd)
                .with_builtin_tools()
                .with_permissions_config(permissions_path)
                .with_session_store(store)
                .with_autocompact()
                .with_skills(skills.clone());
            if cli.no_context_files {
                builder = builder.disable_context_discovery();
            }
            builder = builder
                .with_headless_permissions()
                .with_extra_tools(mcp_tools.clone())
                .with_mcp_servers(mcp_pairs.clone());
            let app = match builder.build_with_session(resume_session_id).await {
                Ok(app) => app,
                Err(err) => {
                    disconnect_mcp_pairs(&mcp_pairs).await;
                    return Err(err).context("failed to build App");
                }
            };
            app.set_permission_mode(initial_permission_mode).await;
            app.emit_settings_snapshot().await;
            let result = tokio::select! {
                res = run_print_mode(&app, msg) => res,
                _ = tokio::signal::ctrl_c() => {
                    eprintln!("\n  shutting down (ctrl-C)…");
                    Err(anyhow::anyhow!("interrupted"))
                }
            };
            app.disconnect_mcp().await;
            result
        }
        Mode::Json(msg) => {
            let mut builder = AppBuilder::new()
                .with_settings(settings)
                .with_auth(auth)
                .with_cwd(&cwd)
                .with_builtin_tools()
                .with_permissions_config(permissions_path)
                .with_session_store(store)
                .with_autocompact()
                .with_skills(skills.clone());
            if cli.no_context_files {
                builder = builder.disable_context_discovery();
            }
            builder = builder
                .with_headless_permissions()
                .with_extra_tools(mcp_tools.clone())
                .with_mcp_servers(mcp_pairs.clone());
            let app = match builder.build_with_session(resume_session_id).await {
                Ok(app) => app,
                Err(err) => {
                    disconnect_mcp_pairs(&mcp_pairs).await;
                    return Err(err).context("failed to build App");
                }
            };
            app.set_permission_mode(initial_permission_mode).await;
            app.emit_settings_snapshot().await;
            let mut stdout = std::io::stdout().lock();
            let result = run_json_mode(&app, msg, &cwd, &mut stdout).await;
            app.disconnect_mcp().await;
            result
        }
        Mode::Rpc => {
            // Channels first — App and writer must share one ui_tx so the
            // PermissionExtension's PermissionRequested events reach the
            // bridge (see run_rpc_mode's channel-ownership constraint).
            let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<capo_agent::UiEvent>(256);
            let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<capo_agent::Command>(64);

            let mut builder = AppBuilder::new()
                .with_settings(settings)
                .with_auth(auth)
                .with_cwd(&cwd)
                .with_builtin_tools()
                .with_permissions_config(permissions_path)
                .with_ui_channel(ui_tx.clone())
                .with_session_store(store)
                .with_autocompact()
                .with_skills(skills.clone());
            if cli.no_context_files {
                builder = builder.disable_context_discovery();
            }
            builder = builder
                .with_extra_tools(mcp_tools.clone())
                .with_mcp_servers(mcp_pairs.clone());
            let app = match builder.build_with_session(resume_session_id).await {
                Ok(app) => app,
                Err(err) => {
                    disconnect_mcp_pairs(&mcp_pairs).await;
                    return Err(err).context("failed to build App");
                }
            };
            app.set_permission_mode(initial_permission_mode).await;
            app.emit_settings_snapshot().await;
            let app = std::sync::Arc::new(app);
            let result = run_rpc_mode(
                std::sync::Arc::clone(&app),
                cwd.clone(),
                tokio::io::stdin(),
                tokio::io::stdout(),
                ui_tx,
                ui_rx,
                cmd_tx,
                cmd_rx,
            )
            .await;
            app.disconnect_mcp().await;
            result
        }
        Mode::InteractiveStub => {
            run_interactive(
                settings,
                auth,
                cwd,
                permissions_path,
                store,
                resume_session_id,
                skills,
                mcp_tools,
                mcp_pairs,
                initial_permission_mode,
                &cli,
            )
            .await
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn run_interactive(
    settings: capo_agent::Settings,
    auth: capo_agent::Auth,
    cwd: std::path::PathBuf,
    permissions_path: std::path::PathBuf,
    store: std::sync::Arc<dyn motosan_agent_loop::SessionStore>,
    resume_session_id: Option<capo_agent::SessionId>,
    skills: Vec<capo_agent::Skill>,
    mcp_tools: Vec<std::sync::Arc<dyn motosan_agent_tool::Tool>>,
    mcp_pairs: Vec<(
        String,
        std::sync::Arc<dyn motosan_agent_loop::mcp::McpServer>,
    )>,
    initial_permission_mode: capo_agent::PermissionMode,
    cli: &Cli,
) -> Result<()> {
    use tokio::sync::mpsc;

    let (ui_tx, ui_rx) = mpsc::channel::<capo_agent::UiEvent>(256);
    let (cmd_tx, cmd_rx) = mpsc::channel::<capo_agent::Command>(64);

    let mut builder = AppBuilder::new()
        .with_settings(settings.clone())
        .with_auth(auth)
        .with_cwd(&cwd)
        .with_builtin_tools()
        .with_permissions_config(permissions_path)
        .with_ui_channel(ui_tx.clone())
        .with_session_store(std::sync::Arc::clone(&store))
        .with_autocompact()
        .with_skills(skills.clone());
    if cli.no_context_files {
        builder = builder.disable_context_discovery();
    }
    builder = builder
        .with_extra_tools(mcp_tools.clone())
        .with_mcp_servers(mcp_pairs.clone());
    let was_resumed = resume_session_id.is_some();
    let app = match builder.build_with_session(resume_session_id).await {
        Ok(app) => app,
        Err(err) => {
            disconnect_mcp_pairs(&mcp_pairs).await;
            return Err(err).context("failed to build App");
        }
    };
    app.set_permission_mode(initial_permission_mode).await;
    app.emit_settings_snapshot().await;

    // Seed the TUI transcript with persisted history when resuming. The
    // agent already has full context (loaded by `AgentSession::resume`
    // and observed inside `send_user_message` via `session.history()`),
    // but the TUI's `state.messages` starts empty — so users on
    // `--continue` / `--session` would see a blank screen and have to
    // ask "what did I just say" to verify context loaded. Replaying
    // User / Assistant text entries here closes that UX gap.
    //
    // System messages are skipped (prompt, not transcript). Tool-call /
    // tool-result replay is deferred — would need bridging
    // `motosan::Message.content_blocks` → capo-tui's
    // `MessageBlock::ToolCall { id, name, args }` shape.
    let history = match app.session_history().await {
        Ok(history) => history,
        Err(err) => {
            app.disconnect_mcp().await;
            return Err(err).context("failed to load session history for transcript replay");
        }
    };

    let mut state = capo_tui::state::AppState {
        current_session_id: Some(app.session_id()),
        ..capo_tui::state::AppState::default()
    };
    state.footer.cwd = cwd.display().to_string();
    state.footer.git_branch = read_git_branch(&cwd);
    state.footer.model = settings.model.name.clone();
    state.footer.skills_loaded = skills.len();
    state.footer.show_cost = settings.ui.footer_show_cost;
    state.settings_snapshot = Some(settings.clone());
    state.file_index = capo_tui::file_index::FileIndex::build(&cwd);
    // Session list for the /resume picker. `list_meta` is the same call
    // the old print-and-exit `--resume` path used; map each SessionMeta
    // to a display row.
    if let Ok(metas) = store.list_meta().await {
        state.sessions = metas
            .into_iter()
            .map(|m| {
                let short = m.session_id.chars().take(10).collect::<String>();
                let name = m.name.unwrap_or_else(|| "(unnamed)".to_string());
                capo_tui::state::SessionSummary {
                    label: format!("{short}  {}  {name}", m.updated_at_ms),
                    id: m.session_id,
                }
            })
            .collect();
    }
    // fork_candidates + branches are now lazy-loaded — see the
    // `tokio::spawn` below the TUI start that runs `emit_fork_candidates`
    // and `emit_branch_tree` in the background. Previously these blocked
    // startup (~10-30 ms total walking the current session's entries),
    // gating the first paint behind work the user only sees when they
    // open `/fork` or `/tree`. The pickers now open instantly with their
    // last-known values (empty on a fresh launch); the background task
    // populates them within tens of milliseconds.
    if cli.resume {
        state.bottom_pane = Some(capo_tui::state::BottomPane::ResumePicker(
            capo_tui::state::ResumePickerState::default(),
        ));
    }
    for msg in &history {
        let text = msg.text();
        if text.is_empty() {
            continue;
        }
        match msg.role() {
            motosan_agent_loop::Role::User => {
                state.messages.push(capo_tui::state::MessageBlock::User {
                    text,
                    attachment_filenames: Vec::new(),
                })
            }
            motosan_agent_loop::Role::Assistant => state
                .messages
                .push(capo_tui::state::MessageBlock::Assistant { text }),
            _ => {}
        }
    }

    // Reflow-safe session opening card. The eprintln!s above print to
    // stderr ABOVE the inline viewport and get wiped on the first
    // reflow (known issue tracked in MEMORY.md); this Notice survives
    // reflow because it lives in state.messages and goes through the
    // standard block_to_lines replay path.
    {
        let mut body_lines: Vec<String> = Vec::new();
        body_lines.push(format!(
            "  provider: {} · model: {}",
            settings.model.provider, settings.model.name,
        ));
        let branch_suffix = state
            .footer
            .git_branch
            .as_ref()
            .map(|b| format!(" ({b})"))
            .unwrap_or_default();
        body_lines.push(format!("  cwd: {}{}", cwd.display(), branch_suffix));
        let mut stats: Vec<String> = Vec::new();
        stats.push(format!("skills: {}", skills.len()));
        if !mcp_pairs.is_empty() {
            stats.push(format!("mcp: {} server(s)", mcp_pairs.len()));
        }
        body_lines.push(format!("  {}", stats.join(" · ")));
        // Detect "this is a resume" via `was_resumed` captured before
        // `resume_session_id` was moved into AppBuilder. Using
        // `state.current_session_id.is_some()` won't work — that's
        // always set (to the newly minted id for a fresh session).
        if was_resumed {
            let id_str = state
                .current_session_id
                .as_ref()
                .map(|s| s.as_str().chars().take(10).collect::<String>())
                .unwrap_or_default();
            body_lines.push(format!("  resuming: {id_str}"));
        }
        state.messages.push(capo_tui::state::MessageBlock::Notice {
            title: "capo · session started".into(),
            body: body_lines.join("\n"),
        });
    }

    let app = std::sync::Arc::new(app);
    tokio::spawn(forward_commands(
        std::sync::Arc::clone(&app),
        ui_tx.clone(),
        cmd_rx,
    ));
    // Populate extension-command awareness before the user can interact
    // with the slash palette.
    let _ = cmd_tx.send(capo_agent::Command::ListExtensions).await;

    // Lazy-load fork_candidates + branch_tree in the background so the
    // TUI footer paints immediately instead of waiting for these to walk
    // the current session's entries. The pickers open with their
    // last-known values; this task populates them within tens of ms.
    // (Per-event refresh during turns still happens via
    // `emit_fork_candidates`/`emit_branch_tree` in the event handlers
    // below — this just covers the initial-state case.)
    tokio::spawn({
        let app = std::sync::Arc::clone(&app);
        let ui_tx = ui_tx.clone();
        async move {
            emit_fork_candidates(&app, &ui_tx).await;
            emit_branch_tree(&app, &ui_tx).await;
        }
    });

    let ui_stream = Box::pin(futures::stream::unfold(ui_rx, |mut rx| async move {
        rx.recv().await.map(|event| (event, rx))
    }));
    let result = tokio::select! {
        res = capo_tui::run_tui(state, ui_stream, cmd_tx) => {
            res.map(|_| ()).map_err(|e| anyhow::anyhow!("TUI error: {e}"))
        }
        _ = tokio::signal::ctrl_c() => {
            eprintln!("\n  shutting down (ctrl-C)…");
            Err(anyhow::anyhow!("interrupted"))
        }
    };
    app.disconnect_mcp().await;
    result
}

async fn disconnect_mcp_pairs(
    servers: &[(
        String,
        std::sync::Arc<dyn motosan_agent_loop::mcp::McpServer>,
    )],
) {
    for (name, server) in servers {
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), server.disconnect()).await;
        tracing::debug!(target: "mcp", server = %name, "disconnected after build failure");
    }
}

/// Recompute the `/fork` candidate list and push it to the TUI.
async fn emit_fork_candidates(
    app: &std::sync::Arc<capo_agent::App>,
    ui_tx: &tokio::sync::mpsc::Sender<capo_agent::UiEvent>,
) {
    if let Ok(candidates) = app.fork_candidates().await {
        let _ = ui_tx
            .send(capo_agent::UiEvent::ForkCandidates(candidates))
            .await;
    }
}

/// Recompute the session branch tree and push it to the TUI.
async fn emit_branch_tree(
    app: &std::sync::Arc<capo_agent::App>,
    ui_tx: &tokio::sync::mpsc::Sender<capo_agent::UiEvent>,
) {
    if let Ok(tree) = app.branches().await {
        let _ = ui_tx.send(capo_agent::UiEvent::BranchTree(tree)).await;
    }
}

fn settings_toml_path(agent_dir: &std::path::Path) -> std::path::PathBuf {
    agent_dir.join("settings.toml")
}

/// Best-effort current git branch for `start`'s cwd, read once at startup.
/// Walks up from `cwd` to find a `.git` directory, reads `.git/HEAD`, and
/// returns the branch name from a `ref: refs/heads/<branch>` line. Returns
/// `None` for detached HEAD, missing repo, or any IO error (never panics).
fn read_git_branch(cwd: &std::path::Path) -> Option<String> {
    let mut dir = cwd;
    loop {
        let head = dir.join(".git").join("HEAD");
        if let Ok(contents) = std::fs::read_to_string(&head) {
            let line = contents.trim();
            return line.strip_prefix("ref: refs/heads/").map(|b| b.to_string());
        }
        dir = dir.parent()?;
    }
}

fn load_settings_toml(agent_dir: &std::path::Path) -> anyhow::Result<capo_agent::Settings> {
    capo_agent::Settings::load_with(agent_dir, &capo_agent::CliOverrides::default(), |name| {
        std::env::var(name).ok()
    })
    .map_err(anyhow::Error::from)
}

fn persist_settings_toml(
    agent_dir: &std::path::Path,
    settings: &capo_agent::Settings,
) -> anyhow::Result<()> {
    std::fs::create_dir_all(agent_dir)
        .with_context(|| format!("failed to create {}", agent_dir.display()))?;
    let toml_string = toml::to_string_pretty(settings).context("failed to serialize settings")?;
    let final_path = settings_toml_path(agent_dir);
    let tmp_path = agent_dir.join("settings.toml.tmp");
    std::fs::write(&tmp_path, toml_string)
        .with_context(|| format!("failed to write {}", tmp_path.display()))?;
    std::fs::rename(&tmp_path, &final_path).with_context(|| {
        format!(
            "failed to rename {} to {}",
            tmp_path.display(),
            final_path.display()
        )
    })?;
    Ok(())
}

async fn forward_commands(
    app: std::sync::Arc<capo_agent::App>,
    ui_tx: tokio::sync::mpsc::Sender<capo_agent::UiEvent>,
    mut cmd_rx: tokio::sync::mpsc::Receiver<capo_agent::Command>,
) {
    use capo_agent::Command;

    let mut active_turn: Option<tokio::task::JoinHandle<()>> = None;
    while let Some(cmd) = cmd_rx.recv().await {
        if active_turn
            .as_ref()
            .map(|handle| handle.is_finished())
            .unwrap_or(false)
        {
            active_turn = None;
        }

        match cmd {
            Command::SendUserMessage { text, attachments } => {
                if active_turn.is_some() {
                    tracing::debug!("turn already running; ignoring extra SendUserMessage");
                    continue;
                }

                let app = std::sync::Arc::clone(&app);
                let ui_tx = ui_tx.clone();
                active_turn = Some(tokio::spawn(async move {
                    use capo_agent::extensions::dispatcher::{
                        dispatch_before_user_message, BeforeUserMessageOutcome,
                    };
                    // Fire before_user_message hook chain. If cancelled,
                    // emit ExtensionCancelled and skip the LLM turn AND
                    // the post-stream emits (the turn never started, so
                    // there's nothing to refresh).
                    let registry = app.extension_registry();
                    let final_msg =
                        match dispatch_before_user_message(&registry, text, attachments).await {
                            BeforeUserMessageOutcome::Proceed { text, attachments } => {
                                UserMessage { text, attachments }
                            }
                            BeforeUserMessageOutcome::Cancelled {
                                extension_name,
                                reason,
                            } => {
                                let _ = ui_tx
                                    .send(capo_agent::UiEvent::ExtensionCancelled {
                                        extension_name,
                                        reason,
                                    })
                                    .await;
                                return;
                            }
                        };

                    let mut stream = Box::pin(app.send_user_message(final_msg));
                    while let Some(ev) = stream.next().await {
                        if ui_tx.send(ev).await.is_err() {
                            return;
                        }
                    }
                    // Post-turn refresh (fork candidates + branch tree) is
                    // detached so it does NOT keep `active_turn` alive past the
                    // turn's stream. Otherwise a queued follow-up drained on
                    // `AgentTurnComplete` (the queue's auto-send) races these
                    // emits: the drained `SendUserMessage` arrives while
                    // `active_turn` is still `Some`, hits the guard above, and
                    // is dropped — leaving the TUI stuck on its optimistic
                    // thinking spinner forever with no turn to cancel. With the
                    // emits detached, `active_turn` is finished the moment the
                    // stream ends, so the drained turn is accepted. These
                    // events are advisory and ordering-independent of the next
                    // turn.
                    tokio::spawn(async move {
                        emit_fork_candidates(&app, &ui_tx).await;
                        emit_branch_tree(&app, &ui_tx).await;
                    });
                }));
            }
            Command::CancelAgent => {
                // B: trip the real cancel token. The engine's streaming select
                // observes this and returns AgentError::Cancelled mid-stream.
                app.cancel();
                // A (backstop): if the turn hasn't unwound within the grace
                // window (e.g. blocked in a deferred-slot wait the cancel token
                // doesn't race), hard-abort it and synthesize turn completion so
                // the TUI clears its spinner. Mirrors codex handle_task_abort:
                // cancel -> ~100ms grace -> handle.abort() -> TurnAborted.
                if let Some(mut handle) = active_turn.take() {
                    let ui_tx = ui_tx.clone();
                    tokio::spawn(async move {
                        tokio::select! {
                            _ = &mut handle => {
                                // Unwound cleanly; its stream already emitted
                                // AgentTurnComplete via the Cancelled translation.
                            }
                            _ = tokio::time::sleep(std::time::Duration::from_millis(150)) => {
                                if !handle.is_finished() {
                                    handle.abort();
                                    let _ = ui_tx
                                        .send(capo_agent::UiEvent::AgentTurnComplete)
                                        .await;
                                }
                            }
                        }
                    });
                }
            }
            Command::Quit => {
                if let Some(handle) = active_turn.take() {
                    handle.abort();
                }
                return;
            }
            Command::ResolvePermission(resolution) => {
                use capo_agent::PermissionChoice;

                if resolution.choice == PermissionChoice::AllowSession {
                    let key = capo_agent::permissions::SessionCache::key(
                        &resolution.tool,
                        &resolution.args,
                    );
                    app.permissions_cache()
                        .insert(key, capo_agent::Decision::Allowed);
                }
                // AllowOnce and Deny are already communicated via the
                // oneshot channel inside the modal; no cache write needed.
            }
            Command::SetPermissionMode { mode } => {
                app.set_permission_mode(mode).await;
                // App::set_permission_mode itself emits the
                // UiEvent::PermissionModeChanged; the TUI's handler will
                // update state.permission_mode when it arrives via the
                // existing event stream.
            }
            Command::RunInlineBash {
                command,
                send_to_llm,
            } => {
                if send_to_llm && active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "cannot run !cmd while an agent turn is running".into(),
                        ))
                        .await;
                    continue;
                }

                let output = run_inline_bash(&command).await;

                // Surface "$ command → output" in the user's transcript in
                // BOTH paths. `!!cmd` is the transcript-only branch; `!cmd`
                // ALSO feeds the LLM, but the user still needs to see the
                // shell side independently of the composed user-message the
                // agent receives (otherwise it looks like nothing happened —
                // the LLM just starts thinking out of nowhere). Previously
                // only the `!!` branch emitted this, so `!cmd` produced no
                // visible artifact for the user despite the bash having run.
                let _ = ui_tx
                    .send(capo_agent::UiEvent::InlineBashOutput {
                        command: command.clone(),
                        output: output.clone(),
                    })
                    .await;

                if send_to_llm {
                    // `!cmd` — feed the command + output to the agent as a
                    // user message. Reuses the SendUserMessage path so the
                    // single-turn guard / active_turn bookkeeping applies.
                    // Dynamic fence so output containing triple-backticks (e.g.
                    // `!cat` on a Markdown file) doesn't break the outer fence
                    // and confuse the LLM. See [issue #44].
                    let fence = fence_for(&output);
                    let composed = format!(
                        "I ran `{command}` in the shell. Output:\n\n{fence}\n{output}\n{fence}"
                    );
                    let app = std::sync::Arc::clone(&app);
                    let ui_tx = ui_tx.clone();
                    active_turn = Some(tokio::spawn(async move {
                        let mut stream =
                            Box::pin(app.send_user_message(UserMessage::text(composed)));
                        while let Some(ev) = stream.next().await {
                            if ui_tx.send(ev).await.is_err() {
                                break;
                            }
                        }
                        emit_fork_candidates(&app, &ui_tx).await;
                        emit_branch_tree(&app, &ui_tx).await;
                    }));
                }
                // `!!cmd` (send_to_llm=false): the InlineBashOutput emit
                // above is the only side effect. No LLM dispatch.
            }
            Command::Compact => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "cannot compact while an agent turn is running".into(),
                        ))
                        .await;
                    continue;
                }
                match app.compact().await {
                    Err(err) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Error(err.to_string()))
                            .await;
                    }
                    Ok(None) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Notice {
                                title: "/compact".into(),
                                body: "Nothing to compact yet. Need at least one user message with prior history to summarize.".into(),
                            })
                            .await;
                    }
                    Ok(Some(result)) => {
                        // Estimate summary token count -- CompactionResult exposes
                        // only the summary string. Use a 4-chars-per-token heuristic
                        // matching motosan's own approx_visible_chars/4.
                        let summary_tokens = result.summary.len() / 4;
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Compacted {
                                turns_removed: 0,
                                summary_tokens,
                                source: capo_agent::CompactSource::Manual,
                            })
                            .await;
                        emit_fork_candidates(&app, &ui_tx).await;
                        emit_branch_tree(&app, &ui_tx).await;
                    }
                }
            }
            Command::ApplySettings { settings } => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "finish or cancel the current turn before saving settings".into(),
                        ))
                        .await;
                    continue;
                }

                let agent_dir = capo_agent::agent_dir();

                if let Err(err) = app.reload_settings(settings.clone()).await {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(format!(
                            "settings: rebuild failed: {err}"
                        )))
                        .await;
                    continue;
                }

                emit_fork_candidates(&app, &ui_tx).await;
                emit_branch_tree(&app, &ui_tx).await;
                let _ = ui_tx
                    .send(capo_agent::UiEvent::SettingsSnapshot {
                        settings: settings.clone(),
                    })
                    .await;

                if let Err(err) = persist_settings_toml(&agent_dir, &settings) {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(format!(
                            "settings: applied to live session but failed to persist to disk: {err}"
                        )))
                        .await;
                    continue;
                }

                let _ = ui_tx
                    .send(capo_agent::UiEvent::Notice {
                        title: "/settings".into(),
                        body: "settings applied and saved".into(),
                    })
                    .await;
            }
            Command::ReloadSettings => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "finish or cancel the current turn before reloading settings".into(),
                        ))
                        .await;
                    continue;
                }

                let agent_dir = capo_agent::agent_dir();
                let new_settings = match load_settings_toml(&agent_dir) {
                    Ok(settings) => settings,
                    Err(err) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Error(format!(
                                "settings: reload failed: {err}"
                            )))
                            .await;
                        continue;
                    }
                };

                let settings_for_event = new_settings.clone();
                if let Err(err) = app.reload_settings(new_settings).await {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(format!(
                            "settings: rebuild failed (settings.toml may diverge from live session): {err}"
                        )))
                        .await;
                    continue;
                }

                emit_fork_candidates(&app, &ui_tx).await;
                emit_branch_tree(&app, &ui_tx).await;

                let _ = ui_tx
                    .send(capo_agent::UiEvent::SettingsSnapshot {
                        settings: settings_for_event,
                    })
                    .await;
                let _ = ui_tx
                    .send(capo_agent::UiEvent::Notice {
                        title: "/settings".into(),
                        body: "settings reloaded from disk".into(),
                    })
                    .await;
            }
            Command::NewSession => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "finish or cancel the current turn before /new".into(),
                        ))
                        .await;
                    continue;
                }
                match app.new_session().await {
                    Ok(()) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::SessionReplaced {
                                session_id: app.session_id(),
                                history: Vec::new(),
                            })
                            .await;
                        emit_fork_candidates(&app, &ui_tx).await;
                        emit_branch_tree(&app, &ui_tx).await;
                    }
                    Err(err) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Error(format!("/new failed: {err}")))
                            .await;
                    }
                }
            }
            Command::GetMessages => {
                let history = app.session_history().await.unwrap_or_default();
                let _ = ui_tx
                    .send(capo_agent::UiEvent::MessagesSnapshot(history))
                    .await;
            }
            Command::GetState => {
                let _ = ui_tx
                    .send(capo_agent::UiEvent::StateSnapshot {
                        session_id: app.session_id(),
                        model: app.settings().model.name,
                        active_turn: active_turn.is_some(),
                    })
                    .await;
            }
            Command::ListExtensions => {
                use capo_agent::events::{ExtensionInfo, UiEvent};
                let registry = app.extension_registry();
                let mut extensions: Vec<ExtensionInfo> = registry
                    .extensions
                    .iter()
                    .map(|reg| {
                        let command = reg.entry.command.as_str();
                        let exists = is_executable_file(std::path::Path::new(command))
                            || which_in_path(command);
                        ExtensionInfo {
                            name: reg.entry.name.clone(),
                            hooks: reg.entry.hooks.clone(),
                            commands: reg.entry.commands.clone(),
                            healthy: exists,
                            diagnostic: if exists {
                                None
                            } else {
                                Some(format!("binary not found: {command}"))
                            },
                        }
                    })
                    .collect();
                for diag in app.extension_diagnostics().iter() {
                    let severity = match diag.severity {
                        capo_agent::extensions::DiagnosticSeverity::Warn => "warn",
                        capo_agent::extensions::DiagnosticSeverity::Error => "error",
                    };
                    let name = diag.extension_name.clone();
                    let message = diag.message.as_str();
                    extensions.push(ExtensionInfo {
                        name,
                        hooks: Vec::new(),
                        commands: Vec::new(),
                        healthy: false,
                        diagnostic: Some(format!("{severity}: {message}")),
                    });
                }
                let _ = ui_tx.send(UiEvent::ExtensionList { extensions }).await;
            }
            Command::InvokeExtensionCommand { name, args } => {
                use capo_agent::extensions::dispatcher::{
                    dispatch_before_user_message, dispatch_command, BeforeUserMessageOutcome,
                    CommandOutcome,
                };

                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(format!(
                            "finish or cancel the current turn before /{name}"
                        )))
                        .await;
                    continue;
                }

                let app = std::sync::Arc::clone(&app);
                let ui_tx = ui_tx.clone();
                active_turn = Some(tokio::spawn(async move {
                    let registry = app.extension_registry();
                    let outcome = dispatch_command(&registry, &name, &args).await;
                    match outcome {
                        CommandOutcome::Unknown => {
                            let _ = ui_tx
                                .send(capo_agent::UiEvent::Error(format!(
                                    "no extension owns slash command `/{name}`"
                                )))
                                .await;
                        }
                        CommandOutcome::NoOp => {
                            // Extension intentionally did nothing.
                            tracing::debug!(
                                target: "extensions",
                                "command /{name} returned no-op"
                            );
                        }
                        CommandOutcome::Reply { text } => {
                            let _ = ui_tx
                                .send(capo_agent::UiEvent::Notice {
                                    title: format!("/{name}"),
                                    body: text,
                                })
                                .await;
                        }
                        CommandOutcome::Send { text, attachments } => {
                            let final_msg =
                                match dispatch_before_user_message(&registry, text, attachments)
                                    .await
                                {
                                    BeforeUserMessageOutcome::Proceed { text, attachments } => {
                                        capo_agent::UserMessage { text, attachments }
                                    }
                                    BeforeUserMessageOutcome::Cancelled {
                                        extension_name,
                                        reason,
                                    } => {
                                        let _ = ui_tx
                                            .send(capo_agent::UiEvent::ExtensionCancelled {
                                                extension_name,
                                                reason,
                                            })
                                            .await;
                                        return;
                                    }
                                };

                            let mut stream = Box::pin(app.send_user_message(final_msg));
                            while let Some(ev) = stream.next().await {
                                if ui_tx.send(ev).await.is_err() {
                                    break;
                                }
                            }
                            emit_fork_candidates(&app, &ui_tx).await;
                            emit_branch_tree(&app, &ui_tx).await;
                        }
                        CommandOutcome::Cancelled {
                            extension_name,
                            reason,
                        } => {
                            let _ = ui_tx
                                .send(capo_agent::UiEvent::ExtensionCancelled {
                                    extension_name,
                                    reason,
                                })
                                .await;
                        }
                    }
                }));
            }
            Command::CloneSession => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "finish or cancel the current turn before /clone".into(),
                        ))
                        .await;
                    continue;
                }
                match app.clone_session().await {
                    Ok(_new_id) => {
                        let history = app.session_history().await.unwrap_or_default();
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::SessionReplaced {
                                session_id: app.session_id(),
                                history,
                            })
                            .await;
                        emit_fork_candidates(&app, &ui_tx).await;
                        emit_branch_tree(&app, &ui_tx).await;
                    }
                    Err(err) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Error(format!("/clone failed: {err}")))
                            .await;
                    }
                }
            }
            Command::LoadSession(id) => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "finish or cancel the current turn before /resume".into(),
                        ))
                        .await;
                    continue;
                }
                match app.load_session(&id).await {
                    Ok(()) => {
                        let history = app.session_history().await.unwrap_or_default();
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::SessionReplaced {
                                session_id: app.session_id(),
                                history,
                            })
                            .await;
                        emit_fork_candidates(&app, &ui_tx).await;
                        emit_branch_tree(&app, &ui_tx).await;
                    }
                    Err(err) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Error(format!("/resume failed: {err}")))
                            .await;
                    }
                }
            }
            Command::SwitchModel(model) => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "finish or cancel the current turn before /model".into(),
                        ))
                        .await;
                    continue;
                }
                match app.switch_model(&model).await {
                    Ok(()) => {
                        let _ = ui_tx.send(capo_agent::UiEvent::ModelSwitched(model)).await;
                    }
                    Err(err) => {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::Error(format!("/model failed: {err}")))
                            .await;
                    }
                }
            }
            Command::ForkFrom { from, message } => {
                if active_turn.is_some() {
                    let _ = ui_tx
                        .send(capo_agent::UiEvent::Error(
                            "finish or cancel the current turn before /fork".into(),
                        ))
                        .await;
                    continue;
                }
                let app = std::sync::Arc::clone(&app);
                let ui_tx = ui_tx.clone();
                active_turn = Some(tokio::spawn(async move {
                    let mut stream = Box::pin(app.fork_from(from, UserMessage::text(message)));
                    let mut saw_turn_started = false;
                    while let Some(ev) = stream.next().await {
                        saw_turn_started |= matches!(ev, capo_agent::UiEvent::AgentTurnStarted);
                        if ui_tx.send(ev).await.is_err() {
                            break;
                        }
                    }
                    if !saw_turn_started {
                        return;
                    }
                    // After the fork turn: rebuild the transcript onto the
                    // new branch, then refresh the fork-candidate list.
                    if let Ok(history) = app.session_history().await {
                        let _ = ui_tx
                            .send(capo_agent::UiEvent::SessionReplaced {
                                session_id: app.session_id(),
                                history,
                            })
                            .await;
                    }
                    emit_fork_candidates(&app, &ui_tx).await;
                    emit_branch_tree(&app, &ui_tx).await;
                }));
            }
        }
    }

    if let Some(handle) = active_turn.take() {
        let _ = handle.await;
    }
}

/// Run a user-typed `!`/`!!` shell command, returning combined
/// stdout+stderr (truncated). Runs via the system shell so pipes and
/// globs work like the user expects.
async fn run_inline_bash(command: &str) -> String {
    run_inline_bash_with_timeout(command, std::time::Duration::from_secs(30)).await
}

/// Pick a backtick fence length that won't collide with any content in
/// `output`. CommonMark §4.5 says a fenced block closes only on a line
/// that starts (optionally indented by ≤ 3 spaces) with at least as many
/// backticks as the opening. So we scan each line, strip up to 3
/// leading spaces, count the leading backtick run, and pick
/// `(longest_run + 1).max(3)` as the outer fence length. Output with
/// no backticks → ``` (3, unchanged). Output with one ```` ``` ```` line
/// → ```` ```` (4). And so on.
///
/// Closes [issue #44](https://github.com/motosan-dev/capo/issues/44):
/// `!cat README.md` on a Markdown file with code fences used to break
/// the outer fence, leaving the LLM with imbalanced markdown.
fn fence_for(output: &str) -> String {
    let longest_run = output
        .lines()
        .map(|line| {
            // CommonMark allows ≤ 3 spaces of leading indent on a closing
            // fence. Strip those before counting backticks.
            let trimmed = line.trim_start_matches(' ');
            trimmed.bytes().take_while(|&b| b == b'`').count()
        })
        .max()
        .unwrap_or(0);
    "`".repeat((longest_run + 1).max(3))
}

async fn run_inline_bash_with_timeout(command: &str, timeout: std::time::Duration) -> String {
    use tokio::process::Command as ProcCommand;
    const MAX: usize = 30 * 1024;

    let mut child = ProcCommand::new("sh");
    child.arg("-c").arg(command).kill_on_drop(true);
    let result = tokio::time::timeout(timeout, child.output()).await;
    match result {
        Ok(Ok(out)) => {
            let mut s = String::from_utf8_lossy(&out.stdout).into_owned();
            let err = String::from_utf8_lossy(&out.stderr);
            if !err.is_empty() {
                s.push_str(&err);
            }
            if s.len() > MAX {
                truncate_utf8(&mut s, MAX);
                s.push_str("\n… (output truncated)");
            }
            if s.is_empty() {
                "(no output)".to_string()
            } else {
                s
            }
        }
        Ok(Err(e)) => format!("failed to run command: {e}"),
        Err(_) => format!("command timed out after {}s", timeout.as_secs()),
    }
}

fn truncate_utf8(s: &mut String, max: usize) {
    let mut end = max.min(s.len());
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    s.truncate(end);
}

async fn run_print_mode(app: &capo_agent::App, msg: capo_agent::UserMessage) -> Result<()> {
    let mut events = vec![
        UiEvent::PermissionModeChanged {
            mode: app.permission_mode().await,
        },
        UiEvent::SettingsSnapshot {
            settings: app.settings(),
        },
    ];
    events.extend(app.send_user_message(msg).collect::<Vec<_>>().await);
    let rendered = render_print_mode_events(&events)?;
    print!("{}", rendered.stdout);
    eprint!("{}", rendered.stderr);
    Ok(())
}

/// `--json` one-shot: emit a session header, then stream the turn's
/// `UiEvent`s as line-delimited JSON. Generic over the writer so tests can
/// capture into a `Vec<u8>`; the binary passes a locked stdout handle.
async fn run_json_mode<W: std::io::Write>(
    app: &capo_agent::App,
    msg: capo_agent::UserMessage,
    cwd: &std::path::Path,
    out: &mut W,
) -> Result<()> {
    let header =
        capo_agent::protocol::SessionHeader::new(app.session_id(), cwd.display().to_string());
    writeln!(
        out,
        "{}",
        serde_json::to_string(&header).unwrap_or_default()
    )?;
    writeln!(
        out,
        "{}",
        capo_agent::protocol::encode_event(&UiEvent::PermissionModeChanged {
            mode: app.permission_mode().await,
        })
    )?;
    writeln!(
        out,
        "{}",
        capo_agent::protocol::encode_event(&UiEvent::SettingsSnapshot {
            settings: app.settings(),
        })
    )?;

    let stream = app.send_user_message(msg);
    futures::pin_mut!(stream);
    let mut error_message: Option<String> = None;
    while let Some(event) = stream.next().await {
        if let UiEvent::Error(message) = &event {
            error_message.get_or_insert_with(|| message.clone());
        }
        writeln!(out, "{}", capo_agent::protocol::encode_event(&event))?;
    }
    out.flush()?;
    if let Some(message) = error_message {
        Err(anyhow::anyhow!("agent error: {message}"))
    } else {
        Ok(())
    }
}

/// `--rpc` server: a persistent JSONL-over-stdio process around the
/// existing `forward_commands` loop. Channels are passed in (not
/// created here) so the App built by the caller shares the writer's
/// `ui_tx` — see the channel-ownership note above. Generic over the
/// I/O streams so tests use in-memory `tokio::io::duplex` pipes.
#[allow(clippy::too_many_arguments)]
async fn run_rpc_mode<R, W>(
    app: std::sync::Arc<capo_agent::App>,
    cwd: std::path::PathBuf,
    input: R,
    mut output: W,
    ui_tx: tokio::sync::mpsc::Sender<capo_agent::UiEvent>,
    ui_rx: tokio::sync::mpsc::Receiver<capo_agent::UiEvent>,
    cmd_tx: tokio::sync::mpsc::Sender<capo_agent::Command>,
    cmd_rx: tokio::sync::mpsc::Receiver<capo_agent::Command>,
) -> Result<()>
where
    R: tokio::io::AsyncRead + Unpin + Send + 'static,
    W: tokio::io::AsyncWrite + Unpin + Send + 'static,
{
    use tokio::io::AsyncWriteExt;

    // Header line — same shape as --json's.
    let header =
        capo_agent::protocol::SessionHeader::new(app.session_id(), cwd.display().to_string());
    let header_line = serde_json::to_string(&header).unwrap_or_default();
    output.write_all(header_line.as_bytes()).await?;
    output.write_all(b"\n").await?;
    output.flush().await?;

    let bridge = std::sync::Arc::new(rpc::PermissionBridge::with_session_cache(
        app.permissions_cache(),
    ));

    let mut forwarder = tokio::spawn(forward_commands(
        std::sync::Arc::clone(&app),
        ui_tx.clone(),
        cmd_rx,
    ));
    let (writer_shutdown_tx, writer_shutdown_rx) = tokio::sync::oneshot::channel();
    let mut writer = tokio::spawn(rpc::writer_task_until_shutdown(
        output,
        ui_rx,
        std::sync::Arc::clone(&bridge),
        writer_shutdown_rx,
    ));
    let mut reader = tokio::spawn(rpc::reader_task(
        tokio::io::BufReader::new(input),
        cmd_tx,
        ui_tx,
        std::sync::Arc::clone(&bridge),
    ));

    tokio::select! {
        reader_join = &mut reader => {
            if let Ok(joined) = reader_join {
                joined?;
            }
            // EOF and explicit quit both route Command::Quit, so the
            // forwarder aborts any active turn (including a permission wait)
            // and exits promptly.
            let _ = forwarder.await;
            let _ = writer_shutdown_tx.send(());
            match writer.await {
                Ok(result) => result?,
                Err(err) if err.is_cancelled() => {}
                Err(err) => return Err(anyhow::anyhow!("rpc writer task failed: {err}")),
            }
            Ok(())
        }
        writer_join = &mut writer => {
            app.cancel();
            forwarder.abort();
            reader.abort();
            match writer_join {
                Ok(result) => result.map_err(anyhow::Error::from),
                Err(err) if err.is_cancelled() => Ok(()),
                Err(err) => Err(anyhow::anyhow!("rpc writer task failed: {err}")),
            }
        }
        forwarder_join = &mut forwarder => {
            let _ = forwarder_join;
            reader.abort();
            let _ = writer_shutdown_tx.send(());
            match writer.await {
                Ok(result) => result?,
                Err(err) if err.is_cancelled() => {}
                Err(err) => return Err(anyhow::anyhow!("rpc writer task failed: {err}")),
            }
            Ok(())
        }
    }
}

#[derive(Debug, Default, PartialEq, Eq)]
struct RenderedPrintMode {
    stdout: String,
    stderr: String,
}

fn render_print_mode_events(events: &[UiEvent]) -> Result<RenderedPrintMode> {
    let mut rendered = RenderedPrintMode::default();
    let mut saw_text_delta = false;

    for event in events {
        match event {
            UiEvent::AgentTurnStarted => {}
            UiEvent::AgentThinking => rendered.stderr.push_str("[thinking]\n"),
            UiEvent::AgentTextDelta(chunk) => {
                saw_text_delta = true;
                rendered.stdout.push_str(chunk);
            }
            UiEvent::AgentThinkingDelta(_chunk) => {
                // Non-interactive renderer: emit one finalized line per
                // ThinkingComplete; per-token deltas are dropped.
            }
            UiEvent::ThinkingComplete { text } => {
                rendered.stderr.push_str(&format!("[thinking] {text}\n"));
            }
            UiEvent::AgentMessageComplete(text) => {
                if !saw_text_delta && !text.is_empty() {
                    rendered.stdout.push_str(text);
                    rendered.stdout.push('\n');
                }
            }
            UiEvent::ToolCallStarted { name, args, .. } => {
                rendered
                    .stderr
                    .push_str(&format!("[tool:start] {name} {args}\n"));
            }
            // Progress chunks are TUI-only for M2; print mode intentionally
            // ignores them to keep stdout/stderr stable and non-stream-fragmented.
            UiEvent::ToolCallProgress { .. } => {}
            UiEvent::ToolCallCompleted { result, .. } => {
                if result.is_error {
                    rendered
                        .stderr
                        .push_str(&format!("[tool:error] {}\n", result.text));
                } else {
                    rendered.stderr.push_str("[tool:done]\n");
                }
            }
            UiEvent::AgentTurnComplete => {
                if saw_text_delta {
                    rendered.stdout.push('\n');
                }
            }
            UiEvent::TurnStats {
                input_tokens,
                output_tokens,
                cumulative_input,
                cumulative_output,
                model,
            } => {
                rendered.stderr.push_str(&format!(
                    "[tokens] +{input_tokens} in / +{output_tokens} out (Σ {cumulative_input} / {cumulative_output} for {model})\n"
                ));
            }
            UiEvent::PermissionRequested { .. } => {
                // Phase A protocol surface only; permission UX lands in Phase E/F.
            }
            UiEvent::PermissionModeChanged { mode } => {
                rendered.stderr.push_str(&format!("[mode: {mode}]\n"));
            }
            UiEvent::InlineBashOutput { .. } => {}
            UiEvent::Notice { title, body } => {
                rendered
                    .stderr
                    .push_str(&format!("[notice] {title}: {body}\n"));
            }
            UiEvent::Compacted {
                turns_removed,
                summary_tokens,
                source,
            } => {
                let prefix = match source {
                    capo_agent::CompactSource::Manual => "Compacted",
                    capo_agent::CompactSource::Auto => "Autocompacted",
                };
                let body = if *turns_removed == 0 {
                    format!("Summarized older history into ~{summary_tokens} tokens.")
                } else {
                    format!("Summarized {turns_removed} turn(s) into ~{summary_tokens} tokens.")
                };
                rendered
                    .stderr
                    .push_str(&format!("[notice] {prefix}: {body}\n"));
            }
            UiEvent::ExtensionFailed { name, error } => {
                rendered
                    .stderr
                    .push_str(&format!("[error] extension '{name}' failed: {error}\n"));
            }
            UiEvent::SessionReplaced { .. }
            | UiEvent::ModelSwitched(_)
            | UiEvent::ForkCandidates(_)
            | UiEvent::BranchTree(_)
            | UiEvent::MessagesSnapshot(_)
            | UiEvent::StateSnapshot { .. } => {}
            UiEvent::SettingsSnapshot { .. } => {
                // Print mode ignores settings snapshots — they're an
                // interactive-only signal. No stderr output.
            }
            UiEvent::ExtensionList { .. } => {
                // Print mode ignores extension introspection — it's an
                // interactive-only query.
            }
            UiEvent::AttachmentError { kind: _, message } => {
                rendered.stderr.push_str(&format!("[error] {message}\n"));
                return Err(anyhow::anyhow!("agent error: {message}"));
            }
            UiEvent::ExtensionCancelled {
                extension_name,
                reason,
            } => {
                let line = match reason {
                    Some(r) => format!("[error] extension `{extension_name}` cancelled: {r}\n"),
                    None => format!("[error] extension `{extension_name}` cancelled the action\n"),
                };
                rendered.stderr.push_str(&line);
                return Err(anyhow::anyhow!(
                    "extension `{extension_name}` cancelled the action"
                ));
            }
            UiEvent::Error(msg) => {
                rendered.stderr.push_str(&format!("[error] {msg}\n"));
                return Err(anyhow::anyhow!("agent error: {msg}"));
            }
        }
    }

    Ok(rendered)
}

/// Returns `true` if `name` looks like a PATH lookup (no `/`) and is
/// resolvable to an executable on the current PATH.
fn which_in_path(name: &str) -> bool {
    if name.contains('/') {
        return false;
    }
    let Ok(path_env) = std::env::var("PATH") else {
        return false;
    };
    for dir in path_env.split(':') {
        let candidate = std::path::Path::new(dir).join(name);
        if is_executable_file(&candidate) {
            return true;
        }
    }
    false
}

fn is_executable_file(path: &std::path::Path) -> bool {
    if !path.is_file() {
        return false;
    }
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        path.metadata()
            .map(|m| m.permissions().mode() & 0o111 != 0)
            .unwrap_or(false)
    }
    #[cfg(not(unix))]
    {
        true
    }
}

/// Wire tracing to ~/.capo/agent/capo.log. Returns the WorkerGuard that
/// must outlive the process so background flushes don't drop late writes.
///
/// Codex precedent: codex-rs/tui/src/lib.rs:1180-1213 uses the same
/// tracing-appender::non_blocking + tracing_subscriber::registry pattern
/// to avoid polluting the terminal in inline-viewport mode. Pre-v0.12,
/// alt-screen swallowed stderr writes; v0.12 has no alt-screen so logs
/// must go to a file.
fn init_tracing(
    verbosity: u8,
    agent_dir: &std::path::Path,
) -> tracing_appender::non_blocking::WorkerGuard {
    use tracing_subscriber::prelude::*;

    let default_level = match verbosity {
        0 => "capo_agent=info,capo_tui=info,capo=info",
        1 => "capo_agent=debug,capo_tui=debug,capo=debug",
        _ => "trace",
    };
    let env_filter =
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(default_level));

    // Create agent_dir if missing (e.g., first run). Best-effort: a chmod
    // error or io error here shouldn't kill startup — fall back to a
    // no-op guard.
    let log_path = agent_dir.join("capo.log");
    let _ = std::fs::create_dir_all(agent_dir);

    let mut log_file_opts = std::fs::OpenOptions::new();
    log_file_opts.create(true).append(true);
    #[cfg(unix)]
    {
        use std::os::unix::fs::OpenOptionsExt;
        log_file_opts.mode(0o600);
    }

    let log_file = match log_file_opts.open(&log_path) {
        Ok(f) => {
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                let _ = f.set_permissions(std::fs::Permissions::from_mode(0o600));
            }
            f
        }
        Err(err) => {
            // Fatal? No — logging failure should not block startup. Print
            // once to stderr (we're still pre-TUI here, so this lands in
            // the user's terminal before raw mode) and return a no-op
            // guard via a sink writer.
            #[allow(clippy::print_stderr)]
            {
                eprintln!(
                    "warning: cannot open log file {}: {err}",
                    log_path.display()
                );
            }
            let (sink, guard) = tracing_appender::non_blocking(std::io::sink());
            let _ = tracing_subscriber::registry()
                .with(
                    tracing_subscriber::fmt::layer()
                        .with_writer(sink)
                        .with_filter(env_filter),
                )
                .try_init();
            return guard;
        }
    };

    let (non_blocking, guard) = tracing_appender::non_blocking(log_file);
    let file_layer = tracing_subscriber::fmt::layer()
        .with_writer(non_blocking)
        .with_target(true)
        .with_ansi(false)
        .with_span_events(
            tracing_subscriber::fmt::format::FmtSpan::NEW
                | tracing_subscriber::fmt::format::FmtSpan::CLOSE,
        )
        .with_filter(env_filter);

    let _ = tracing_subscriber::registry().with(file_layer).try_init();

    guard
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::time::Duration;

    use async_trait::async_trait;
    use capo_agent::{
        AppBuilder, Command, Config, PermissionChoice, PermissionResolution, UiToolResult,
    };
    use motosan_agent_loop::{
        ChatOutput, ContentPart, LlmClient, LlmResponse, Message, SessionStore, ToolCallItem,
    };
    use motosan_agent_tool::ToolDef;

    use super::*;

    #[test]
    fn fence_for_plain_output_uses_three_backticks() {
        // Output with no backticks → conventional 3-backtick fence (unchanged
        // from the pre-fix behaviour). Covers the common case.
        assert_eq!(fence_for(""), "```");
        assert_eq!(fence_for("plain text\nno backticks here"), "```");
    }

    #[test]
    fn fence_for_output_with_triple_backticks_uses_four() {
        // Closes issue #44: `!cat` on a Markdown file with ```` ``` ```` fences
        // used to break the outer fence. Dynamic length picks one larger.
        let md = "Some text\n```rust\nfn main() {}\n```\nmore text";
        assert_eq!(fence_for(md), "````");
    }

    #[test]
    fn fence_for_scales_to_longest_run() {
        // A 6-backtick line forces a 7-backtick outer fence.
        let exotic = "before\n``````\nafter";
        assert_eq!(fence_for(exotic), "```````");
    }

    #[test]
    fn fence_for_inline_backticks_dont_force_longer_fence() {
        // Inline `code` spans with single backticks don't change the fence
        // (CommonMark fences only close at line start; the helper still
        // counts leading runs only).
        let inline = "Use `grep` and `sed` for this, but not ``foo``.";
        assert_eq!(fence_for(inline), "```");
    }

    #[test]
    fn fence_for_indented_closing_fence_still_counted() {
        // CommonMark allows ≤ 3 spaces of indent on a closing fence. The
        // helper strips those before counting backticks so an indented
        // ` ```` line still bumps the outer fence to 4.
        let indented = "intro\n   ```\nbody\n   ```";
        assert_eq!(fence_for(indented), "````");
    }

    static ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    struct AgentDirEnvGuard {
        _guard: tokio::sync::MutexGuard<'static, ()>,
    }

    impl Drop for AgentDirEnvGuard {
        fn drop(&mut self) {
            std::env::remove_var("CAPO_AGENT_DIR");
        }
    }

    async fn with_agent_dir_env(dir: &std::path::Path) -> AgentDirEnvGuard {
        let guard = ENV_LOCK.lock().await;
        std::env::set_var("CAPO_AGENT_DIR", dir);
        AgentDirEnvGuard { _guard: guard }
    }

    async fn recv_settings_notice(ui_rx: &mut tokio::sync::mpsc::Receiver<UiEvent>) -> bool {
        let deadline = tokio::time::sleep(std::time::Duration::from_secs(2));
        tokio::pin!(deadline);
        loop {
            tokio::select! {
                Some(ev) = ui_rx.recv() => {
                    if matches!(&ev, UiEvent::Notice { title, .. } if title == "/settings") {
                        return true;
                    }
                }
                _ = &mut deadline => return false,
            }
        }
    }

    async fn recv_reload_error(ui_rx: &mut tokio::sync::mpsc::Receiver<UiEvent>) -> bool {
        let deadline = tokio::time::sleep(std::time::Duration::from_secs(2));
        tokio::pin!(deadline);
        loop {
            tokio::select! {
                Some(ev) = ui_rx.recv() => {
                    if matches!(&ev, UiEvent::Error(msg) if msg.contains("reload failed")) {
                        return true;
                    }
                }
                _ = &mut deadline => return false,
            }
        }
    }

    async fn recv_state_snapshot_model(
        ui_rx: &mut tokio::sync::mpsc::Receiver<UiEvent>,
    ) -> Option<String> {
        let deadline = tokio::time::sleep(std::time::Duration::from_secs(2));
        tokio::pin!(deadline);
        loop {
            tokio::select! {
                Some(ev) = ui_rx.recv() => {
                    if let UiEvent::StateSnapshot { model, .. } = ev {
                        return Some(model);
                    }
                }
                _ = &mut deadline => return None,
            }
        }
    }

    #[test]
    fn read_git_branch_parses_head_ref() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let git = dir.path().join(".git");
        std::fs::create_dir_all(&git).unwrap();
        std::fs::write(git.join("HEAD"), "ref: refs/heads/main\n").unwrap();
        assert_eq!(read_git_branch(dir.path()).as_deref(), Some("main"));

        // a nested cwd still finds the repo root by walking up
        let sub = dir.path().join("a/b");
        std::fs::create_dir_all(&sub).unwrap();
        assert_eq!(read_git_branch(&sub).as_deref(), Some("main"));

        // detached HEAD (raw sha) → None
        std::fs::write(git.join("HEAD"), "0123456789abcdef\n").unwrap();
        assert_eq!(read_git_branch(dir.path()), None);

        // no repo → None
        let bare = tempfile::tempdir().unwrap();
        assert_eq!(read_git_branch(bare.path()), None);
    }

    #[test]
    fn load_settings_toml_accepts_partial_toml_defaults() {
        let dir = tempfile::tempdir().expect("tempdir");
        let agent_dir = dir.path().join("agent");
        std::fs::create_dir_all(&agent_dir).expect("create agent dir");
        std::fs::write(
            agent_dir.join("settings.toml"),
            r#"
[ui]
footer_show_cost = false
"#,
        )
        .expect("write settings.toml");

        let settings = load_settings_toml(&agent_dir).expect("load settings");
        assert!(!settings.ui.footer_show_cost);
        assert_eq!(settings.ui.theme, capo_agent::Settings::default().ui.theme);
        assert_eq!(
            settings.model.name,
            capo_agent::Settings::default().model.name
        );
    }

    #[tokio::test]
    async fn apply_settings_rebuilds_and_writes_disk() {
        use capo_agent::{test_support::ScriptedLlm, Settings};

        let dir = tempfile::tempdir().expect("tempdir");
        let agent_dir = dir.path().join("agent");
        let _env_guard = with_agent_dir_env(&agent_dir).await;

        let llm = Arc::new(ScriptedLlm::builder().then_message("ok").build());
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm)
            .with_session_store(Arc::new(motosan_agent_loop::FileSessionStore::new(
                dir.path().join("sessions"),
            )))
            .build()
            .await
            .expect("build");
        let app = Arc::new(app);

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel(64);
        let runtime = tokio::spawn(forward_commands(Arc::clone(&app), ui_tx, cmd_rx));

        let mut new_settings = Settings::default();
        new_settings.ui.footer_show_cost = false;
        cmd_tx
            .send(Command::ApplySettings {
                settings: new_settings,
            })
            .await
            .expect("send");

        assert!(
            recv_settings_notice(&mut ui_rx).await,
            "expected Notice event"
        );
        let saved = std::fs::read_to_string(agent_dir.join("settings.toml"))
            .expect("settings.toml not written");
        assert!(saved.contains("footer_show_cost = false"), "{saved}");

        let _ = cmd_tx.send(Command::Quit).await;
        let _ = runtime.await;
    }

    #[tokio::test]
    async fn reload_settings_reads_disk_and_rebuilds() {
        use capo_agent::{test_support::ScriptedLlm, Settings};

        let dir = tempfile::tempdir().expect("tempdir");
        let agent_dir = dir.path().join("agent");
        std::fs::create_dir_all(&agent_dir).expect("create agent dir");
        let _env_guard = with_agent_dir_env(&agent_dir).await;

        let mut disk_settings = Settings::default();
        disk_settings.model.name = "claude-opus-4-7".into();
        disk_settings.ui.footer_show_cost = false;
        let toml = toml::to_string_pretty(&disk_settings).expect("toml");
        std::fs::write(agent_dir.join("settings.toml"), toml).expect("write settings.toml");

        let llm = Arc::new(ScriptedLlm::builder().then_message("ok").build());
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm)
            .with_session_store(Arc::new(motosan_agent_loop::FileSessionStore::new(
                dir.path().join("sessions"),
            )))
            .build()
            .await
            .expect("build");
        let app = Arc::new(app);

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel(64);
        let runtime = tokio::spawn(forward_commands(Arc::clone(&app), ui_tx, cmd_rx));

        cmd_tx.send(Command::ReloadSettings).await.expect("send");

        assert!(
            recv_settings_notice(&mut ui_rx).await,
            "expected Notice event after ReloadSettings"
        );

        cmd_tx
            .send(Command::GetState)
            .await
            .expect("send get_state");
        assert_eq!(
            recv_state_snapshot_model(&mut ui_rx).await.as_deref(),
            Some("claude-opus-4-7")
        );

        let _ = cmd_tx.send(Command::Quit).await;
        let _ = runtime.await;
    }

    #[tokio::test]
    async fn reload_settings_with_malformed_toml_emits_error_keeps_session() {
        use capo_agent::test_support::ScriptedLlm;

        let dir = tempfile::tempdir().expect("tempdir");
        let agent_dir = dir.path().join("agent");
        std::fs::create_dir_all(&agent_dir).expect("create agent dir");
        let _env_guard = with_agent_dir_env(&agent_dir).await;
        std::fs::write(agent_dir.join("settings.toml"), "not = valid = toml").expect("write");

        let llm = Arc::new(ScriptedLlm::builder().then_message("ok").build());
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm)
            .with_session_store(Arc::new(motosan_agent_loop::FileSessionStore::new(
                dir.path().join("sessions"),
            )))
            .build()
            .await
            .expect("build");
        let app = Arc::new(app);

        let session_id = app.session_id();
        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel(64);
        let runtime = tokio::spawn(forward_commands(Arc::clone(&app), ui_tx, cmd_rx));

        cmd_tx.send(Command::ReloadSettings).await.expect("send");

        assert!(
            recv_reload_error(&mut ui_rx).await,
            "expected Error event after malformed reload"
        );
        assert_eq!(app.session_id(), session_id);

        let _ = cmd_tx.send(Command::Quit).await;
        let _ = runtime.await;
    }

    #[test]
    fn print_mode_does_not_duplicate_streamed_final_text() {
        let events = vec![
            UiEvent::AgentTurnStarted,
            UiEvent::AgentThinking,
            UiEvent::AgentTextDelta("hello".into()),
            UiEvent::AgentTextDelta(" world".into()),
            UiEvent::AgentMessageComplete("hello world".into()),
            UiEvent::AgentTurnComplete,
        ];

        let rendered = render_print_mode_events(&events)
            .unwrap_or_else(|e| panic!("unexpected render error: {e}"));

        assert_eq!(rendered.stdout, "hello world\n");
        assert_eq!(rendered.stderr, "[thinking]\n");
    }

    #[test]
    fn print_mode_prints_non_streamed_final_text_once() {
        let events = vec![
            UiEvent::AgentTurnStarted,
            UiEvent::AgentMessageComplete("hello world".into()),
            UiEvent::AgentTurnComplete,
        ];

        let rendered = render_print_mode_events(&events)
            .unwrap_or_else(|e| panic!("unexpected render error: {e}"));

        assert_eq!(rendered.stdout, "hello world\n");
        assert!(rendered.stderr.is_empty());
    }

    #[test]
    fn print_mode_renders_tool_events_to_stderr() {
        let events = vec![
            UiEvent::ToolCallStarted {
                id: "tool_1".into(),
                name: "bash".into(),
                args: serde_json::json!({"command": "echo hi"}),
            },
            UiEvent::ToolCallCompleted {
                id: "tool_1".into(),
                result: UiToolResult {
                    is_error: false,
                    text: "ok".into(),
                },
            },
        ];

        let rendered = render_print_mode_events(&events)
            .unwrap_or_else(|e| panic!("unexpected render error: {e}"));

        assert!(rendered.stderr.contains("[tool:start] bash"));
        assert!(rendered.stderr.contains("[tool:done]"));
    }

    #[test]
    fn truncate_utf8_never_splits_multibyte_characters() {
        let mut text = "a😀".repeat(10_000);
        truncate_utf8(&mut text, 30 * 1024);
        assert!(text.len() <= 30 * 1024);
        assert!(text.is_char_boundary(text.len()));
    }

    #[tokio::test]
    async fn inline_bash_times_out() {
        let output = run_inline_bash_with_timeout("sleep 5", Duration::from_millis(20)).await;
        assert!(output.contains("timed out"), "got: {output}");
    }

    fn extension_fixture_path(name: &str) -> std::path::PathBuf {
        std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .parent()
            .unwrap_or_else(|| panic!("capo crate should have workspace parent"))
            .join("capo-agent/tests/fixtures/extensions")
            .join(name)
    }

    fn extension_entry(
        name: &str,
        fixture: &str,
        hooks: Vec<&str>,
        commands: Vec<&str>,
    ) -> capo_agent::extensions::ExtensionEntry {
        capo_agent::extensions::ExtensionEntry {
            name: name.into(),
            command: extension_fixture_path(fixture)
                .to_string_lossy()
                .into_owned(),
            args: Vec::new(),
            env: std::collections::HashMap::new(),
            timeout_ms: None,
            hooks: hooks.into_iter().map(String::from).collect(),
            commands: commands.into_iter().map(String::from).collect(),
        }
    }

    async fn app_with_extensions(
        cwd: &std::path::Path,
        entries: Vec<capo_agent::extensions::ExtensionEntry>,
    ) -> (
        Arc<capo_agent::App>,
        Arc<capo_agent::test_support::ScriptedLlm>,
    ) {
        use capo_agent::extensions::{ExtensionManifestFile, ExtensionRegistry};
        use capo_agent::test_support::ScriptedLlm;

        let mut diagnostics = Vec::new();
        let registry = Arc::new(ExtensionRegistry::build(
            ExtensionManifestFile {
                extensions: entries,
            },
            &mut diagnostics,
        ));
        assert!(
            diagnostics.is_empty(),
            "extension diagnostics: {diagnostics:?}"
        );

        let llm = Arc::new(ScriptedLlm::builder().then_message("done").build());
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(cwd)
            .with_builtin_tools()
            .with_llm(llm.clone())
            .with_extension_registry(registry)
            .build()
            .await
            .unwrap_or_else(|e| panic!("build app: {e}"));

        (Arc::new(app), llm)
    }

    async fn app_with_extension(
        cwd: &std::path::Path,
        fixture: &str,
        hooks: Vec<&str>,
        commands: Vec<&str>,
    ) -> (
        Arc<capo_agent::App>,
        Arc<capo_agent::test_support::ScriptedLlm>,
    ) {
        app_with_extensions(
            cwd,
            vec![extension_entry("fixture", fixture, hooks, commands)],
        )
        .await
    }

    #[tokio::test]
    async fn before_user_message_cancel_emits_extension_cancelled_and_skips_turn() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let (app, llm) = app_with_extension(
            dir.path(),
            "cancel_with_reason.sh",
            vec!["before_user_message"],
            Vec::new(),
        )
        .await;

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(64);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        cmd_tx
            .send(Command::SendUserMessage {
                text: "go".into(),
                attachments: Vec::new(),
            })
            .await
            .unwrap_or_else(|e| panic!("send command: {e}"));

        let first = tokio::time::timeout(Duration::from_secs(2), ui_rx.recv())
            .await
            .unwrap_or_else(|e| panic!("timed out waiting for cancel event: {e}"))
            .unwrap_or_else(|| panic!("ui channel closed"));
        assert!(
            matches!(&first, UiEvent::ExtensionCancelled { extension_name, reason }
                if extension_name == "fixture"
                    && reason.as_deref() == Some("fixture cancel reason")),
            "expected ExtensionCancelled; got {first:?}"
        );

        tokio::time::sleep(Duration::from_millis(20)).await;
        assert!(
            llm.captured_inputs().is_empty(),
            "LLM should not have been invoked"
        );

        cmd_tx
            .send(Command::Quit)
            .await
            .unwrap_or_else(|e| panic!("send quit: {e}"));
        forwarder
            .await
            .unwrap_or_else(|e| panic!("forwarder join: {e}"));
    }

    async fn run_extension_send_and_capture_user_text(
        app: Arc<capo_agent::App>,
        llm: Arc<capo_agent::test_support::ScriptedLlm>,
        command_name: &str,
    ) -> String {
        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(64);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        cmd_tx
            .send(Command::InvokeExtensionCommand {
                name: command_name.into(),
                args: String::new(),
            })
            .await
            .unwrap_or_else(|e| panic!("send command: {e}"));

        let saw_turn_complete = tokio::time::timeout(Duration::from_secs(2), async {
            while let Some(event) = ui_rx.recv().await {
                if matches!(event, UiEvent::AgentTurnComplete) {
                    return true;
                }
            }
            false
        })
        .await
        .unwrap_or_else(|e| panic!("timed out waiting for turn complete: {e}"));
        assert!(saw_turn_complete, "turn did not complete");

        let captured = llm.captured_inputs();
        assert!(!captured.is_empty(), "LLM should have been invoked");
        let last_call = captured.last().unwrap_or_else(|| panic!("captured"));
        let user_msg = last_call
            .iter()
            .rev()
            .find(|m| matches!(m, Message::User { .. }))
            .unwrap_or_else(|| panic!("user message"));
        let parts = match user_msg {
            Message::User { content, .. } => content.as_slice(),
            _ => unreachable!(),
        };
        let text = match &parts[0] {
            ContentPart::Text { text, .. } => text.clone(),
            other => panic!("expected text part; got {other:?}"),
        };

        cmd_tx
            .send(Command::Quit)
            .await
            .unwrap_or_else(|e| panic!("send quit: {e}"));
        forwarder
            .await
            .unwrap_or_else(|e| panic!("forwarder join: {e}"));

        text
    }

    #[tokio::test]
    async fn invoke_extension_command_send_starts_a_real_turn() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let (app, llm) =
            app_with_extension(dir.path(), "send_command.sh", Vec::new(), vec!["sendecho"]).await;

        let text = run_extension_send_and_capture_user_text(app, llm, "sendecho").await;

        assert_eq!(text, "sent from fixture");
    }

    #[tokio::test]
    async fn invoke_extension_command_send_runs_before_user_message_hooks() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let entries = vec![
            extension_entry("sender", "send_command.sh", Vec::new(), vec!["sendecho"]),
            extension_entry(
                "prefixer",
                "transform_text.sh",
                vec!["before_user_message"],
                Vec::new(),
            ),
        ];
        let (app, llm) = app_with_extensions(dir.path(), entries).await;

        let text = run_extension_send_and_capture_user_text(app, llm, "sendecho").await;

        assert_eq!(text, "[ext] sent from fixture");
    }

    struct PendingLlm;

    #[async_trait]
    impl LlmClient for PendingLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            futures::future::pending::<motosan_agent_loop::Result<ChatOutput>>().await
        }
    }

    #[tokio::test]
    async fn invoke_extension_command_send_blocks_session_switch_while_running() {
        use capo_agent::extensions::{ExtensionManifestFile, ExtensionRegistry};

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let entry = extension_entry("sender", "send_command.sh", Vec::new(), vec!["sendecho"]);
        let mut diagnostics = Vec::new();
        let registry = Arc::new(ExtensionRegistry::build(
            ExtensionManifestFile {
                extensions: vec![entry],
            },
            &mut diagnostics,
        ));
        assert!(
            diagnostics.is_empty(),
            "extension diagnostics: {diagnostics:?}"
        );

        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(PendingLlm))
            .with_extension_registry(registry)
            .build()
            .await
            .unwrap_or_else(|e| panic!("build app: {e}"));

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(64);
        let forwarder = tokio::spawn(forward_commands(Arc::new(app), ui_tx, cmd_rx));

        cmd_tx
            .send(Command::InvokeExtensionCommand {
                name: "sendecho".into(),
                args: String::new(),
            })
            .await
            .unwrap_or_else(|e| panic!("send extension command: {e}"));

        match tokio::time::timeout(Duration::from_secs(2), ui_rx.recv()).await {
            Ok(Some(UiEvent::AgentTurnStarted)) => {}
            Ok(Some(other)) => panic!("expected turn start, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(err) => panic!("timed out waiting for turn start: {err}"),
        }

        cmd_tx
            .send(Command::NewSession)
            .await
            .unwrap_or_else(|e| panic!("send /new: {e}"));

        let mut saw_guard = false;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await
        {
            if matches!(event, UiEvent::Error(message) if message.contains("before /new")) {
                saw_guard = true;
                break;
            }
        }
        assert!(
            saw_guard,
            "/new was not rejected while extension send turn ran"
        );

        cmd_tx
            .send(Command::Quit)
            .await
            .unwrap_or_else(|e| panic!("send quit: {e}"));
        forwarder
            .await
            .unwrap_or_else(|e| panic!("forwarder join: {e}"));
    }

    #[tokio::test]
    async fn command_forwarder_handles_resolution_while_turn_is_running() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(PendingLlm))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (ui_tx, _ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(Arc::clone(&app), ui_tx, cmd_rx));

        if cmd_tx
            .send(Command::SendUserMessage {
                text: "hi".into(),
                attachments: Vec::new(),
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }
        tokio::task::yield_now().await;

        let args = serde_json::json!({"command": "echo hi"});
        let key = capo_agent::permissions::SessionCache::key("bash", &args);
        if cmd_tx
            .send(Command::ResolvePermission(PermissionResolution {
                tool: "bash".into(),
                args,
                choice: PermissionChoice::AllowSession,
            }))
            .await
            .is_err()
        {
            panic!("command channel closed");
        }

        let observed = tokio::time::timeout(Duration::from_millis(200), async {
            loop {
                if app.permissions_cache().get(&key).is_some() {
                    break true;
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        })
        .await
        .unwrap_or(false);

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), forwarder).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => panic!("forwarder join failed: {err}"),
            Err(_) => panic!("forwarder did not exit"),
        }
        assert!(observed, "session cache was not updated while turn ran");
    }

    /// Shared setup for the `inline_bash_*` test family: tempdir-backed app
    /// with `PendingLlm`, bounded mpsc channels, and a spawned
    /// `forward_commands` task. Returns the harness; tests destructure or
    /// access fields directly, and call `quit_harness` to shut down.
    ///
    /// Scoped to the inline-bash tests because the broader forwarder tests
    /// in this file have setup variations (Arc::clone, different LLM stubs,
    /// `_ui_rx` discard, etc.) that don't fit a single helper. Extending
    /// the harness to cover them is a separate refactor.
    struct InlineBashHarness {
        dir: tempfile::TempDir,
        cmd_tx: tokio::sync::mpsc::Sender<Command>,
        ui_rx: tokio::sync::mpsc::Receiver<UiEvent>,
        forwarder: tokio::task::JoinHandle<()>,
    }

    async fn setup_inline_bash_harness() -> InlineBashHarness {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(PendingLlm))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        InlineBashHarness {
            dir,
            cmd_tx,
            ui_rx,
            forwarder,
        }
    }

    /// Quit the forwarder cleanly and assert it exits within `timeout`.
    /// Returns the `TempDir` so callers can run post-teardown filesystem
    /// assertions before letting it drop.
    async fn quit_harness(harness: InlineBashHarness, timeout: Duration) -> tempfile::TempDir {
        if harness.cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(timeout, harness.forwarder).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => panic!("forwarder join failed: {err}"),
            Err(_) => panic!("forwarder did not exit"),
        }
        harness.dir
    }

    #[tokio::test]
    async fn command_forwarder_rejects_inline_bash_send_before_running_shell() {
        let mut harness = setup_inline_bash_harness().await;
        let marker = harness.dir.path().join("inline-bash-should-not-run");

        if harness
            .cmd_tx
            .send(Command::SendUserMessage {
                text: "first".into(),
                attachments: Vec::new(),
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), harness.ui_rx.recv()).await {
            Ok(Some(UiEvent::AgentTurnStarted)) => {}
            Ok(Some(other)) => panic!("expected turn start, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for first turn"),
        }

        let command = format!("touch {}", marker.display());
        if harness
            .cmd_tx
            .send(Command::RunInlineBash {
                command,
                send_to_llm: true,
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }

        let mut saw_rejection = false;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(200), harness.ui_rx.recv()).await
        {
            if matches!(event, UiEvent::Error(message) if message.contains("cannot run !cmd")) {
                saw_rejection = true;
                break;
            }
        }

        let _dir = quit_harness(harness, Duration::from_millis(200)).await;
        assert!(saw_rejection, "inline bash send was not rejected");
        assert!(!marker.exists(), "inline bash command ran before rejection");
    }

    #[tokio::test]
    async fn inline_bash_send_to_llm_also_emits_transcript_output() {
        // Regression: `!cmd` (send_to_llm: true) used to feed the LLM but
        // emit NO `InlineBashOutput`, so the user saw nothing in transcript
        // about the bash execution — looked like "the exclamation mark
        // didn't work". Now both branches must emit InlineBashOutput so the
        // user sees `$ cmd → output` regardless of whether the LLM was fed.
        let mut harness = setup_inline_bash_harness().await;

        // No active turn → send_to_llm: true should run bash AND emit the
        // transcript output AND spawn the LLM task (which we don't wait on).
        if harness
            .cmd_tx
            .send(Command::RunInlineBash {
                command: "echo MARKER_42".into(),
                send_to_llm: true,
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }

        // Collect events for up to 500ms; we're looking for InlineBashOutput.
        let mut saw_output = false;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(500), harness.ui_rx.recv()).await
        {
            if let UiEvent::InlineBashOutput { command, output } = event {
                assert_eq!(command, "echo MARKER_42");
                assert!(
                    output.contains("MARKER_42"),
                    "bash output should contain the echo'd marker, got: {output:?}"
                );
                saw_output = true;
                break;
            }
        }

        let _dir = quit_harness(harness, Duration::from_millis(500)).await;
        assert!(
            saw_output,
            "InlineBashOutput must be emitted for send_to_llm=true (was missing before the fix)"
        );
    }

    #[tokio::test]
    async fn inline_bash_no_llm_emits_transcript_output_only() {
        // Symmetric counterpart to inline_bash_send_to_llm_also_emits_transcript_output:
        // `!!cmd` (send_to_llm: false) must emit InlineBashOutput AND must NOT
        // spawn an LLM turn (no AgentTurnStarted event). Locks both halves of
        // the contract — that the emit DOES happen, and that the path stays
        // out of the agent layer entirely. The corresponding send_to_llm:true
        // test only checks the emit; this one also checks the side-effect
        // boundary.
        let mut harness = setup_inline_bash_harness().await;

        if harness
            .cmd_tx
            .send(Command::RunInlineBash {
                command: "echo NOLLM_MARKER".into(),
                send_to_llm: false,
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }

        // Drain events for the full collection window so the negative
        // assertion about AgentTurnStarted is meaningful — breaking on the
        // first InlineBashOutput would silently pass a buggy version that
        // emits an LLM turn afterwards.
        let mut saw_output = false;
        let mut saw_turn_started = false;
        let collect_deadline = std::time::Instant::now() + Duration::from_millis(500);
        while std::time::Instant::now() < collect_deadline {
            match tokio::time::timeout(Duration::from_millis(100), harness.ui_rx.recv()).await {
                Ok(Some(UiEvent::InlineBashOutput { command, output })) => {
                    assert_eq!(command, "echo NOLLM_MARKER");
                    assert!(
                        output.contains("NOLLM_MARKER"),
                        "bash output should contain the marker, got: {output:?}"
                    );
                    saw_output = true;
                }
                Ok(Some(UiEvent::AgentTurnStarted)) => {
                    saw_turn_started = true;
                }
                Ok(Some(_)) => {}
                Ok(None) | Err(_) => break,
            }
        }

        let _dir = quit_harness(harness, Duration::from_millis(500)).await;
        assert!(saw_output, "InlineBashOutput must be emitted for !!cmd");
        assert!(
            !saw_turn_started,
            "!!cmd must NOT spawn an LLM turn (saw AgentTurnStarted)"
        );
    }

    #[tokio::test]
    async fn command_forwarder_rejects_compact_while_turn_is_running() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(PendingLlm))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx
            .send(Command::SendUserMessage {
                text: "first".into(),
                attachments: Vec::new(),
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await {
            Ok(Some(UiEvent::AgentTurnStarted)) => {}
            Ok(Some(other)) => panic!("expected turn start, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for first turn"),
        }

        if cmd_tx.send(Command::Compact).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await {
            Ok(Some(UiEvent::AgentThinking)) => {}
            Ok(Some(other)) => panic!("expected thinking before compact rejection, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for thinking event"),
        }
        match tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await {
            Ok(Some(UiEvent::Error(message))) => {
                assert!(message.contains("cannot compact while an agent turn is running"));
            }
            Ok(Some(other)) => panic!("expected compact rejection, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for compact rejection"),
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), forwarder).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => panic!("forwarder join failed: {err}"),
            Err(_) => panic!("forwarder did not exit"),
        }
    }

    struct DoneLlm;

    #[async_trait]
    impl LlmClient for DoneLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            Ok(ChatOutput::new(motosan_agent_loop::LlmResponse::Message(
                "done".into(),
            )))
        }
    }

    #[tokio::test]
    async fn get_state_reply_carries_session_id_and_model() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };
        let expected_id = app.session_id();

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx.send(Command::GetState).await.is_err() {
            panic!("command channel closed");
        }
        let mut snapshot = None;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await
        {
            if let UiEvent::StateSnapshot {
                session_id,
                active_turn,
                ..
            } = event
            {
                snapshot = Some((session_id, active_turn));
                break;
            }
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        let _ = tokio::time::timeout(Duration::from_millis(200), forwarder).await;

        let (got_id, active) = snapshot.expect("state_snapshot not emitted");
        assert_eq!(got_id, expected_id);
        assert!(!active, "no turn was started");
    }

    #[tokio::test]
    async fn list_extensions_surfaces_load_diagnostics() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let diagnostics = Arc::new(vec![capo_agent::ExtensionDiagnostic {
            extension_name: "<manifest>".into(),
            severity: capo_agent::extensions::DiagnosticSeverity::Error,
            message: "could not parse extensions.toml".into(),
        }]);
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .with_extension_registry(Arc::new(capo_agent::ExtensionRegistry::default()))
            .with_extension_diagnostics(diagnostics)
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx.send(Command::ListExtensions).await.is_err() {
            panic!("command channel closed");
        }
        let event = match tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await {
            Ok(Some(event)) => event,
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for extension list"),
        };
        match event {
            UiEvent::ExtensionList { extensions } => {
                assert_eq!(extensions.len(), 1);
                assert!(!extensions[0].healthy);
                assert_eq!(extensions[0].name, "<manifest>");
                assert!(extensions[0]
                    .diagnostic
                    .as_deref()
                    .unwrap_or_default()
                    .contains("could not parse"));
            }
            other => panic!("expected ExtensionList, got {other:?}"),
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        let _ = tokio::time::timeout(Duration::from_millis(200), forwarder).await;
    }

    #[tokio::test]
    async fn cancelled_fork_does_not_replace_transcript() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let script = dir.path().join("cancel-fork.sh");
        std::fs::write(
            &script,
            "cat > /dev/null\necho '{\"action\":\"cancel\",\"reason\":\"stop fork\"}'\n",
        )
        .unwrap_or_else(|e| panic!("write script: {e}"));

        let entry = capo_agent::ExtensionEntry {
            name: "fork_guard".into(),
            command: "/bin/sh".into(),
            args: vec![script.to_string_lossy().into_owned()],
            env: std::collections::HashMap::new(),
            timeout_ms: Some(1000),
            hooks: vec!["session_before_switch".into()],
            commands: Vec::new(),
        };
        let mut diagnostics = Vec::new();
        let registry = capo_agent::ExtensionRegistry::build(
            capo_agent::extensions::ExtensionManifestFile {
                extensions: vec![entry],
            },
            &mut diagnostics,
        );
        assert!(
            diagnostics.is_empty(),
            "unexpected diagnostics: {diagnostics:?}"
        );

        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .with_extension_registry(Arc::new(registry))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx
            .send(Command::ForkFrom {
                from: "not-polled-on-cancel".into(),
                message: "branch".into(),
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(500), ui_rx.recv()).await {
            Ok(Some(UiEvent::Error(message))) => assert!(message.contains("stop fork")),
            Ok(Some(other)) => panic!("expected cancellation error, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for fork cancellation"),
        }
        match tokio::time::timeout(Duration::from_millis(150), ui_rx.recv()).await {
            Ok(Some(UiEvent::SessionReplaced { .. })) => {
                panic!("cancelled fork replaced transcript")
            }
            Ok(Some(other)) => panic!("cancelled fork emitted unexpected event: {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => {}
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        let _ = tokio::time::timeout(Duration::from_millis(200), forwarder).await;
    }

    #[tokio::test]
    async fn get_messages_reply_returns_empty_for_fresh_session() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx.send(Command::GetMessages).await.is_err() {
            panic!("command channel closed");
        }
        let mut got = None;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await
        {
            if let UiEvent::MessagesSnapshot(msgs) = event {
                got = Some(msgs);
                break;
            }
        }
        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        let _ = tokio::time::timeout(Duration::from_millis(200), forwarder).await;
        assert!(got.expect("snapshot").is_empty());
    }

    #[tokio::test]
    async fn rpc_mode_eof_while_permission_pending_exits() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let policy = dir.path().join("permissions.toml");
        tokio::fs::write(&policy, "")
            .await
            .unwrap_or_else(|e| panic!("write policy: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_permissions_config(policy)
            .with_llm(Arc::new(BashPromptLlm))
            .with_ui_channel(ui_tx.clone())
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };
        let (mut client_to_server_write, server_stdin) = tokio::io::duplex(4096);
        let (_server_stdout_read, server_stdout) = tokio::io::duplex(4096);
        let mut server = tokio::spawn(run_rpc_mode(
            Arc::clone(&app),
            dir.path().to_path_buf(),
            server_stdin,
            server_stdout,
            ui_tx,
            ui_rx,
            cmd_tx,
            cmd_rx,
        ));

        use tokio::io::AsyncWriteExt;
        client_to_server_write
            .write_all(b"{\"type\":\"send_user_message\",\"payload\":{\"text\":\"run bash\"}}\n")
            .await
            .expect("send");
        drop(client_to_server_write);

        match tokio::time::timeout(Duration::from_millis(500), &mut server).await {
            Ok(joined) => {
                joined.expect("server panicked").expect("server io");
            }
            Err(_) => {
                server.abort();
                panic!("rpc server did not exit after EOF during permission prompt");
            }
        }
    }

    #[tokio::test]
    async fn rpc_mode_exits_when_stdout_writer_fails_while_stdin_open() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .with_ui_channel(ui_tx.clone())
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };
        let (mut client_to_server_write, server_stdin) = tokio::io::duplex(4096);
        let mut server = tokio::spawn(run_rpc_mode(
            Arc::clone(&app),
            dir.path().to_path_buf(),
            server_stdin,
            FailAfterHeaderWriter::default(),
            ui_tx,
            ui_rx,
            cmd_tx,
            cmd_rx,
        ));

        use tokio::io::AsyncWriteExt;
        client_to_server_write
            .write_all(b"{\"type\":\"send_user_message\",\"payload\":{\"text\":\"hello\"}}\n")
            .await
            .expect("send");

        match tokio::time::timeout(Duration::from_millis(500), &mut server).await {
            Ok(joined) => {
                let err = joined
                    .expect("server panicked")
                    .expect_err("stdout write failure should be returned");
                assert!(format!("{err}").contains("client disconnected"), "{err}");
            }
            Err(_) => {
                server.abort();
                panic!("rpc server did not exit after stdout writer failed");
            }
        }
    }

    #[tokio::test]
    async fn rpc_mode_executes_a_send_user_message_command() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());

        // Channels first — App, forwarder, and writer all share one ui_tx
        // so PermissionExtension events reach the bridge (see Step 3).
        let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);

        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .with_ui_channel(ui_tx.clone())
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        // In-memory client ↔ server pipes.
        let (mut client_to_server_write, server_stdin) = tokio::io::duplex(4096);
        let (mut server_stdout_read, server_stdout) = tokio::io::duplex(4096);

        let server = tokio::spawn(run_rpc_mode(
            Arc::clone(&app),
            dir.path().to_path_buf(),
            server_stdin,
            server_stdout,
            ui_tx,
            ui_rx,
            cmd_tx,
            cmd_rx,
        ));

        // Drive a one-message turn, then quit.
        use tokio::io::AsyncWriteExt;
        client_to_server_write
            .write_all(b"{\"type\":\"send_user_message\",\"payload\":{\"text\":\"hello\"}}\n")
            .await
            .expect("send");
        client_to_server_write
            .write_all(b"{\"type\":\"quit\"}\n")
            .await
            .expect("send");
        drop(client_to_server_write);

        // Read everything the server emitted.
        use tokio::io::AsyncReadExt;
        let mut buf = String::new();
        server_stdout_read
            .read_to_string(&mut buf)
            .await
            .expect("read");
        server.await.expect("server panicked").expect("server io");

        let lines: Vec<&str> = buf.lines().collect();
        assert!(!lines.is_empty(), "no output");
        assert!(
            lines[0].contains(r#""type":"session""#),
            "first line should be session header: {}",
            lines[0]
        );
        // Every line is valid JSON.
        for l in &lines {
            serde_json::from_str::<serde_json::Value>(l)
                .unwrap_or_else(|e| panic!("bad line {l:?}: {e}"));
        }
        // Some agent event appeared.
        assert!(lines.iter().any(|l| l.contains(r#""type":"agent_"#)));
    }

    #[tokio::test]
    async fn rpc_bridge_round_trip_with_bash_tool_call() {
        use capo_agent::test_support::ScriptedLlm;
        use tokio::io::{AsyncBufReadExt, AsyncWriteExt};

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());

        // Script: first call → bash echo, second call → "done".
        let llm = ScriptedLlm::builder()
            .then_tool_call(
                "tc-1",
                "bash",
                serde_json::json!({ "command": "echo hi-from-bridge" }),
            )
            .then_message("done")
            .build();

        let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);

        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(llm))
            .with_ui_channel(ui_tx.clone())
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (mut client_to_server_write, server_stdin) = tokio::io::duplex(8192);
        let (mut server_stdout_read, server_stdout) = tokio::io::duplex(8192);

        let server = tokio::spawn(run_rpc_mode(
            Arc::clone(&app),
            dir.path().to_path_buf(),
            server_stdin,
            server_stdout,
            ui_tx,
            ui_rx,
            cmd_tx,
            cmd_rx,
        ));

        client_to_server_write
            .write_all(b"{\"type\":\"send_user_message\",\"payload\":{\"text\":\"run it\"}}\n")
            .await
            .expect("send user message");

        // Read stdout line-by-line until we see permission_requested. Then
        // extract its id, write the resolve_permission reply, then send quit
        // and drain to EOF.
        let mut reader = tokio::io::BufReader::new(&mut server_stdout_read);
        let mut all_lines: Vec<String> = Vec::new();
        let mut permission_id: Option<String> = None;
        let mut sent_resolve = false;

        loop {
            let mut line = String::new();
            let n = match tokio::time::timeout(
                std::time::Duration::from_secs(5),
                reader.read_line(&mut line),
            )
            .await
            {
                Ok(result) => result.expect("read failed"),
                Err(err) => panic!(
                    "read timed out — bridge may have hung: {err}; lines so far: {all_lines:?}"
                ),
            };
            if n == 0 {
                break; // EOF
            }
            let trimmed = line.trim_end_matches('\n').to_string();
            all_lines.push(trimmed.clone());

            if permission_id.is_none() {
                let v: serde_json::Value = serde_json::from_str(&trimmed).expect("valid json");
                if v["type"] == "permission_requested" {
                    let id = v["id"]
                        .as_str()
                        .expect("permission_requested missing id")
                        .to_string();
                    permission_id = Some(id);
                }
            }

            if let (Some(id), false) = (&permission_id, sent_resolve) {
                let reply = format!(
                    "{{\"type\":\"resolve_permission\",\"id\":\"{id}\",\"choice\":\"allow_once\"}}\n"
                );
                client_to_server_write
                    .write_all(reply.as_bytes())
                    .await
                    .expect("send resolve_permission");
                sent_resolve = true;
            }

            if sent_resolve && trimmed.contains(r#""type":"agent_turn_complete""#) {
                client_to_server_write
                    .write_all(b"{\"type\":\"quit\"}\n")
                    .await
                    .expect("send quit");
                drop(client_to_server_write);
                break;
            }
        }

        match tokio::time::timeout(std::time::Duration::from_secs(5), server).await {
            Ok(joined) => joined.expect("server panicked").expect("server io"),
            Err(_) => panic!("rpc server did not exit after quit; lines: {all_lines:?}"),
        }

        // Assertions: the bridge fired, the tool ran, the turn completed.
        assert!(
            permission_id.is_some(),
            "no permission_requested seen in: {all_lines:?}"
        );
        assert!(
            all_lines
                .iter()
                .any(|l| l.contains(r#""type":"tool_call_completed""#)
                    && !l.contains(r#""is_error":true"#)),
            "bash should have run successfully: {all_lines:?}"
        );
        assert!(
            all_lines
                .iter()
                .any(|l| l.contains(r#""type":"agent_message_complete""#) && l.contains("done")),
            "agent should have emitted its 'done' message: {all_lines:?}"
        );
    }

    #[tokio::test]
    async fn rpc_send_user_message_with_attachment_forwards_multipart() {
        use capo_agent::test_support::ScriptedLlm;
        use tokio::io::{AsyncBufReadExt, AsyncWriteExt};

        let path = png_fixture("e2e-rpc");
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let llm = Arc::new(ScriptedLlm::builder().then_message("acknowledged").build());

        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm.clone())
            .with_ui_channel(ui_tx.clone())
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build app: {err}"),
        };

        let (mut client_to_server_write, server_stdin) = tokio::io::duplex(8192);
        let (mut server_stdout_read, server_stdout) = tokio::io::duplex(8192);
        let server = tokio::spawn(run_rpc_mode(
            Arc::clone(&app),
            dir.path().to_path_buf(),
            server_stdin,
            server_stdout,
            ui_tx,
            ui_rx,
            cmd_tx,
            cmd_rx,
        ));

        let cmd = Command::SendUserMessage {
            text: "describe".into(),
            attachments: vec![capo_agent::Attachment::Image { path }],
        };
        let line = serde_json::to_string(&cmd).unwrap_or_else(|e| panic!("serialize: {e}"));
        assert!(
            line.contains(r#""type":"send_user_message""#),
            "wire form check: {line}"
        );
        client_to_server_write
            .write_all(format!("{line}\n").as_bytes())
            .await
            .unwrap_or_else(|e| panic!("send command: {e}"));

        let mut reader = tokio::io::BufReader::new(&mut server_stdout_read);
        let mut all_lines: Vec<String> = Vec::new();
        loop {
            let mut line = String::new();
            let n = match tokio::time::timeout(Duration::from_secs(5), reader.read_line(&mut line))
                .await
            {
                Ok(result) => result.unwrap_or_else(|e| panic!("read failed: {e}")),
                Err(err) => panic!(
                    "read timed out — rpc image test may have hung: {err}; lines so far: {all_lines:?}"
                ),
            };
            if n == 0 {
                break;
            }
            let trimmed = line.trim_end_matches('\n').to_string();
            all_lines.push(trimmed.clone());
            if trimmed.contains(r#""type":"agent_turn_complete""#) {
                client_to_server_write
                    .write_all(b"{\"type\":\"quit\"}\n")
                    .await
                    .unwrap_or_else(|e| panic!("send quit: {e}"));
                break;
            }
        }

        drop(client_to_server_write);
        drop(reader);
        let mut rest = String::new();
        use tokio::io::AsyncReadExt as _;
        server_stdout_read
            .read_to_string(&mut rest)
            .await
            .unwrap_or_else(|e| panic!("read rest: {e}"));
        server
            .await
            .unwrap_or_else(|e| panic!("server panicked: {e}"))
            .unwrap_or_else(|e| panic!("server io: {e}"));
        assert!(
            all_lines
                .iter()
                .any(|l| l.contains(r#""type":"agent_message_complete""#)),
            "expected message-complete event: {all_lines:?}"
        );

        let captured = llm.captured_inputs();
        assert!(!captured.is_empty(), "LLM never called");
        let last_call = captured.last().unwrap_or_else(|| panic!("captured"));
        let user_msg = last_call
            .iter()
            .rev()
            .find(|m| matches!(m, Message::User { .. }))
            .unwrap_or_else(|| panic!("user message"));
        let parts = match user_msg {
            Message::User { content, .. } => content.as_slice(),
            _ => unreachable!(),
        };
        assert!(matches!(&parts[0], ContentPart::Text { text, .. } if text == "describe"));
        assert!(parts.iter().any(|p| matches!(p, ContentPart::Image { .. })));
    }

    #[tokio::test]
    async fn rpc_allow_session_caches_identical_calls() {
        use capo_agent::test_support::ScriptedLlm;
        use tokio::io::AsyncWriteExt;

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());

        // Same bash call twice, then done. With allow_session on the first
        // resolution, the second call must not re-prompt.
        let args = serde_json::json!({"command": "echo cache-me"});
        let llm = ScriptedLlm::builder()
            .then_tool_call("tc-1", "bash", args.clone())
            .then_tool_call("tc-2", "bash", args)
            .then_message("done")
            .build();

        let (ui_tx, ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(64);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);

        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(llm))
            .with_ui_channel(ui_tx.clone())
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (mut client_write, server_stdin) = tokio::io::duplex(8192);
        let (mut server_stdout, server_out) = tokio::io::duplex(8192);

        let server = tokio::spawn(run_rpc_mode(
            Arc::clone(&app),
            dir.path().to_path_buf(),
            server_stdin,
            server_out,
            ui_tx,
            ui_rx,
            cmd_tx,
            cmd_rx,
        ));

        client_write
            .write_all(b"{\"type\":\"send_user_message\",\"payload\":{\"text\":\"run twice\"}}\n")
            .await
            .expect("send user message");

        let mut reader = tokio::io::BufReader::new(&mut server_stdout);
        let mut all_lines: Vec<String> = Vec::new();
        let mut resolve_sent = false;

        loop {
            let mut line = String::new();
            let n = tokio::time::timeout(
                std::time::Duration::from_secs(5),
                tokio::io::AsyncBufReadExt::read_line(&mut reader, &mut line),
            )
            .await
            .expect("read timed out")
            .expect("read failed");
            if n == 0 {
                break;
            }
            let trimmed = line.trim_end_matches('\n').to_string();
            all_lines.push(trimmed.clone());

            if !resolve_sent && trimmed.contains(r#""type":"permission_requested""#) {
                let v: serde_json::Value = serde_json::from_str(&trimmed).expect("json");
                let id = v["id"].as_str().expect("id").to_string();
                let reply = format!(
                    "{{\"type\":\"resolve_permission\",\"id\":\"{id}\",\"choice\":\"allow_session\"}}\n"
                );
                client_write
                    .write_all(reply.as_bytes())
                    .await
                    .expect("send resolve");
                resolve_sent = true;
            }

            if resolve_sent && trimmed.contains(r#""type":"agent_turn_complete""#) {
                client_write
                    .write_all(b"{\"type\":\"quit\"}\n")
                    .await
                    .expect("send quit");
                drop(client_write);
                break;
            }
        }

        match tokio::time::timeout(std::time::Duration::from_secs(5), server).await {
            Ok(joined) => joined.expect("server panicked").expect("server io"),
            Err(_) => panic!("rpc server did not exit; lines: {all_lines:?}"),
        }

        let prompt_count = all_lines
            .iter()
            .filter(|l| l.contains(r#""type":"permission_requested""#))
            .count();
        assert_eq!(
            prompt_count, 1,
            "allow_session must cache: expected 1 permission_requested, got {prompt_count}: {all_lines:?}"
        );
        // Both bash calls completed successfully.
        let success_count = all_lines
            .iter()
            .filter(|l| {
                l.contains(r#""type":"tool_call_completed""#) && !l.contains(r#""is_error":true"#)
            })
            .count();
        assert_eq!(
            success_count, 2,
            "both bash calls should succeed: got {success_count} successes: {all_lines:?}"
        );
    }

    struct BashPromptLlm;

    #[async_trait]
    impl LlmClient for BashPromptLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            Ok(ChatOutput::new(LlmResponse::ToolCalls(vec![
                ToolCallItem {
                    id: "bash-1".into(),
                    name: "bash".into(),
                    args: serde_json::json!({"command": "ls /etc"}),
                },
            ])))
        }
    }

    struct ErrorLlm;

    #[derive(Default)]
    struct FailAfterHeaderWriter {
        fail_after_newline: bool,
    }

    impl tokio::io::AsyncWrite for FailAfterHeaderWriter {
        fn poll_write(
            mut self: std::pin::Pin<&mut Self>,
            _cx: &mut std::task::Context<'_>,
            buf: &[u8],
        ) -> std::task::Poll<std::io::Result<usize>> {
            if self.fail_after_newline {
                return std::task::Poll::Ready(Err(std::io::Error::new(
                    std::io::ErrorKind::BrokenPipe,
                    "client disconnected",
                )));
            }
            if let Some(pos) = buf.iter().position(|b| *b == b'\n') {
                self.fail_after_newline = true;
                return std::task::Poll::Ready(Ok(pos + 1));
            }
            std::task::Poll::Ready(Ok(buf.len()))
        }

        fn poll_flush(
            self: std::pin::Pin<&mut Self>,
            _cx: &mut std::task::Context<'_>,
        ) -> std::task::Poll<std::io::Result<()>> {
            std::task::Poll::Ready(Ok(()))
        }

        fn poll_shutdown(
            self: std::pin::Pin<&mut Self>,
            _cx: &mut std::task::Context<'_>,
        ) -> std::task::Poll<std::io::Result<()>> {
            std::task::Poll::Ready(Ok(()))
        }
    }

    #[async_trait]
    impl LlmClient for ErrorLlm {
        async fn chat(
            &self,
            _messages: &[Message],
            _tools: &[ToolDef],
        ) -> motosan_agent_loop::Result<ChatOutput> {
            Err(motosan_agent_loop::AgentError::Llm("boom".into()))
        }
    }

    fn png_fixture(name: &str) -> std::path::PathBuf {
        use std::io::Write as _;

        let mut path = std::env::temp_dir();
        path.push(format!("capo-v06-{name}-{}.png", std::process::id()));
        let mut bytes = vec![0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A];
        bytes.extend_from_slice(&[0u8; 64]);
        std::fs::File::create(&path)
            .unwrap_or_else(|e| panic!("create fixture: {e}"))
            .write_all(&bytes)
            .unwrap_or_else(|e| panic!("write fixture: {e}"));
        path
    }

    #[tokio::test]
    async fn json_mode_with_image_emits_well_formed_stream_and_forwards_multipart() {
        use capo_agent::test_support::ScriptedLlm;

        let path = png_fixture("e2e-json");
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let llm = Arc::new(ScriptedLlm::builder().then_message("acknowledged").build());

        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(llm.clone())
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build app: {err}"),
        };

        let msg = UserMessage {
            text: "describe".into(),
            attachments: vec![capo_agent::Attachment::Image { path }],
        };

        let mut out: Vec<u8> = Vec::new();
        run_json_mode(&app, msg, dir.path(), &mut out)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let stream = String::from_utf8(out).unwrap_or_else(|e| panic!("utf8 stream: {e}"));
        assert!(
            stream.contains("\"type\":\"session\""),
            "stream missing session header: {stream}"
        );
        assert!(
            stream.contains("\"type\":\"agent_message_complete\""),
            "stream missing message-complete: {stream}"
        );

        let captured = llm.captured_inputs();
        let last_call = captured.last().unwrap_or_else(|| panic!("captured input"));
        let user_msg = last_call
            .iter()
            .rev()
            .find(|m| matches!(m, Message::User { .. }))
            .unwrap_or_else(|| panic!("user message"));
        let parts = match user_msg {
            Message::User { content, .. } => content.as_slice(),
            _ => unreachable!(),
        };
        assert!(matches!(&parts[0], ContentPart::Text { text, .. } if text == "describe"));
        assert!(parts.iter().any(|p| matches!(p, ContentPart::Image { .. })));
    }

    #[tokio::test]
    async fn json_mode_headless_deny_blocks_bash_without_prompting() {
        use capo_agent::test_support::ScriptedLlm;

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());

        // Script: agent calls bash, sees the (denied) tool result, finishes.
        let llm = ScriptedLlm::builder()
            .then_tool_call(
                "tc-1",
                "bash",
                serde_json::json!({"command": "echo should-not-run"}),
            )
            .then_message("done")
            .build();

        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(llm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };

        let mut buf: Vec<u8> = Vec::new();
        run_json_mode(&app, UserMessage::text("do it"), dir.path(), &mut buf)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        let lines: Vec<&str> = text.lines().collect();

        // No interactive prompt should appear in headless mode.
        assert!(
            !lines
                .iter()
                .any(|l| l.contains(r#""type":"permission_requested""#)),
            "headless mode must not emit permission_requested: {lines:?}"
        );
        // The bash call comes back as a tool error citing non-interactive denial.
        let denial = lines
            .iter()
            .find(|l| l.contains(r#""type":"tool_call_completed""#))
            .unwrap_or_else(|| panic!("no tool_call_completed: {lines:?}"));
        assert!(
            denial.contains(r#""is_error":true"#),
            "tool should be denied: {denial}"
        );
        assert!(
            denial.contains("non-interactive") || denial.contains("Permission denied"),
            "denial should cite the headless rule: {denial}"
        );
        // Agent saw the failure and finished its scripted reply.
        assert!(
            lines
                .iter()
                .any(|l| l.contains(r#""type":"agent_message_complete""#) && l.contains("done")),
            "agent should complete with 'done': {lines:?}"
        );
    }

    #[tokio::test]
    async fn json_mode_emits_permission_mode_after_session_header() {
        use capo_agent::test_support::ScriptedLlm;

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let llm = ScriptedLlm::builder().then_message("done").build();
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(llm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };
        app.set_permission_mode(capo_agent::PermissionMode::AcceptEdits)
            .await;

        let mut buf: Vec<u8> = Vec::new();
        run_json_mode(&app, UserMessage::text("do it"), dir.path(), &mut buf)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        let mut lines = text.lines();
        let session = lines.next().unwrap_or_else(|| panic!("missing session"));
        assert!(session.contains(r#""type":"session""#), "{session}");
        let mode = lines.next().unwrap_or_else(|| panic!("missing mode"));
        assert!(
            mode.contains(r#""type":"permission_mode_changed""#) && mode.contains("accept-edits"),
            "second line should be mode event: {mode}\nfull stream: {text}"
        );
    }

    #[tokio::test]
    async fn json_mode_emits_settings_snapshot_after_session_header() {
        use capo_agent::test_support::ScriptedLlm;

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let llm = ScriptedLlm::builder().then_message("done").build();
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(llm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };

        let mut buf: Vec<u8> = Vec::new();
        run_json_mode(&app, UserMessage::text("do it"), dir.path(), &mut buf)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        let mut lines = text.lines();
        let session = lines.next().unwrap_or_else(|| panic!("missing session"));
        assert!(
            session.contains(r#""type":"session""#),
            "session header: {session}"
        );
        let mode = lines.next().unwrap_or_else(|| panic!("missing mode"));
        assert!(
            mode.contains(r#""type":"permission_mode_changed""#),
            "mode event: {mode}"
        );
        let settings = lines
            .next()
            .unwrap_or_else(|| panic!("missing settings_snapshot"));
        assert!(
            settings.contains(r#""type":"settings_snapshot""#) && settings.contains(r#""model""#),
            "expected settings_snapshot as third line: {settings}\nfull stream: {text}"
        );
    }

    #[tokio::test]
    async fn json_mode_dangerously_allow_all_runs_bash_unchecked() {
        use capo_agent::test_support::ScriptedLlm;

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());

        let llm = ScriptedLlm::builder()
            .then_tool_call(
                "tc-1",
                "bash",
                serde_json::json!({"command": "echo allow-all-marker"}),
            )
            .then_message("done")
            .build();

        // Binary bypass still installs PermissionExtension so invariant
        // hard-blocks are enforced; set_permission_mode switches it to AllowAll.
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(llm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };
        app.set_permission_mode(capo_agent::PermissionMode::Bypass)
            .await;

        let mut buf: Vec<u8> = Vec::new();
        run_json_mode(&app, UserMessage::text("do it"), dir.path(), &mut buf)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        let lines: Vec<&str> = text.lines().collect();

        assert!(
            !lines
                .iter()
                .any(|l| l.contains(r#""type":"permission_requested""#)),
            "allow-all must never prompt: {lines:?}"
        );
        // tool_call_completed must NOT be an error.
        let completed = lines
            .iter()
            .find(|l| l.contains(r#""type":"tool_call_completed""#))
            .unwrap_or_else(|| panic!("no tool_call_completed: {lines:?}"));
        assert!(
            !completed.contains(r#""is_error":true"#),
            "bash should have run unchecked: {completed}"
        );
        // No top-level error event either.
        assert!(
            !lines.iter().any(|l| l.contains(r#""type":"error""#)),
            "no error events expected: {lines:?}"
        );
    }

    #[tokio::test]
    async fn json_mode_bypass_still_hard_denies_envrc_write() {
        use capo_agent::test_support::ScriptedLlm;

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let target = dir.path().join(".envrc");

        let llm = ScriptedLlm::builder()
            .then_tool_call(
                "tc-1",
                "write",
                serde_json::json!({"path": target.display().to_string(), "content": "SECRET=1"}),
            )
            .then_message("done")
            .build();

        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_llm(Arc::new(llm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };
        app.set_permission_mode(capo_agent::PermissionMode::Bypass)
            .await;

        let mut buf: Vec<u8> = Vec::new();
        run_json_mode(&app, UserMessage::text("do it"), dir.path(), &mut buf)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        let lines: Vec<&str> = text.lines().collect();
        let denial = lines
            .iter()
            .find(|l| l.contains(r#""type":"tool_call_completed""#))
            .unwrap_or_else(|| panic!("no tool_call_completed: {lines:?}"));
        assert!(
            denial.contains(r#""is_error":true"#) && denial.contains("blocked path"),
            "bypass must hard-deny .envrc writes: {denial}"
        );
        assert!(!target.exists(), "hard-denied .envrc should not be written");
    }

    #[tokio::test]
    async fn write_to_blocked_path_is_hard_denied() {
        use capo_agent::test_support::ScriptedLlm;

        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());

        // Default Policy blocks nothing; declare the hard-block explicitly.
        // Schema matches the policy.rs test fixture.
        let perms_path = dir.path().join("permissions.toml");
        std::fs::write(&perms_path, "[write]\nblocked_paths = [\".git/**\"]\n")
            .unwrap_or_else(|e| panic!("write permissions.toml: {e}"));

        let target = dir.path().join(".git/leak");
        let llm = ScriptedLlm::builder()
            .then_tool_call(
                "tc-1",
                "write",
                serde_json::json!({"path": target.display().to_string(), "content": "x"}),
            )
            .then_message("done")
            .build();

        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_builtin_tools()
            .with_permissions_config(perms_path)
            .with_llm(Arc::new(llm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };

        let mut buf: Vec<u8> = Vec::new();
        run_json_mode(&app, UserMessage::text("do it"), dir.path(), &mut buf)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        let lines: Vec<&str> = text.lines().collect();

        let denial = lines
            .iter()
            .find(|l| l.contains(r#""type":"tool_call_completed""#))
            .unwrap_or_else(|| panic!("no tool_call_completed: {lines:?}"));
        assert!(
            denial.contains(r#""is_error":true"#),
            "blocked-path write should be denied: {denial}"
        );
        // The step-1 message is "Permission denied: <abs path> is in a blocked path"
        // (see PermissionExtension::decide(), extension.rs:60). Assert on a
        // discriminator that proves it's the step-1 path (not the headless message
        // "non-interactive: tool requires approval").
        assert!(
            denial.contains("blocked path"),
            "denial should cite step-1 blocked-path, not the headless message: {denial}"
        );
    }

    #[tokio::test]
    async fn json_mode_returns_error_when_stream_emits_error_event() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(ErrorLlm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };

        let mut buf: Vec<u8> = Vec::new();
        let err = match run_json_mode(&app, UserMessage::text("hello"), dir.path(), &mut buf).await
        {
            Ok(()) => panic!("json mode should fail after an error event"),
            Err(err) => err,
        };
        assert!(format!("{err}").contains("agent error"), "{err}");

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        assert!(text.contains(r#""type":"error""#), "{text}");
    }

    #[tokio::test]
    async fn json_mode_emits_header_then_jsonl_events() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .with_headless_permissions()
            .build()
            .await
        {
            Ok(app) => app,
            Err(err) => panic!("build: {err}"),
        };

        let mut buf: Vec<u8> = Vec::new();
        run_json_mode(&app, UserMessage::text("hello"), dir.path(), &mut buf)
            .await
            .unwrap_or_else(|e| panic!("run_json_mode: {e}"));

        let text = String::from_utf8(buf).unwrap_or_else(|e| panic!("utf8: {e}"));
        let lines: Vec<&str> = text.lines().collect();
        assert!(!lines.is_empty(), "no output");
        // First line is the session header.
        assert!(
            lines[0].contains(r#""type":"session""#),
            "header: {}",
            lines[0]
        );
        assert!(lines[0].contains(r#""protocol_version":1"#));
        // Every line is valid JSON.
        for l in &lines {
            serde_json::from_str::<serde_json::Value>(l)
                .unwrap_or_else(|e| panic!("bad json line {l:?}: {e}"));
        }
    }

    #[tokio::test]
    async fn command_forwarder_refreshes_branch_tree_after_successful_compact() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("sessions"),
        ));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .with_session_store(store)
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        for i in 0..4 {
            let _: Vec<_> = app
                .send_user_message(UserMessage::text(format!("turn {i}")))
                .collect()
                .await;
        }

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx.send(Command::Compact).await.is_err() {
            panic!("command channel closed");
        }

        let mut saw_compacted = false;
        let mut saw_branch_tree = false;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await
        {
            match event {
                UiEvent::Compacted {
                    turns_removed,
                    source,
                    ..
                } => {
                    saw_compacted = true;
                    assert_eq!(turns_removed, 0);
                    assert_eq!(source, capo_agent::CompactSource::Manual);
                }
                UiEvent::BranchTree(_) => {
                    saw_branch_tree = true;
                    break;
                }
                _ => {}
            }
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), forwarder).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => panic!("forwarder join failed: {err}"),
            Err(_) => panic!("forwarder did not exit"),
        }
        assert!(saw_compacted, "compact should emit Compacted");
        assert!(saw_branch_tree, "compact should refresh the branch tree");
    }

    #[tokio::test]
    async fn command_forwarder_clone_switches_session_and_emits_replaced() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let store: Arc<dyn SessionStore> = Arc::new(motosan_agent_loop::FileSessionStore::new(
            dir.path().join("sessions"),
        ));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .with_session_store(store)
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };
        let _: Vec<_> = app
            .send_user_message(UserMessage::text("hello"))
            .collect()
            .await;
        let original_id = app.session_id();

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let app_for_check = Arc::clone(&app);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx.send(Command::CloneSession).await.is_err() {
            panic!("command channel closed");
        }

        let mut saw_replaced = false;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await
        {
            if matches!(event, UiEvent::SessionReplaced { .. }) {
                saw_replaced = true;
                break;
            }
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), forwarder).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => panic!("forwarder join failed: {err}"),
            Err(_) => panic!("forwarder did not exit"),
        }
        assert!(saw_replaced, "clone should emit SessionReplaced");
        assert_ne!(
            app_for_check.session_id(),
            original_id,
            "clone should switch the live session to the copy"
        );
    }

    #[tokio::test]
    async fn command_forwarder_reports_compact_errors_once() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(DoneLlm))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        for i in 0..4 {
            let _: Vec<_> = app
                .send_user_message(UserMessage::text(format!("turn {i}")))
                .collect()
                .await;
        }

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx.send(Command::Compact).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await {
            Ok(Some(UiEvent::Error(message))) => {
                assert_eq!(message.matches("compaction failed").count(), 1, "{message}");
            }
            Ok(Some(other)) => panic!("expected compact error, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for compact error"),
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), forwarder).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => panic!("forwarder join failed: {err}"),
            Err(_) => panic!("forwarder did not exit"),
        }
    }

    #[tokio::test]
    async fn command_forwarder_ignores_second_send_while_turn_is_running() {
        let dir = tempfile::tempdir().unwrap_or_else(|e| panic!("tempdir: {e}"));
        let mut cfg = Config::default();
        cfg.anthropic.api_key = Some("sk-unused".into());
        let app = match AppBuilder::new()
            .with_config(cfg)
            .with_cwd(dir.path())
            .with_llm(Arc::new(PendingLlm))
            .build()
            .await
        {
            Ok(app) => Arc::new(app),
            Err(err) => panic!("build: {err}"),
        };

        let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel::<UiEvent>(16);
        let (cmd_tx, cmd_rx) = tokio::sync::mpsc::channel::<Command>(16);
        let forwarder = tokio::spawn(forward_commands(app, ui_tx, cmd_rx));

        if cmd_tx
            .send(Command::SendUserMessage {
                text: "first".into(),
                attachments: Vec::new(),
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), ui_rx.recv()).await {
            Ok(Some(UiEvent::AgentTurnStarted)) => {}
            Ok(Some(other)) => panic!("expected turn start, got {other:?}"),
            Ok(None) => panic!("ui channel closed"),
            Err(_) => panic!("timed out waiting for first turn"),
        }

        if cmd_tx
            .send(Command::SendUserMessage {
                text: "second".into(),
                attachments: Vec::new(),
            })
            .await
            .is_err()
        {
            panic!("command channel closed");
        }

        let mut saw_concurrency_error = false;
        let mut extra_turn_starts = 0;
        while let Ok(Some(event)) =
            tokio::time::timeout(Duration::from_millis(25), ui_rx.recv()).await
        {
            match event {
                UiEvent::Error(message) if message.contains("another turn") => {
                    saw_concurrency_error = true;
                }
                UiEvent::AgentTurnStarted => extra_turn_starts += 1,
                _ => {}
            }
        }

        if cmd_tx.send(Command::Quit).await.is_err() {
            panic!("command channel closed");
        }
        match tokio::time::timeout(Duration::from_millis(200), forwarder).await {
            Ok(Ok(())) => {}
            Ok(Err(err)) => panic!("forwarder join failed: {err}"),
            Err(_) => panic!("forwarder did not exit"),
        }
        assert!(!saw_concurrency_error, "second send leaked to App");
        assert_eq!(extra_turn_starts, 0, "second send started another turn");
    }
}
