rpi_cli/

session.rs

//! Harness construction + session-storage wiring. Mirrors the Rust-side
//! equivalent of the TS `packages/coding-agent/src/core/sdk.ts`
//! (`createAgentSession`) — build the env, tools, durable session storage, and
//! `AgentHarnessOptions`, then `AgentHarness::create`.
//!
//! v1 scope cuts vs the TS SDK (tracked in `docs/m6-cli-open-questions.md`):
//! - **Skill / prompt-template / context-file discovery IS wired**
//!   (`--no-skills`/`-ns`, `--no-prompt-templates`/`-np`, `--no-context-files`/
//!   `-nc` each suppress one channel; project `.rpi/<sub>` + legacy
//!   `.rpi/<sub>` + global
//!   `agent_dir()<sub>` discovery with project-wins dedupe via
//!   [`crate::resource_dirs`]; SYSTEM.md/APPEND_SYSTEM.md project-wins
//!   precedence). **Extension `resources_discover` (B5b) feeds the SAME loaders:
//!   a plugin's discovered skill/prompt paths merge with the static dirs and
//!   re-run through `load_skills`/`load_prompt_templates` (individual `.md` files
//!   load too — `load_skills` accepts both dirs and files). Package themes are
//!   parsed by the TUI when selected via settings or `--theme`.**
//!   Project resources load by default without a prompt. `--no-approve` or a
//!   stored negative `trust.json` decision disables them explicitly.
//! - **No `ModelRuntime`/multi-provider registry.** The resolver supports the
//!   built-in Anthropic/OpenAI-compatible providers and `models.json`, but
//!   runtime catalog mutation remains outside this layer.
//! - **Built-in tools**: `read`, `bash`, `edit`, `write`, and the read-only
//!   `docs` lookup tool. The former rpi-only `grep`, `find`, `ls`, and
//!   `powershell` tools remain library modules but are not registered by the
//!   CLI.
//! - **Session restore (`-c`/`-r`/`--session`)** is *partially* supported: a
//!   fresh session is always created. The harness's `create` rejects sessions
//!   that already have records unless `allow_existing_session` is enabled.
//!   The interactive `-c`/`-r`/`--session` paths enable that mode and replay
//!   the existing branch before appending new messages. See [`SessionSelection`].

mod backend;

use std::collections::{BTreeMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex, OnceLock};

use rpi_ai::Provider;
use rpi_harness::agent_harness::AgentHarness;
use rpi_harness::context_files::{format_project_context, load_project_context_files};
use rpi_harness::session::memory::{InMemorySessionStorage, SystemClock};
use rpi_harness::session::session::DefaultIdGenerator;
use rpi_harness::session::types::SessionMetadata;
use rpi_harness::session::Session;
use rpi_harness::system_prompt::compose_system_prompt;
use rpi_harness::types::{
    AgentHarnessOptions, AgentHarnessResources, AgentHarnessStreamOptions, DrivingMode,
    HarnessTool, HarnessToolExecution, ToolReplay,
};
#[cfg(windows)]
use rpi_tools::create_powershell_tool;
use rpi_tools::{
    create_bash_tool, create_edit_tool, create_read_tool, create_write_tool, ExecutionToolContext,
    MutationQueueRegistry, OsExecutionEnv,
};

use crate::args::Args;
use crate::docs_tool::create_docs_tool;
use crate::provider::ResolvedModel;
use crate::resource_dirs::{
    discover_append_system_prompt_file, discover_system_prompt_file, extension_dirs,
    global_extension_dirs, global_prompt_template_dirs, global_skill_dirs,
    load_prompt_templates_with_precedence, load_skills_with_precedence, project_extension_dirs,
    project_prompt_template_dirs, project_skill_dirs, prompt_template_dirs, skill_dirs,
};
use crate::startup_diagnostics::{Phase, StartupSpan};
use rpi_extensions::{
    emit_resources_discover, ExtensionEmitter, ExtensionSession, NullDiagnostics,
    PluginDiagnostics, PluginToolAdapter, TeeEmitter, ToolPromptMetadata,
};

// Preserve the existing public modes::rpc signature. This CLI-only attachment
// table holds no session/router alive and keys by storage allocation identity,
// NOT by a durable ID (two independent restored hosts must never share routing).
// The session/tools/ReloadContext own the actual router. RPC owns its client.
// Reload MUST reuse ReloadContext::file_io, including a disconnected selected
// client backend; rebuilding a default router would silently restore host file
// authority. Missing attachments reject configuration (never infer a router or
// install host-backed replacements). This compatibility bridge can be replaced
// by an explicit neutral harness attachment when its public API is revised.
type FileIOAttachment = (
    std::sync::Weak<dyn rpi_harness::session::types::SessionStorage>,
    std::sync::Weak<rpi_tools::file_io::FileIORouter>,
);
fn file_io_attachments() -> &'static Mutex<Vec<FileIOAttachment>> {
    static ATTACHMENTS: std::sync::OnceLock<Mutex<Vec<FileIOAttachment>>> =
        std::sync::OnceLock::new();
    ATTACHMENTS.get_or_init(|| Mutex::new(Vec::new()))
}
pub(crate) fn register_file_io_router(
    harness: &AgentHarness,
    router: &Arc<rpi_tools::file_io::FileIORouter>,
) {
    let storage = harness.session().storage();
    let mut attachments = file_io_attachments()
        .lock()
        .expect("file I/O attachments poisoned");
    attachments.retain(|(s, r)| {
        s.strong_count() > 0
            && r.strong_count() > 0
            && !s
                .upgrade()
                .is_some_and(|known| Arc::ptr_eq(&known, &storage))
    });
    attachments.push((Arc::downgrade(&storage), Arc::downgrade(router)));
}
pub(crate) fn file_io_router(
    harness: &AgentHarness,
) -> Option<Arc<rpi_tools::file_io::FileIORouter>> {
    let storage = harness.session().storage();
    let mut attachments = file_io_attachments()
        .lock()
        .expect("file I/O attachments poisoned");
    attachments.retain(|(s, r)| s.strong_count() > 0 && r.strong_count() > 0);
    attachments.iter().find_map(|(s, r)| {
        s.upgrade()
            .filter(|known| Arc::ptr_eq(known, &storage))
            .and_then(|_| r.upgrade())
    })
}

/// The coding tools registered by the CLI by default.
pub const BUILTIN_TOOL_NAMES: &[&str] = &["read", "bash", "edit", "write", "docs"];

/// The default coding system prompt for the CLI's built-in tools.
///
/// This retains the two-argument API for library callers. Session startup and
/// reload use [`default_system_prompt_with_tool_prompts`] with the actual active
/// tools and their winning extension metadata.
pub fn default_system_prompt(cwd: &str, prior_reasoning_replayed: bool) -> String {
    let tools = BUILTIN_TOOL_NAMES
        .iter()
        .map(|name| (*name).to_owned())
        .collect::<Vec<_>>();
    default_system_prompt_with_tools(cwd, prior_reasoning_replayed, &tools)
}

/// A coding system prompt for the selected tools. A condensed port of the TS
/// `packages/coding-agent/src/core/system-prompt.ts` base prompt.
///
/// `tools` is the active tool set after `--tools`/`--exclude-tools` and
/// extension merging. Like pi, the text list includes only tools with an
/// explicit prompt snippet; other active tools remain available through their
/// provider-facing schemas. Tool-specific guidelines still follow the full
/// active set, so restricted runs get no guidance for unavailable tools.
pub fn default_system_prompt_with_tools(
    cwd: &str,
    prior_reasoning_replayed: bool,
    tools: &[String],
) -> String {
    default_system_prompt_with_tool_prompts(cwd, prior_reasoning_replayed, tools, &BTreeMap::new())
}

/// Render pi-style explicit prompt contributions for the active tool set.
/// Entries override built-in metadata, including an empty entry for an
/// extension that replaces a built-in without contributing a prompt snippet.
/// Inactive metadata is ignored. Schemas and callable tools are not modified.
pub fn default_system_prompt_with_tool_prompts(
    cwd: &str,
    prior_reasoning_replayed: bool,
    tools: &[String],
    overrides: &BTreeMap<String, ToolPromptMetadata>,
) -> String {
    let metadata: BTreeMap<_, _> = tools
        .iter()
        .map(|name| {
            let prompt = overrides
                .get(name)
                .cloned()
                .unwrap_or_else(|| ToolPromptMetadata {
                    prompt_snippet: builtin_tool_snippet(name).map(str::to_owned),
                    prompt_guidelines: Vec::new(),
                })
                .normalized();
            (name.as_str(), prompt)
        })
        .collect();
    // The working-state rule must match what actually comes back on the next
    // request. With `compat.requiresThinkingAsText` (or the native thinking
    // channels of anthropic/responses) the model *does* see its own reasoning
    // again, and claiming otherwise makes it re-derive and restate the plan
    // every turn — the re-plan symptom in `docs/llm-repetition-forensics.md`.
    let working_state_rule = if prior_reasoning_replayed {
        "- Your reasoning is carried back to you on the next turn: do not re-derive it or restate
  the plan. Continue from the last unfinished step, and when a step lands, say which one.
  Re-stating the same plan without acting on it is a bug, not progress."
    } else {
        "- Your reasoning is not carried into your next turn; only your visible replies and tool
  output come back. Before each batch of tool calls, write one short line naming the step you
  are on and what remains, and say which step is done when it lands.
  Re-stating the same plan without acting on it is a bug, not progress."
    };

    // The text list is a selective summary, not the tool registry. In
    // particular, do not repeat extension tool names without a prompt snippet.
    let tool_lines = tools
        .iter()
        .filter_map(|name| {
            metadata
                .get(name.as_str())?
                .prompt_snippet
                .as_ref()
                .map(|snippet| format!("- {name:<5} — {snippet}"))
        })
        .collect::<Vec<_>>()
        .join("\n");
    let tool_lines = if tool_lines.is_empty() {
        "(none)".to_string()
    } else {
        tool_lines
    };

    // Tool-specific rules (Bash on Windows, docs before guessing) live in the
    // tool's own line or description, not here a second time.
    let guidelines = [
        "- Reply in the same language as the user's latest message (the conversation's language when
  that message is too short to tell), unless the user asks for another.",
        "- Be concise in your responses",
        "- Show file paths clearly when working with files",
        "- Prefer the smallest change that solves the problem",
        working_state_rule,
        "- Read only what you need. If a file or search result was already shown earlier in this
  conversation, use it instead of fetching it again.",
    ];
    // Tools contribute rules even without snippets. Like pi, trim and dedupe
    // across the full active set and the core rules, preserving first occurrence.
    let mut seen = HashSet::new();
    let tool_rules = tools
        .iter()
        .flat_map(|name| &metadata[name.as_str()].prompt_guidelines)
        .map(String::as_str);
    let core_rules = guidelines
        .iter()
        .map(|rule| rule.strip_prefix("- ").unwrap_or(rule));
    let guidelines = tool_rules
        .chain(core_rules)
        .map(str::trim)
        .filter(|rule| !rule.is_empty() && seen.insert(*rule))
        .map(|rule| format!("- {rule}"))
        .collect::<Vec<_>>()
        .join("\n");

    format!(
        "You are an expert coding assistant operating inside rpi, a coding agent harness. \
You help users by reading files, executing commands, editing code, and writing new files.

Available tools:
{tool_lines}

In addition to the tools above, you may have access to other custom tools depending on the project.

Guidelines:
{guidelines}

Current working directory: {cwd}"
    )
}

/// Take metadata from the same winning extension definitions used for tool
/// merging, including empty metadata for overrides. Called on startup/reload.
fn extension_tool_prompt_metadata(
    session: &ExtensionSession,
) -> BTreeMap<String, ToolPromptMetadata> {
    session
        .snapshot()
        .map(|snapshot| {
            snapshot
                .tools()
                .iter()
                .map(|tool| (tool.tool.name.clone(), tool.prompt.clone()))
                .collect()
        })
        .unwrap_or_default()
}

/// One-line prompt descriptions of the tools rpi itself registers.
fn builtin_tool_snippet(name: &str) -> Option<&'static str> {
    Some(match name {
        "read" => "Read file contents",
        "bash" => "Execute Bash commands (Bash/POSIX syntax, also on Windows)",
        "edit" => "Find/replace edits to existing files",
        "write" => "Create or overwrite files",
        "docs" => "Look up rpi commands, extensions, Pi package compatibility and .rpi configuration instead of guessing",
        "grep" => "Search file contents for a pattern",
        "find" => "Find files by glob pattern",
        "ls" => "List directory contents",
        "powershell" => "Execute PowerShell commands on Windows",
        _ => return None,
    })
}

/// How the user asked to select a session. v1 honors `NoSession` (ephemeral
/// `InMemorySessionStorage`), `New` (a fresh JSONL file), and — new this pass —
/// `Latest` / `ById`, which **restore** an existing JSONL session on launch
/// (`--continue`/`-c`, `--resume`/`-r`, `--session <id|path>`). The restored
/// transcript renders into the TUI on startup and the run continues appending
/// to the same file.
#[derive(Debug, Clone)]
pub enum SessionSelection {
    /// `--no-session`: ephemeral, in-memory, nothing persisted.
    Ephemeral,
    /// Fresh durable JSONL session under `--session-dir` (or the default dir).
    New { dir: PathBuf, name: Option<String> },
    /// Explicit fresh print-only experimental backend; never a restore selection.
    ExperimentalNative { dir: PathBuf },
    /// `-c` / `-r`: restore the most recent session in the default dir.
    Latest { dir: PathBuf },
    /// `--session <id|path>`: restore the session whose id matches, or whose
    /// file name contains the id.
    ById { id: String, dir: PathBuf },
    /// `--session-id <id>`: use the EXACT session id, creating it if missing.
    ByExactId { id: String, dir: PathBuf },
    /// `--fork <path|id>`: fork the given session into a new one and start in
    /// the fork.
    Fork { source: String, dir: PathBuf },
}

/// Decide the session selection from parsed args + the resolved cwd.
pub fn select_session(args: &Args, cwd: &Path) -> SessionSelection {
    select_session_in_dir(args, configured_session_dir(args, cwd))
}

/// Resolve the root once at startup; runtime actions retain this selection.
pub(crate) fn configured_session_dir(args: &Args, cwd: &Path) -> PathBuf {
    if let Some(root) = &args.experimental_native_store {
        return if root.is_absolute() {
            root.clone()
        } else {
            cwd.join(root)
        };
    }
    let dir = args
        .session_dir
        .clone()
        .or_else(|| {
            crate::settings::load_settings()
                .ok()
                .and_then(|settings| settings.session_dir)
        })
        .unwrap_or_else(|| default_session_dir(cwd));
    // Relative roots belong to the launch project, even if an extension later
    // changes the process working directory.
    if dir.is_relative() && !dir.has_root() {
        cwd.join(dir)
    } else {
        dir
    }
}

pub(crate) fn select_session_in_dir(args: &Args, dir: PathBuf) -> SessionSelection {
    if args.experimental_native_store.is_some() {
        return SessionSelection::ExperimentalNative { dir };
    }
    if args.no_session {
        return SessionSelection::Ephemeral;
    }
    // `--session-dir` wins; then the saved `sessionDir` (upstream's key, same
    // format as the flag); then the built-in default. Keep this directory on
    // restore selections too, not only on fresh sessions.
    if args.continue_session || args.resume {
        return SessionSelection::Latest { dir };
    }
    if let Some(s) = &args.fork {
        return SessionSelection::Fork {
            source: s.clone(),
            dir,
        };
    }
    if let Some(s) = &args.session_id {
        return SessionSelection::ByExactId { id: s.clone(), dir };
    }
    if let Some(s) = &args.session {
        return SessionSelection::ById { id: s.clone(), dir };
    }
    SessionSelection::New {
        dir,
        name: args.name.clone(),
    }
}

/// The default session directory is `<cwd>/.rpi/sessions`, unless
/// `RPI_CODING_AGENT_SESSION_DIR` is set (in which case its value is used).
pub fn default_session_dir(cwd: &Path) -> PathBuf {
    std::env::var("RPI_CODING_AGENT_SESSION_DIR")
        .ok()
        .map(PathBuf::from)
        .unwrap_or_else(|| cwd.join(".rpi").join("sessions"))
}

/// Extensions are loaded exactly once before model resolution. Their action
/// host remains unbound until the harness and authenticated catalog exist.
pub struct PreparedExtensions {
    resource_leases: Arc<Vec<crate::resource_management::Lease>>,
    resource_selection: Option<crate::resource_selection::ResolvedSelection>,
    session_selection: SessionSelection,
    transition_session: Option<Session>,
    pub extension_session: ExtensionSession,
    native_store: Option<PathBuf>,
    session_dir: PathBuf,
    runtime: tokio::runtime::Handle,
    tool_context: rpi_extensions::ToolCallContext,
    action_host: Arc<crate::extensions_actions::HarnessActionHost>,
    harness_cell: Arc<std::sync::OnceLock<rpi_harness::agent_harness::WeakAgentHarness>>,
    reload_mailbox: rpi_extensions::ReloadMailbox,
    /// Request channel for `runtime_action(ReloadModels)` (action id 31).
    reload_models_mailbox: rpi_extensions::ReloadModelsMailbox,
    ui_dialog_mailbox: rpi_extensions::UiDialogMailbox,
    ext_status_mailbox: rpi_extensions::ExtensionStatusMailbox,
    editor_text_mailbox: rpi_extensions::EditorTextMailbox,
    action_bridge: Arc<rpi_extensions::ActionBridge>,
}

/// Validate native durable resource ownership BEFORE loading any executable code.
/// The synchronous public prepare_extensions API is retained: a scoped reader
/// thread uses its own runtime so this also works inside current-thread Tokio.
/// It does no registration, provider callbacks, session creation, or lock claiming.
pub(crate) fn preflight_extension_startup(
    args: &Args,
    cwd: &Path,
    trusted: bool,
) -> Result<
    (
        Option<crate::resource_selection::ResolvedSelection>,
        SessionSelection,
    ),
    BuildError,
> {
    let mut resources = crate::resource_selection::ResolvedSelection::load(args, cwd, trusted)
        .map_err(BuildError::HarnessCreate)?;
    let selection = select_session(args, cwd);
    let pinned = std::thread::scope(|scope| {
        scope
            .spawn(|| {
                tokio::runtime::Builder::new_current_thread()
                    .enable_all()
                    .build()
                    .map_err(|e| BuildError::SessionOpen(e.to_string()))?
                    .block_on(async {
                        if let Some(selected) = &mut resources {
                            let env: Arc<dyn rpi_tools::ExecutionEnv> =
                                Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf()));
                            selected
                                .validate_skills(&env)
                                .await
                                .map_err(BuildError::HarnessCreate)?;
                        }
                        let cwd_str = cwd.to_string_lossy();
                        let target = match &selection {
                            SessionSelection::Fork { source, dir } => {
                                match session_metadata_at(Path::new(source)) {
                                    Some(meta) => Some(meta),
                                    None => find_session_metadata(source, &cwd_str, dir).await?,
                                }
                            }
                            _ => resolve_restore_target(&selection, &cwd_str).await?,
                        };
                        let Some(meta) = target else {
                            if resources.as_ref().is_some_and(|r| r.transition.is_some()) {
                                return Err(BuildError::SessionOpen("resource transition requires an existing native session".into()));
                            }
                            return Ok(selection.clone());
                        };
                        if resources.as_ref().is_some_and(|r| r.transition.is_some()) && matches!(selection, SessionSelection::Fork { .. }) {
                            return Err(BuildError::SessionOpen("resource transition cannot fork".into()));
                        }
                        use rpi_harness::session::jsonl::{
                            JsonlSessionRepo, JsonlSessionRepoOptions,
                        };
                        let dir = selection_dir(&selection);
                        let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
                            fs: Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())),
                            sessions_root: dir.to_string_lossy().into_owned(),
                            clock: Arc::new(SystemClock),
                            ids: Arc::new(DefaultIdGenerator::new()),
                        });
                        let binding_facts = StartupSpan::start(Phase::BindingFacts);
                        let facts = if resources.as_ref().is_some_and(|r| r.transition.is_some()) {
                            rpi_harness::session::jsonl::JsonlSessionStorage::read_resource_transition_facts(
                                Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())), &meta.path,
                                Arc::new(SystemClock), Arc::new(DefaultIdGenerator::new()),
                            ).await
                        } else { repo.read_facts_by_jsonl_metadata(&meta).await }
                            .map_err(|e| BuildError::SessionOpen(e.to_string()))?;
                        crate::resource_selection::validate_session_binding(
                            &facts.values,
                            resources.as_ref(),
                        )
                        .map_err(BuildError::SessionOpen)?;
                        binding_facts.finish();
                        Ok(match selection {
                            SessionSelection::Fork { .. } => SessionSelection::Fork {
                                source: meta.path,
                                dir,
                            },
                            _ => SessionSelection::ByExactId { id: meta.id, dir },
                        })
                    })
            })
            .join()
            .map_err(|_| BuildError::SessionOpen("resource preflight reader panicked".into()))?
    })?;
    Ok((resources, pinned))
}

pub fn prepare_extensions(
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
) -> Result<PreparedExtensions, BuildError> {
    let native_store =
        crate::native_store::root(args, cwd).map_err(BuildError::NativeStorePolicy)?;
    let resource_leases =
        crate::resource_management::host_leases(cwd).map_err(BuildError::HarnessCreate)?;
    let (resource_selection, session_selection) =
        preflight_extension_startup(args, cwd, project_trusted)?;
    // Transition hosts own the writer before any DLL registration. The strict
    // loader cannot repair history/progress or migrate metadata on rejection.
    let transition_session = if resource_selection
        .as_ref()
        .is_some_and(|r| r.transition.is_some())
    {
        Some(std::thread::scope(|scope| {
            scope.spawn(|| {
            tokio::runtime::Builder::new_current_thread().enable_all().build()
                .map_err(|e| BuildError::SessionOpen(e.to_string()))?.block_on(async {
                    let meta = resolve_restore_target(&session_selection, &cwd.to_string_lossy()).await?
                        .ok_or_else(|| BuildError::SessionOpen("transition session disappeared".into()))?;
                    let path = std::fs::canonicalize(&meta.path).map_err(|e| BuildError::SessionOpen(e.to_string()))?;
                    claim_session_lock_inner(&path.to_string_lossy(), false).map_err(BuildError::SessionOpen)?;
                    let storage = rpi_harness::session::jsonl::JsonlSessionStorage::load_for_resource_transition(
                        Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())), &path.to_string_lossy(),
                        Arc::new(SystemClock), Arc::new(DefaultIdGenerator::new()),
                    ).await.map_err(|e| BuildError::SessionOpen(e.to_string()))?;
                    let session = Session::new(Arc::new(storage), None);
                    crate::resource_selection::bind_session(&session, resource_selection.as_ref()).await.map_err(BuildError::SessionOpen)?;
                    Ok::<_, BuildError>(session)
                })
        }).join().map_err(|_| BuildError::SessionOpen("transition writer preparation panicked".into()))?
        })?)
    } else {
        None
    };
    let session_dir = if matches!(session_selection, SessionSelection::Ephemeral) {
        configured_session_dir(args, cwd)
    } else {
        selection_dir(&session_selection)
    };
    let cwd_str = cwd.to_string_lossy().to_string();
    let tool_context = rpi_extensions::ToolCallContext::new(cwd_str.clone());
    if !project_trusted && args.verbose {
        eprintln!(
            "warning: current-project settings, resources, and discovered extensions are explicitly disabled (use --approve or /trust yes to re-enable)"
        );
    }
    // Pi packages are explicitly opt-in because discovery can start Node and
    // execute package code. An explicit project opt-out limits discovery to
    // global settings.

    // ---- B5a: build the action bridge BEFORE extension load ----
    // Extensions load before `AgentHarness::create` (extensions provide tools the
    // harness is built with), but a plugin stores the `ActionBridge`'s raw
    // `user_data` pointer during `register` and it must remain valid + the host
    // must be ready for the whole session. So:
    //  1. Capture the current tokio `Handle` (the async main-thread runtime) —
    //     the bridge spawns dispatch from any thread via `Handle::spawn`.
    //  2. Build an *empty* `HarnessActionHost` (its harness cell is unset; no
    //     plugin can call a runtime action before the harness runs).
    //  3. Wrap it as `Arc<dyn RuntimeActionHost>` + `ActionBridge`, thread
    //     `Some(bridge)` into `load_extensions` so every plugin's `user_data`
    //     points at this bridge.
    //  4. After `AgentHarness::create` succeeds, call `set_harness(&cell, …)` to
    //     fill the host cell the bridge recovers on the first action call.
    let runtime = tokio::runtime::Handle::try_current().map_err(|e| {
        BuildError::HarnessCreate(format!("no tokio runtime for action bridge: {e}"))
    })?;
    let catalog = Vec::new();
    // Request channel for `runtime_action(ReloadModels)` (action id 31).
    // Created before the host so the host can carry it; the TUI installs its
    // sender once it starts.
    let reload_models_mailbox = rpi_extensions::ReloadModelsMailbox::new();
    let (action_host, harness_cell) = crate::extensions_actions::HarnessActionHost::new_empty(
        catalog.clone(),
        cwd.to_path_buf(),
        runtime.clone(),
        args.unknown_flags.clone(),
        tool_context.clone(),
    );
    let action_host = Arc::new(
        action_host
            .with_session_dir(session_dir.clone())
            .with_fresh_session_only(native_store.is_some())
            .with_reload_models_mailbox(reload_models_mailbox.clone()),
    );
    let host_arc: Arc<dyn rpi_extensions::RuntimeActionHost> = action_host.clone();
    // `runtime` is reused below (B5c: `PluggableProvider` needs a captured
    // `Handle` to `spawn_blocking` the sync `ProviderRequestFn`), so clone here.
    //
    // B5d: build the initial bridge WITH a reload callback backed by a session-
    // long `ReloadMailbox` (cloned into `ReloadContext` + handed to the TUI). A
    // plugin's `runtime_action(Reload)` then signals the TUI's main loop instead
    // of hitting the "not configured" fallback. The same mailbox is reused on
    // `/reload` (the fresh bridge carries `ctx.mailbox`), so the bridge always
    // points at the one TUI-installed sender across reloads.
    let reload_mailbox = rpi_extensions::ReloadMailbox::new();
    // Session-scoped UI-dialog mailbox (runtime action 17 / `ask_user`). The
    // initial bridge carries it and the TUI attaches itself once it starts
    // polling; a headless run leaves it detached so `ask_user` fails loudly
    // instead of fabricating a user answer. The SAME mailbox is reused across
    // `/reload` so in-flight prompts + TUI attachment survive a plugin swap.
    let ui_dialog_mailbox = rpi_extensions::UiDialogMailbox::new();
    // Session-scoped extension status registry (runtime action 18 / `SetStatus`).
    // The bridge lets plugins write; the TUI polls the revision and repaints its
    // footer. Reused across `/reload` so a plugin's status survives the swap.
    let ext_status_mailbox = rpi_extensions::ExtensionStatusMailbox::new();
    // Session-scoped editor-text queue (runtime action 19 / `SetEditorText`).
    // The bridge lets a plugin drop a draft into the prompt editor; the TUI
    // drains it each tick. Reused across `/reload` so a pending transcription
    // survives a plugin swap.
    let editor_text_mailbox = rpi_extensions::EditorTextMailbox::new();
    let action_bridge = rpi_extensions::ActionBridge::with_reload_and_ui(
        runtime.clone(),
        host_arc,
        rpi_extensions::reload_callback_from_mailbox(reload_mailbox.clone()),
        ui_dialog_mailbox.clone(),
        ext_status_mailbox.clone(),
        editor_text_mailbox.clone(),
    );

    let extension_register = StartupSpan::start(Phase::ExtensionRegister);
    let extension_session = if let Some(selected) = &resource_selection {
        let session = rpi_extensions::load_session_mixed_guarded(
            &[],
            &selected.extensions,
            Arc::new(NullDiagnostics),
            Some(Arc::clone(&action_bridge)),
            Some(resource_leases.clone()),
        );
        if session.loaded_paths() != selected.extensions.as_slice() {
            return Err(BuildError::HarnessCreate(
                "selected native extension failed to load/register".into(),
            ));
        }
        if session.owns_stdio() {
            return Err(BuildError::HarnessCreate(
                "stdio-owning extensions are unsupported with --resource-config".into(),
            ));
        }
        session
    } else if args.no_extensions {
        ExtensionSession::none()
    } else {
        load_extensions(args, cwd, project_trusted, Some(Arc::clone(&action_bridge)))
    };
    extension_register.finish();
    if args.verbose {
        if let Some(s) = extension_session.summary() {
            eprintln!("extensions: {s}");
        }
        report_deferred_renderers(&extension_session);
    }
    crate::native_store::validate_stdio(native_store.is_some(), extension_session.owns_stdio())
        .map_err(BuildError::NativeStorePolicy)?;
    // Publish the extension registry so `ListCommands` can enumerate slash
    // commands. Done here rather than at host construction because the registry
    // only exists after extensions load; `/reload` republishes the new snapshot
    // via `RuntimeActionHost::publish_registry`.
    action_host.publish_registry_impl(extension_session.snapshot_arc());
    Ok(PreparedExtensions {
        resource_leases,
        resource_selection,
        session_selection,
        transition_session,
        native_store,
        session_dir,
        runtime,
        tool_context,
        action_host,
        harness_cell,
        reload_mailbox,
        reload_models_mailbox,
        ui_dialog_mailbox,
        ext_status_mailbox,
        editor_text_mailbox,
        action_bridge,
        extension_session,
    })
}

/// Build the `AgentHarness` from the resolved model + parsed args + cwd.
///
/// This is the v1 equivalent of TS `createAgentSession`. It:
/// 1. Builds the `OsExecutionEnv` rooted at `cwd`.
/// 2. Constructs the built-in tools (optionally filtered by `--tools`/
///    `--exclude-tools`/`--no-tools`/`--no-builtin-tools`).
/// 3. Resolves the session storage (ephemeral vs fresh JSONL vs restore-error).
/// 4. Assembles `AgentHarnessOptions` and calls `AgentHarness::create`.
///
/// Returns the harness plus a `broadcast::Receiver<AgentEvent>` carrying the
/// live `AgentEvent` stream from every run (backed by a `BroadcastEmitter`
/// installed on the harness). Interactive mode drains this to render streaming
/// responses; the non-interactive modes simply drop it.
/// Returns the harness, the live `AgentEvent` broadcast receiver, and a
/// [`ReloadContext`] the interactive TUI holds to drive `/reload` (and a
/// plugin's `runtime_action(Reload)` via the mailbox). Non-interactive modes
/// drop the context (no `/reload` surface in print/json mode).
pub async fn build(
    resolved: &ResolvedModel,
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
) -> Result<
    (
        AgentHarness,
        tokio::sync::broadcast::Receiver<rpi_agent::AgentEvent>,
        ReloadContext,
    ),
    BuildError,
> {
    let prepared = prepare_extensions(args, cwd, project_trusted)?;
    build_prepared(resolved, args, cwd, project_trusted, prepared).await
}

pub async fn build_prepared(
    resolved: &ResolvedModel,
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
    prepared: PreparedExtensions,
) -> Result<
    (
        AgentHarness,
        tokio::sync::broadcast::Receiver<rpi_agent::AgentEvent>,
        ReloadContext,
    ),
    BuildError,
> {
    let native_store =
        crate::native_store::root(args, cwd).map_err(BuildError::NativeStorePolicy)?;
    if native_store != prepared.native_store {
        return Err(BuildError::NativeStorePolicy(
            "experimental session selection changed after extension preparation".into(),
        ));
    }
    crate::native_store::validate_stdio(
        native_store.is_some(),
        prepared.extension_session.owns_stdio(),
    )
    .map_err(BuildError::NativeStorePolicy)?;
    let fresh_session_only = native_store.is_some();
    let cwd_str = cwd.to_string_lossy().to_string();
    // Plugin tools get the session's identity injected into every call's
    // arguments (see `rpi_extensions::ToolCallContext`). It is created here,
    // before the tools are built, and completed with the session id below once
    // the session exists — the tool set has to exist before the harness does.
    let PreparedExtensions {
        resource_leases,
        resource_selection,
        session_selection: selection,
        transition_session,
        native_store: _,
        session_dir,
        runtime,
        tool_context,
        action_host,
        harness_cell,
        reload_mailbox,
        reload_models_mailbox,
        ui_dialog_mailbox,
        ext_status_mailbox,
        editor_text_mailbox,
        action_bridge,
        extension_session,
    } = prepared;
    let catalog = crate::provider::authenticated_catalog(resolved);
    action_host.set_catalog(catalog.clone());

    // ---- Execution env + tools ----
    let env = Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf()));
    let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();
    let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
    let _registry = Arc::new(MutationQueueRegistry::new());
    // `env_dyn` is shared between the tool context (moved in) and the resource
    // loaders below (borrowed); clone one branch so both hold a reference.
    let file_io = Arc::new(rpi_tools::file_io::FileIORouter::new());
    let ctx =
        ExecutionToolContext::new(env_dyn.clone(), Some(mut_env)).with_file_io(file_io.clone());

    let mut tools = build_tools(&ctx, args);

    // ---- Extensions (Part B2) ----
    // Load cdylib plugins from the resolved extension dirs, merge their tools
    // into the built-in set (extension overrides same-named built-in; first-
    // extension-wins across plugins; explicit `--tools`/`--exclude-tools` still
    // apply to the merged set), and keep the loaded `Library` handles alive for
    // the harness lifetime via the returned session guard. `--no-extensions`
    // skips discovery entirely (no dirs scanned, no plugins loaded).
    merge_extension_tools(&mut tools, &extension_session, args, &tool_context);
    let active = if let Some(selected) = &resource_selection {
        selected
            .active_tools(
                &tools
                    .iter()
                    .map(|t| t.tool.schema().name.clone())
                    .collect::<Vec<_>>(),
            )
            .map_err(BuildError::HarnessCreate)?
    } else {
        filter_active_tool_names(active_tool_names(&tools, args), args)
    };

    // ---- Session storage ----
    // The root is the one `prepare_extensions` froze at startup, never a fresh
    // `configured_session_dir(args, cwd)`: settings and `RPI_CODING_AGENT_SESSION_DIR`
    // can change while extensions load, and an adopted session cwd would point
    // at another project's tree. Both the cwd peek and this build must read the
    // same root, so the caller pins it into `args` before either runs.
    let session = match transition_session {
        Some(session) => session,
        None => build_session(&selection, &cwd_str, resource_selection.as_ref()).await?,
    };
    crate::resource_selection::bind_session(&session, resource_selection.as_ref())
        .await
        .map_err(BuildError::SessionOpen)?;
    // Plugin tools now know which session they serve: `todo` and friends key
    // their per-session state off it. A session without an id (ephemeral) leaves
    // the field unset and a plugin is expected to fall back to project scope.
    if let Ok(metadata) = session.get_metadata().await {
        publish_session_identity(&tool_context, &metadata.id);
    }
    // `--name`/`-n` sets the session display name durably (a `name` fact in the
    // log). It applies to whichever session this launch works in — a fresh one
    // or a restored one — and an empty value clears a previous name. This is the
    // only place the flag is applied: `SessionSelection::New` carries it, but
    // builds that do not create a session (restore/fork) must still honor it.
    if let Some(requested) = args.name.as_deref() {
        let name = requested.trim();
        let name = if name.is_empty() { None } else { Some(name) };
        if let Err(error) = session.set_name(name).await {
            if fresh_session_only {
                return Err(BuildError::SessionOpen(format!(
                    "native session name persistence failed: {error}"
                )));
            }
            if args.verbose {
                eprintln!("warning: could not set session name: {error}");
            }
        }
    }
    // ---- System prompt base (precedence: --system-prompt > SYSTEM.md > default) ----
    // Mirrors pi `discoverSystemPromptFile` (`resource-loader.ts:1022-1034`):
    // an explicit `--system-prompt` flag wins; otherwise a discovered
    // `<cwd>/.rpi/SYSTEM.md` wins, then
    // `<agent_dir>/SYSTEM.md`.
    // (global); otherwise the built-in default. **Project-wins** — the same
    // direction as skills/prompts precedence.
    // Whether this model sees its own prior reasoning again decides how the
    // working-state rule is worded (see `default_system_prompt`).
    let prior_reasoning_replayed = rpi_ai::model::prior_reasoning_is_replayed(&resolved.model);
    let prompt_tools = active.clone();
    let prompt_metadata = extension_tool_prompt_metadata(&extension_session);
    let default_prompt = || {
        default_system_prompt_with_tool_prompts(
            &cwd_str,
            prior_reasoning_replayed,
            &prompt_tools,
            &prompt_metadata,
        )
    };
    let mut context_paths = Vec::new();
    let base_prompt = match args.system_prompt.as_deref() {
        Some(explicit) => explicit.to_string(),
        None if project_trusted && resource_selection.is_none() => {
            match discover_system_prompt_file(cwd) {
                Some(path) => {
                    read_prompt_source(&path, &mut context_paths).unwrap_or_else(default_prompt)
                }
                None => default_prompt(),
            }
        }
        None => default_prompt(),
    };

    // ---- Append-text sources (precedence: --append-system-prompt > APPEND_SYSTEM.md) ----
    // Mirrors pi `appendSystemPrompt` (`resource-loader.ts:525-542`). Explicit
    // `--append-system-prompt` flags are joined together; when none are given, a
    // discovered `APPEND_SYSTEM.md` (project-wins over global) provides the
    // append text. `--append-system-prompt` takes a value that may be a literal
    // string OR a readable file path (mirrors TS `resolvePromptInput`).
    let mut append_texts: Vec<String> = Vec::new();
    for extra in &args.append_system_prompt {
        let text = read_append_target(extra, &mut context_paths).unwrap_or_else(|| extra.clone());
        append_texts.push(text);
    }
    if let Some(selected) = &resource_selection {
        append_texts.extend(
            selected
                .instructions
                .iter()
                .map(|(_, _, text)| text.clone()),
        );
    }
    if args.append_system_prompt.is_empty() && resource_selection.is_none() {
        if let Some(path) = project_trusted
            .then(|| discover_append_system_prompt_file(cwd))
            .flatten()
        {
            if let Some(text) = read_prompt_source(&path, &mut context_paths) {
                append_texts.push(text);
            }
        }
    }
    let append_join = if append_texts.is_empty() {
        None
    } else {
        Some(append_texts.join("\n\n"))
    };

    // ---- Resource discovery (skills + prompt-templates + context-files) ----
    // The env is OS-backed, rooted at cwd. Each `--no-*` flag suppresses its
    // channel independently (pi parity). Skills/prompts load project→global,
    // explicit/plugin paths, then static packages; dedupe first-wins-by-name
    // keeps project and user resources ahead of packages. Context files walk
    // global→ancestor(cwd→root), deepest-last (pi parity).
    //
    // **Trust gate (v1 divergence):** pi gates project config discovery on
    // `isProjectTrusted()` (global resources are unconditional). rpi v1 has no
    // trust prompt — project resources are discovered unconditionally (a copied
    // `.rpi/` dir drops in and works). Full trust gating is deferred.
    let agent_dir = crate::config::agent_dir().ok();

    // ---- B5b: extension resources_discover ----
    // If any plugin registered a `resources_discover` handler, fan the event out
    // (reason "startup") and collect skill/prompt/theme paths. These plugin-
    // contributed paths merge WITH the static Part-A dirs (project
    // `.rpi/skills` +
    // `agent_dir/skills`, etc.) and the loaders re-run over the union — the
    // coherence point: a plugin's discovered skills land through the SAME loaders
    // as static skills. Static dirs load FIRST so project skills keep winning name
    // collisions (a plugin must not shadow a project skill of the same name —
    // mirrors pi `extendResources` running AFTER the default load's first-wins
    // map). `load_skills` now accepts both dirs and individual `.md` files, so a
    // plugin returning bare `SKILL.md` paths loads them (the gap this closes).
    // Theme paths are available to the TUI through the package resource list;
    // skill/prompt loaders are the only resources needed by the harness here.
    // A `--no-*` flag suppresses its channel for BOTH static and discovered paths.
    let discovered = extension_session
        .snapshot_arc()
        .filter(|_| resource_selection.is_none())
        .map(|snap| emit_resources_discover(&cwd_str, "startup", &snap))
        .unwrap_or_default();

    let mut skills: Vec<rpi_harness::types::Skill> = Vec::new();
    let mut skill_diags: Vec<rpi_harness::skills::SkillDiagnostic> = Vec::new();
    let mut resource_diags: Vec<rpi_harness::diagnostics::ResourceDiagnostic> = Vec::new();
    if let Some(selected) = &resource_selection {
        // Consume the preflight's parsed resources, never rediscover or reread
        // selected Skills after executable registration / writer opening.
        skills = selected
            .validated_skills()
            .map_err(BuildError::HarnessCreate)?
            .skills
            .clone();
    } else if !args.no_skills {
        let mut dirs = if args.dev_local_only || args.project_only {
            project_skill_dirs(cwd)
        } else if project_trusted {
            skill_dirs(cwd)
        } else {
            global_skill_dirs()
        };
        dirs.extend(args.skill.iter().cloned());
        dirs.extend(discovered.skill_paths.iter().map(PathBuf::from));
        let result = load_skills_with_precedence(&env_dyn, &dirs).await;
        skills = result.skills;
        skill_diags = result.diagnostics;
        resource_diags.extend(result.resource_diagnostics);
    }

    let mut prompt_templates: Vec<rpi_harness::types::PromptTemplate> = Vec::new();
    let mut prompt_diags: Vec<rpi_harness::prompt_templates::PromptTemplateDiagnostic> = Vec::new();
    if !args.no_prompt_templates && resource_selection.is_none() {
        let mut dirs = if args.dev_local_only || args.project_only {
            project_prompt_template_dirs(cwd)
        } else if project_trusted {
            prompt_template_dirs(cwd)
        } else {
            global_prompt_template_dirs()
        };
        dirs.extend(args.prompt_template.iter().cloned());
        dirs.extend(discovered.prompt_paths.iter().map(PathBuf::from));
        let result = load_prompt_templates_with_precedence(&env_dyn, &dirs).await;
        prompt_templates = result.prompt_templates;
        prompt_diags = result.diagnostics;
        resource_diags.extend(result.resource_diagnostics);
    }

    let context_block = if args.no_context_files || !project_trusted || resource_selection.is_some()
    {
        String::new()
    } else {
        // `load_project_context_files` walks the global agentDir first then
        // ancestor-walks cwd→root (deepest last). It needs a real agent_dir; if
        // none is resolvable, pass the cwd dir so only the ancestor-walk runs
        // (the global step returns None anyway).
        let agent_dir_path = agent_dir.clone().unwrap_or_else(|| cwd.to_path_buf());
        let files = load_project_context_files(&env_dyn, cwd, &agent_dir_path).await;
        context_paths.extend(
            files
                .iter()
                .map(|file| file.path.to_string_lossy().into_owned()),
        );
        format_project_context(&files)
    };

    // Surface resource-discovery diagnostics as startup warnings (verbose-only).
    if args.verbose {
        for d in &skill_diags {
            eprintln!(
                "warning: skill {} ({}): {}",
                d.path,
                d.code.as_str(),
                d.message
            );
        }
        for d in &prompt_diags {
            eprintln!(
                "warning: prompt template {} ({}): {}",
                d.path,
                d.code.as_str(),
                d.message
            );
        }
        // Structured collisions: make the winner/loser explicit.
        for d in &resource_diags {
            if let rpi_harness::diagnostics::ResourceDiagnostic::Collision(c) = d {
                eprintln!(
                    "warning: {} name \"{}\" collision: keeping {} (shadowed {})",
                    c.resource_type.as_str(),
                    c.name,
                    c.winner_path,
                    c.loser_path
                );
            }
        }
    }

    // ---- Compose the full system prompt ----
    // Order mirrors pi `buildSystemPrompt` (`system-prompt.ts:28-72`):
    // base → append → context → skills. The skills listing is the harness's own
    // section: `AgentHarness::compose_prompt` appends `<available_skills>` (gated
    // on the `read` tool + `disable_model_invocation`, applied inside
    // `format_skills_for_system_prompt`). So we pass None for skills here (the
    // harness adds the listing itself) and fold only base+append+context into
    // the prompt we hand the harness.
    let system_prompt = compose_system_prompt(
        Some(&base_prompt),
        &[], // skills: harness appends the listing itself
        if context_block.is_empty() {
            None
        } else {
            Some(&context_block)
        },
        append_join.as_deref(),
    );

    // ---- Debug: dump the resolved system-prompt sections (verification) ----
    // A verification affordance for Part-A resource discovery: prints the
    // composed sections + resource counts to stderr so a smoke can confirm
    // `<available_skills>` + `<project_context>` + appended text reached the
    // prompt without parsing a provider round-trip. The harness composes the
    // final prompt (base → append → context → skills); here we print the
    // pre-harness sections (the harness adds the skills listing itself, gated
    // on `read` + `disable_model_invocation`).
    if args.debug_system_prompt {
        eprintln!("=== --debug-system-prompt ===");
        let base_src = if args.system_prompt.is_some() {
            "--system-prompt"
        } else if discover_system_prompt_file(cwd).is_some() {
            "SYSTEM.md"
        } else {
            "default"
        };
        eprintln!("[base source: {base_src}]");
        eprintln!("--- base ---\n{base_prompt}");
        if let Some(append) = append_join.as_deref() {
            eprintln!("--- append ---\n{append}");
        } else {
            eprintln!("--- append: (none) ---");
        }
        if context_block.is_empty() {
            eprintln!("--- context: (none) ---");
        } else {
            eprintln!("--- context ---{context_block}");
        }
        let visible_skills = skills
            .iter()
            .filter(|s| s.disable_model_invocation != Some(true))
            .count();
        eprintln!(
            "--- skills: {} loaded ({} model-visible, {} hidden) ---",
            skills.len(),
            visible_skills,
            skills.len() - visible_skills
        );
        for s in &skills {
            let hidden = if s.disable_model_invocation == Some(true) {
                " [hidden]"
            } else {
                ""
            };
            eprintln!("    {}{hidden} — {}", s.name, s.description);
        }
        eprintln!("--- prompt templates: {} ---", prompt_templates.len());
        for t in &prompt_templates {
            eprintln!("    /{}", t.name);
        }
        // B5b: surface plugin-contributed discovery paths so a smoke can confirm
        // the resources_discover round-trip fed the loaders; package themes are
        // selected by the TUI rather than injected into the harness prompt.
        eprintln!(
            "--- discovered via resources_discover: {} skill(s), {} prompt(s), {} theme(s) ---",
            discovered.skill_paths.len(),
            discovered.prompt_paths.len(),
            discovered.theme_paths.len(),
        );
        for p in &discovered.skill_paths {
            eprintln!("    skill: {p}");
        }
        for p in &discovered.prompt_paths {
            eprintln!("    prompt: {p}");
        }
        eprintln!(
            "--- final composed base+append+context (skills listing added by harness) ---\n{system_prompt}"
        );
        eprintln!("=== end --debug-system-prompt ===");
    }

    // ---- Options ----
    // Install a BroadcastEmitter so the caller (the interactive TUI) can drain
    // AgentEvents live as a run unfolds. The corresponding broadcast::Receiver
    // is returned alongside the harness; non-interactive modes simply drop it.
    let (broadcast, event_rx) = rpi_agent::events::BroadcastEmitter::new(256);
    let broadcast_emitter: Arc<dyn rpi_agent::AgentEmitter> = Arc::new(broadcast);
    // The broadcast half stays live for the whole session (the TUI's drain task
    // holds the receiver); reload re-wraps it in a fresh `TeeEmitter`, so keep
    // a clone for the `ReloadContext` before the tee match consumes the original.
    let broadcast_for_context: Arc<dyn rpi_agent::AgentEmitter> = Arc::clone(&broadcast_emitter);

    // ---- Extensions emitter (Part B3a) ----
    // If extensions loaded + registered any `on()` handlers, wrap the
    // broadcast emitter in a `TeeEmitter` so every `AgentEvent` flows to BOTH
    // the TUI (via the broadcast receiver above) AND the plugin handlers (via
    // the `ExtensionEmitter`, which translates each `AgentEvent` →
    // `StablePluginEvent` and fans out to the handlers registered for its tag).
    // With no extensions the tee degrades to the bare broadcast emitter (a
    // one-child passthrough), so the TUI path is unchanged.
    let emitter: Arc<dyn rpi_agent::AgentEmitter> = match extension_session.snapshot_arc() {
        Some(snapshot) => {
            let ext = ExtensionEmitter::new(snapshot, extension_session.keepalive());
            Arc::new(TeeEmitter::new(vec![broadcast_emitter, Arc::new(ext)]))
        }
        None => broadcast_emitter,
    };

    // Saved user defaults. Best-effort: a missing or malformed file must not stop
    // a session from starting, and every accessor falls back to the harness
    // default when its key is absent.
    let saved_settings = crate::settings::load_settings().unwrap_or_default();

    // `httpProxy` applies to rpi's own HTTP clients. Setting the environment is how
    // they are configured (`pi-ai/src/http.rs`), and upstream does the same thing
    // with this key — an explicit environment variable still wins, so a shell-level
    // proxy override is not silently replaced by the setting.
    if let Some(proxy) = &saved_settings.http_proxy {
        for key in ["HTTP_PROXY", "HTTPS_PROXY"] {
            if std::env::var_os(key).is_none() {
                std::env::set_var(key, proxy);
            }
        }
    }

    // Share the live session cell with both /reload and the approval hook.
    // The hook is installed only when the current snapshot has a tool-call
    // handler; a later reload rebuilds the hook state together with the
    // extension session.
    let extension_session_cell = Arc::new(Mutex::new(extension_session.clone()));
    // Only install the approval hook when a plugin actually registered a
    // tool-call handler. An extension session may contain flags/resources but
    // no handler; dispatching through that empty session needlessly crosses
    // the plugin bridge and can leave a headless RPC tool call waiting.
    let before_tool_call = extension_session
        .snapshot()
        .filter(|snapshot| {
            !snapshot
                .handlers_for(rpi_plugin_sdk::EventTag::ToolCall)
                .is_empty()
        })
        .map(|_| {
            extension_tool_call_hook(
                Arc::clone(&extension_session_cell),
                ui_dialog_mailbox.clone(),
            )
        });

    let options = AgentHarnessOptions {
        model: resolved.model.clone(),
        thinking_level: resolved.thinking_level,
        active_tool_names: active.clone(),
        tools,
        system_prompt: Some(system_prompt),
        resources: AgentHarnessResources {
            context_files: context_paths,
            skills: if skills.is_empty() {
                None
            } else {
                Some(skills)
            },
            prompt_templates: if prompt_templates.is_empty() {
                None
            } else {
                Some(prompt_templates)
            },
        },
        // A restored session (--continue/--resume/--session) already has
        // records — let the harness load it and keep appending.
        allow_existing_session: matches!(
            selection,
            SessionSelection::Latest { .. }
                | SessionSelection::ById { .. }
                | SessionSelection::ByExactId { .. }
                | SessionSelection::Fork { .. }
        ),
        stream_options: AgentHarnessStreamOptions {
            timeout: args.timeout,
            stream_idle_timeout: args.stream_idle_timeout.or_else(|| {
                saved_settings
                    .stream_idle_timeout
                    .map(std::time::Duration::from_secs)
            }),
            thinking_budgets: saved_settings.thinking_budgets,
            ..Default::default()
        },
        // Settings-backed harness options: upstream lets users tune retry,
        // compaction and queue drain modes from settings.json. Global only,
        // matching how `defaultTools` and the other settings-backed options in
        // this file are read.
        retry: saved_settings.retry_policy(),
        compaction: saved_settings.compaction_settings(),
        steering_mode: saved_settings.steering_mode(),
        follow_up_mode: saved_settings.follow_up_mode(),
        tool_execution: HarnessToolExecution::default(),
        drive: DrivingMode::default(),
        session,
        // B5c: inject the resolved gateway provider PLUS one `Arc<dyn Provider>`
        // per registered extension provider (`PluggableProvider` wraps a plugin's
        // sync `ProviderRequestFn`). The harness's `build_stream_fn` resolves a
        // provider lazily per call by `models.iter().find(|p| p.id() == model.provider)`,
        // so a catalog model whose `provider` matches an extension provider's id
        // routes to it. Extension providers land AFTER the gateway so the gateway
        // stays first-match for its own ids (first-wins on a `.find`).
        models: build_models_with_extensions(resolved, &extension_session, runtime.clone()),
        to_provider_messages: None,
        entry_projectors: Default::default(),
        agent_emitter: Some(emitter),
        // B3b loop hooks. `before_tool_call` dispatches the pi `on("tool_call")`
        // extension event and honors a veto (a permission gate can block a tool
        // before it runs). `after_tool_call`/`transform_context` remain `None`
        // (their host-side adapters are not wired yet).
        before_tool_call,
        after_tool_call: None,
        transform_context: None,
        entry_transforms: Vec::new(),
        // Extension provider hooks (B4): plugins subscribing to the
        // BeforeProviderRequest / BeforeProviderHeaders / AfterProviderResponse
        // events observe every provider call (observer semantics — the handler
        // ABI has no patch channel in v1). A session without provider-hook
        // subscribers runs hook-free.
        provider_hooks: rpi_extensions::ExtensionProviderHooks::from_session(&extension_session)
            .map(|h| Arc::new(h) as Arc<dyn rpi_ai::ProviderHooks>),
    };

    let harness = AgentHarness::create(options)
        .await
        .map_err(|e| BuildError::HarnessCreate(e.to_string()))?;
    // Manual skill commands read current bodies through the same environment
    // that discovered these host-owned resources, including after /reload.
    harness.set_skill_execution_env(env_dyn.clone());

    // Retain read leases even for resource-only/--no-extensions hosts.
    let harness_leases = resource_leases.clone();
    harness
        .events()
        .on_lifetime::<rpi_harness::events::RunEndEvent, _>(move |_| {
            let _ = &harness_leases;
        });
    install_extension_settlement_bridge(harness.events(), extension_session_cell.clone());
    if let Some(selected) = &resource_selection {
        harness
            .seal_resources(active)
            .map_err(|e| BuildError::HarnessCreate(e.to_string()))?;
        crate::resource_selection::attach(
            &harness,
            serde_json::json!({
                "version": 1,
                "hostId": uuid::Uuid::new_v4().to_string(),
                "host": {"name": crate::APP_NAME, "version": crate::VERSION, "pid": std::process::id()},
                "selection": selected.selection,
                "package": selected.package,
                "selectedSkills": selected.skills,
                "skillDiagnostics": [],
                "loadedExtensions": extension_session.loaded_paths().iter().map(|p| serde_json::json!({"identity": p.file_stem().and_then(|s| s.to_str()), "path": p})).collect::<Vec<_>>(),
                "instructions": selected.provenance(),
                "baseSystemPrompt": "native-default",
                "projectContext": "disabled",
                "resourceDiscovery": "disabled",
                "changes": "restart-required"
            }),
        );
        let resource_observation = StartupSpan::start(Phase::ResourceObservation);
        crate::resource_selection::get_resource_state(&harness)
            .await
            .map_err(BuildError::HarnessCreate)?;
        resource_observation.finish();
        if selected.transition.is_some() {
            // Re-read persisted facts at commit, not merely the startup mirror.
            // The canonical OS writer claim acquired before registration is
            // retained by the host for the entire session lifetime.
            let meta = resolve_restore_target(&selection, &cwd_str)
                .await?
                .ok_or_else(|| {
                    BuildError::SessionOpen("transition session disappeared before commit".into())
                })?;
            let path = std::fs::canonicalize(&meta.path)
                .map_err(|e| BuildError::SessionOpen(e.to_string()))?;
            let lock_path = PathBuf::from(format!("{}.lock", path.display()));
            if !SESSION_LOCK_PATHS
                .get_or_init(|| Mutex::new(HashSet::new()))
                .lock()
                .unwrap()
                .contains(&lock_path)
            {
                return Err(BuildError::SessionOpen(
                    "resource transition lost canonical writer ownership".into(),
                ));
            }
            let facts =
                rpi_harness::session::jsonl::JsonlSessionStorage::read_resource_transition_facts(
                    Arc::new(OsExecutionEnv::with_cwd(cwd.to_path_buf())),
                    &path.to_string_lossy(),
                    Arc::new(SystemClock),
                    Arc::new(DefaultIdGenerator::new()),
                )
                .await
                .map_err(|e| BuildError::SessionOpen(e.to_string()))?;
            crate::resource_selection::validate_session_binding(&facts.values, Some(selected))
                .map_err(BuildError::SessionOpen)?;
            let mirror = rpi_harness::session::SessionValues::load(harness.session())
                .await
                .map_err(|e| BuildError::SessionOpen(e.to_string()))?;
            if crate::resource_selection::binding_state(&mirror)
                != crate::resource_selection::binding_state(&facts.values)
            {
                return Err(BuildError::SessionOpen(
                    "resource binding changed before commit".into(),
                ));
            }
        }
        let inventory_bind = StartupSpan::start(Phase::InventoryBind);
        crate::resource_selection::bind_inventory(&harness, selected)
            .await
            .map_err(BuildError::SessionOpen)?;
        inventory_bind.finish();
    }
    // Publish only after exact resources have been sealed and validated.
    crate::extensions_actions::HarnessActionHost::set_harness(
        &harness_cell,
        Arc::new(harness.clone()),
    );
    register_file_io_router(&harness, &file_io);

    // ---- Durable operation recovery (runtime layer) ----
    // A previous process can die between `operation_started` and
    // `operation_finished`, leaving the lane with an orphaned operation. Sweep
    // every lane, surface what was found, and emit `run_start`/`run_end`
    // (aborted) pairs so a TUI that already attached sees the settle. This is
    // read-only: nothing is mutated here, so a failed sweep cannot corrupt the
    // log. A clean session (the common case) reports no findings.
    {
        let runtime = rpi_harness::runtime::SessionRuntime::new(harness.session().clone());
        match rpi_harness::runtime::recover_all_lanes(&runtime).await {
            Ok(reports) => {
                for (lane, report) in reports {
                    if report.is_clean() {
                        continue;
                    }
                    if report.is_corrupt() {
                        eprintln!(
                            "warning: session lane '{lane}' is corrupt ({} open operations); \
                             refusing to auto-repair — inspect the session file",
                            report.open_operations.len()
                        );
                        continue;
                    }
                    for decision in runtime.reconcile(&report) {
                        if let rpi_harness::runtime::RecoveryDecision::Abort { run_id } = decision {
                            if args.verbose {
                                eprintln!(
                                    "recovered interrupted run {run_id} on lane '{lane}' (aborted)"
                                );
                            }
                        }
                    }
                    runtime.emit_recovery_events(harness.events(), &report);
                }
            }
            Err(e) => {
                if args.verbose {
                    eprintln!("warning: session recovery sweep failed: {e}");
                }
            }
        }
    }

    // ---- B5d: assemble the ReloadContext the TUI holds ----
    // Every field is cheap to clone (Arc / Vec / args Clone). The cells own the
    // live session + bridge so `/reload` can swap them; the harness itself is
    // NOT held here (the TUI already owns a `&AgentHarness` / clone at the call
    // site — passing it into `reload_extension_resources` keeps this structfree
    // of a harness back-reference so it can be `Clone` into the reload callback).
    // Retain the resolved root in the existing args snapshot so reload never
    // reselects it from mutable settings or environment variables.
    let mut reload_args = args.clone();
    reload_args.session_dir = Some(session_dir);
    let reload_context = ReloadContext {
        resource_leases,
        fresh_session_only,
        extension_session: extension_session_cell,
        action_bridge: Arc::new(Mutex::new(Some(Arc::clone(&action_bridge)))),
        tool_context,
        file_io,
        catalog: Arc::new(std::sync::RwLock::new(catalog)),
        providers: Arc::new(std::sync::RwLock::new(resolved.providers.clone())),
        runtime: runtime.clone(),
        cwd: cwd.to_path_buf(),
        project_trusted,
        discovered_theme_paths: Arc::new(std::sync::Mutex::new(discovered.theme_paths.clone())),
        args: reload_args,
        resolved_model: resolved.model.clone(),
        broadcast: broadcast_for_context,
        mailbox: reload_mailbox,
        // Request channel for `runtime_action(ReloadModels)` — the TUI installs
        // its sender and answers on the request's reply channel.
        reload_models_mailbox,
        ui_dialog: ui_dialog_mailbox,
        ext_status: ext_status_mailbox,
        editor_text: editor_text_mailbox,
        dev_extension: None,
        startup_notices: Vec::new(),
    };

    Ok((harness, event_rx, reload_context))
}

/// Project authoritative harness termination, not per-attempt AgentEnd, to
/// plugins. The bus owns this listener; its session cell follows reloads and
/// retains callback libraries, while the action host binds the harness weakly.
fn install_extension_settlement_bridge(
    events: &rpi_harness::events::HarnessEventBus,
    sessions: ExtensionSessionCell,
) {
    events.on_lifetime::<rpi_harness::events::RunEndEvent, _>(move |event| {
        if event.lane != "main" {
            return;
        }
        let session = sessions.lock().unwrap().clone();
        if let Some(snapshot) = session.snapshot_arc() {
            rpi_extensions::dispatch_empty_event(&snapshot, rpi_plugin_sdk::EventTag::AgentSettled);
        }
    });
}

/// Resolve the current plugin session for each call rather than keeping a
/// startup snapshot. A clone retains its libraries while approval is pending.
fn extension_tool_call_hook(
    sessions: ExtensionSessionCell,
    ui_dialog: rpi_extensions::UiDialogMailbox,
) -> rpi_agent::BeforeToolCall {
    Arc::new(move |ctx: rpi_agent::BeforeToolCallContext<'_>, signal| {
        let session = sessions
            .lock()
            .map(|session| session.clone())
            .map_err(|_| ());
        let ui_dialog = ui_dialog.clone();
        let tool_call_id = ctx.tool_call.id.clone();
        let tool_name = ctx.tool_call.name.clone();
        let params = ctx.args.clone();
        Box::pin(async move {
            let reason = match session {
                Ok(session) => {
                    rpi_extensions::dispatch_tool_call_event(
                        &session,
                        &tool_call_id,
                        &tool_name,
                        &params,
                        &signal,
                        &ui_dialog,
                    )
                    .await
                }
                Err(_) => Some("extension approval session is unavailable".to_string()),
            };
            reason.map(|reason| rpi_agent::BeforeToolCallResult {
                block: true,
                reason: Some(reason),
                terminate: false,
                args: None,
            })
        })
    })
}

// ===========================================================================
// B5d — `/reload`: re-run extension + resource discovery into a LIVE harness
// ===========================================================================
//
// `/reload` (interactive TUI command, or a plugin's `runtime_action(Reload)`)
// re-runs everything `build` did around resources/extensions WITHOUT rebuilding
// the `AgentHarness` itself (rebuilding would tear down the session/lane/event
// wiring + the broadcast drain task the TUI owns). Instead it:
//
//  1. Builds a fresh `ExtensionSession` (re-load the cdylibs) over the same
//     dir set, with a FRESH `ActionBridge` (the old one is `invalidate`d so
//     in-flight plugin→host calls on the old bridge fail fast).
//  2. Fans `resources_discover(_, "reload")` over the fresh snapshot.
//  3. Re-runs the Part-A loaders (skills/prompts/context/SYSTEM.md/
//     APPEND_SYSTEM.md) with the discovered paths merged in — same precedence
//     + `--no-*` gates as startup.
//  4. Rebuilds the harness's live state via the B5d setters
//     (`set_system_prompt`/`set_resources`/`set_agent_emitter`/`set_models`/
//     `set_provider_hooks`/`set_tools`) so the NEXT run observes the reloaded
//     config (in-flight runs finish on the old `ConfigSnapshot`).
//  5. Swaps the cells (`ExtensionSession`, `ActionBridge`, harness action
//     host's harness cell stays — the harness is the same object) and drops
//     the old session + bridge (their keepalives unmap the old cdylibs; the
//     new session's keepalive holds the fresh mappings).
//
// The reload is a `rpi-cli` concern (NOT a harness op): `rpi-extensions`
// carries only the `ActionBridge` staleness flag + a `ReloadMailbox` `()` signal
// (no pi-cli `TuiMessage` type — leaf DAG preserved). The TUI owns the mailbox
// receiver + the actual reload routine; a plugin's
// `runtime_action(Reload)` signals the mailbox and returns `Ok(null)`
// immediately so the calling plugin's cdylib is NOT unmapped while its
// `runtime_action` frame is still on the stack (the self-unmapping race a
// synchronous plugin-initiated reload would have).
//
// `reload_extension_resources` is the shared routine both `/reload` (TUI) and
// a plugin's `runtime_action(Reload)` (via the mailbox) drive. It is `pub` so
// the TUI's main-loop handler + the mailbox-driven path call the same code.

/// The cell that holds the live `ExtensionSession` across a `/reload`. Cloned
/// into every site that needs the current session (the TUI, the reload
/// callback). On reload the old session is `replace`d out (its `active` flag
/// flipped) and the fresh one stored atomically. In-flight callbacks retain
/// the old keepalive until they return; callers never observe an empty
/// placeholder during the swap.
pub type ExtensionSessionCell = Arc<Mutex<ExtensionSession>>;

/// The cell that holds the live `ActionBridge` across a `/reload`. A plugin
/// stores the bridge's raw `user_data` pointer during `register`; on reload the
/// old bridge is `invalidate`d (in-flight calls fail fast) and the fresh one
/// `store`d. The fresh session's plugins are handed the fresh bridge pointer.
pub type ActionBridgeCell = Arc<Mutex<Option<Arc<rpi_extensions::ActionBridge>>>>;

/// Everything `/reload` needs to rebuild extension + resource state into a live
/// harness. Built once in [`build`] (alongside the harness) and held by the TUI
/// (cloned into the reload callback the bridge carries + the `/reload` command
/// handler). The harness itself is NOT held here — the TUI already owns a
/// `&AgentHarness` / a clone; passing it at the call site keeps this struct
/// free of a harness back-reference (so it can be `Clone` and moved into the
/// reload callback without borrowing the harness).
#[derive(Clone)]
pub struct ReloadContext {
    resource_leases: Arc<Vec<crate::resource_management::Lease>>,
    /// Startup routing capability, retained independently of mutable args/settings.
    fresh_session_only: bool,
    /// One neutral router for all built-ins, including those reconstructed on reload.
    pub file_io: Arc<rpi_tools::file_io::FileIORouter>,
    /// The live extension-session cell (swapped on reload).
    pub extension_session: ExtensionSessionCell,
    /// The session's plugin-tool context (`cwd` + session id), reused — not
    /// rebuilt — on reload so the rebuilt adapters keep serving the SAME
    /// session. A reload that made a fresh context would leave every plugin
    /// tool without a session id until the next launch.
    pub tool_context: rpi_extensions::ToolCallContext,
    /// The live action-bridge cell (swapped + old invalidated on reload).
    pub action_bridge: ActionBridgeCell,
    /// Request channel for `runtime_action(ReloadModels)` (action id 31).
    ///
    /// The TUI installs its sender here and performs the (cheap) catalog
    /// refresh on its main loop, then answers on the request's reply channel.
    pub reload_models_mailbox: rpi_extensions::ReloadModelsMailbox,
    /// The extension status registry the TUI renders in its footer (`SetStatus`).
    /// Lives in the context (not the cell) so it survives a bridge swap.
    pub ext_status: rpi_extensions::ExtensionStatusMailbox,
    /// The extension editor-text queue the TUI drains into the prompt editor
    /// (`SetEditorText`, runtime action 19). Lives in the context (not the cell)
    /// so it survives a bridge swap — exactly like `ext_status`. A voice plugin
    /// writes a transcription here as a *draft*; the TUI applies it, starts the
    /// optional auto-send countdown, and submits through the normal path so the
    /// user bubble renders as usual.
    pub editor_text: rpi_extensions::EditorTextMailbox,
    /// Shared authenticated model catalog for runtime actions and the TUI.
    /// Provider-package reloads replace it together with the runtime bindings.
    pub catalog: Arc<std::sync::RwLock<Vec<rpi_ai::Model>>>,
    /// Current configured provider runtimes. Replaced after package reload so
    /// subsequent reloads cannot recover stale plugin snapshots.
    pub providers: Arc<std::sync::RwLock<Vec<Arc<dyn Provider>>>>,
    /// The ambient runtime handle (captured in `build`) — `PluggableProvider`
    /// + the fresh `ActionBridge` need a captured `Handle` to spawn from any
    /// thread.
    pub runtime: tokio::runtime::Handle,
    /// The cwd (for static resource-dir resolution + context-file walk).
    pub cwd: PathBuf,
    /// The project-trust decision captured before provider and session setup.
    /// Startup consumers reuse this value so extension code cannot change the
    /// effective policy by mutating the process cwd while it is loading.
    pub project_trusted: bool,
    /// Theme file paths contributed by extensions via `resources_discover`,
    /// shared so the TUI theme registry can merge them. Filled at startup,
    /// replaced on every `/reload` (paths are unattributed — the v1 divergence
    /// documented in `rpi-extensions::resources`).
    pub discovered_theme_paths: Arc<std::sync::Mutex<Vec<String>>>,
    /// The parsed args (cloned) — `--no-*`/`--tools`/`--exclude-tools`/
    /// `--extensions-dir`/`--no-extensions`/`--system-prompt`/etc all apply on
    /// reload exactly as at startup (a reload re-reads the same flags; it does
    /// not pick up argv changes mid-session, which is the right contract — pi's
    /// `/reload` re-runs discovery with the same config). `session_dir` holds
    /// the resolved startup root, including saved settings/environment defaults.
    pub args: Args,
    /// The resolved model + thinking level (the harness's active model stays
    /// unless `set_model` changed it; reload does not touch the model).
    pub resolved_model: rpi_ai::Model,
    /// The broadcast emitter the harness was built with. Reload rebuilds the
    /// `TeeEmitter` over the fresh `ExtensionEmitter` (the old tee's extension
    /// child is dropped, unsubscribing from the old registry). The broadcast
    /// half stays live the whole session (the TUI's drain task holds the
    /// receiver), so we keep a handle to re-wrap.
    pub broadcast: Arc<dyn rpi_agent::AgentEmitter>,
    /// The session-long reload mailbox (B5d). Build creates one, installs it on
    /// the initial `ActionBridge` via [`reload_callback_from_mailbox`], and hands
    /// a clone to the TUI. The TUI installs its `TuiMessage` sender so a plugin's
    /// `runtime_action(Reload)` signals the main loop — the reload routine reuses
    /// THIS mailbox (not a fresh default) when building the fresh bridge, so the
    /// bridge always carries the mailbox the TUI installed across reloads.
    pub mailbox: rpi_extensions::ReloadMailbox,
    /// Session-long UI-dialog mailbox (runtime action 17). The initial bridge
    /// carries it; the TUI attaches itself and polls `take_pending` from its key
    /// loop, and `/reload` reuses THIS mailbox when it builds the fresh bridge.
    /// A detached mailbox makes `ask_user` fail with "requires an interactive
    /// UI" (headless contract).
    pub ui_dialog: rpi_extensions::UiDialogMailbox,
    /// Active `rpi dev` extension builder. `/reload` rebuilds it before
    /// swapping plugin sessions; its watcher signals `mailbox` after a
    /// successful background build.
    pub dev_extension: Option<Arc<crate::dev_extension::DevExtension>>,
    /// One-line notices the host should show at startup (a session directory
    /// that fell back, an untrusted adopted directory). Filled by the CLI after
    /// the build and rendered as transcript notes in the TUI; headless modes
    /// print them to stderr. Kept as data — never `eprintln!`d from here, which
    /// would corrupt the alternate screen (issue #115).
    pub startup_notices: Vec<String>,
}

impl ReloadContext {
    pub(crate) fn session_dir(&self) -> PathBuf {
        // The fallback also supports contexts constructed by embedding hosts.
        self.args
            .session_dir
            .clone()
            .unwrap_or_else(|| configured_session_dir(&self.args, &self.cwd))
    }
}

/// Dispatch a no-payload lifecycle event (e.g. `SessionStart`, `SessionShutdown`,
/// the rpi-specific `BeforeTuiStart`) to every extension handler subscribed to
/// that tag, with **veto** + **timeout** semantics (P1).
///
/// Resolves the **current** extension session from the reload cell so a
/// `/reload`-swapped session is honored. Each handler runs on a blocking thread
/// with a per-event timeout ([`rpi_extensions::lifecycle_timeout_for`]); a
/// handler returning [`rpi_plugin_sdk::EVENT_HANDLER_ABORT`] vetoes the phase
/// and this returns `Some(reason)`. No-op (`None`) when no plugins loaded, no
/// handler subscribes, or the registry is stale.
///
/// The caller decides what a veto means: a `BeforeTuiStart` veto aborts startup
/// (never enter the TUI); `SessionStart`/`SessionShutdown` vetoes are advisory
/// (the session is already up or already closing).
pub async fn dispatch_session_event_async(
    reload_context: &ReloadContext,
    tag: rpi_plugin_sdk::EventTag,
) -> Option<String> {
    // Take the snapshot Arc + keepalive, drop the session lock, then dispatch —
    // we must not hold the session cell's mutex across the await (a handler
    // could re-enter a reload path that locks it). Holding the keepalive across
    // the await keeps the cdylibs (owning the handler fn pointers) mapped even
    // if a concurrent `/reload` swaps the session cell out.
    let (snapshot, keepalive) = {
        let session = reload_context.extension_session.lock().unwrap();
        (session.snapshot_arc(), session.keepalive())
    };
    match snapshot {
        Some(snapshot) => rpi_extensions::dispatch_lifecycle_event(&snapshot, tag, keepalive).await,
        None => None,
    }
}

/// Join plugin teardown before unloading modules. A detached timed-out handler
/// could still execute code in the old DLL, so it cannot safely permit a swap.
async fn shutdown_extensions_for_reload(
    snapshot: Option<Arc<rpi_extensions::RegistrySnapshot>>,
    keepalive: Arc<rpi_extensions::PluginKeepalive>,
) -> Result<(), String> {
    if let Some(snapshot) = snapshot {
        tokio::task::spawn_blocking(move || {
            let _keepalive = keepalive;
            rpi_extensions::dispatch_empty_event(
                &snapshot,
                rpi_plugin_sdk::EventTag::SessionShutdown,
            );
        })
        .await
        .map_err(|error| error.to_string())?;
    }
    Ok(())
}

/// The outcome of a reload: a human-readable status line for the transcript
/// (counts of what reloaded), and whether any load diagnostics appeared.
pub struct ReloadOutcome {
    /// One-line summary for the transcript note (e.g. "Reloaded 2 plugin(s),
    /// 5 skill(s), 1 prompt(s).").
    pub summary: String,
    /// True iff at least one extension load warning fired (ABI mismatch / skip).
    pub had_warnings: bool,
    /// Formatted diagnostic lines (package / skill / prompt-template warnings)
    /// that the caller should surface in its UI. Kept out of `eprintln!`
    /// because the TUI owns the terminal: direct stderr writes during a run
    /// would corrupt the alternate-screen input row.
    pub details: Vec<String>,
    /// True iff the reload itself failed hard (e.g. the extension build or
    /// settings reload errored) and the previous resources were kept. Soft
    /// diagnostics (skill shadowing, name validation) do not set this.
    pub had_errors: bool,
    /// The project-trust policy this reload actually resolved and applied.
    /// `None` when the reload failed hard and kept the previous resources, so
    /// consumers (e.g. theme discovery) should stay on the snapshot they
    /// already match instead of re-reading a stale startup bool.
    pub project_trusted: Option<bool>,
}

struct PreparedReloadInputs {
    extension_dirs: Vec<PathBuf>,
    extension_files: Vec<PathBuf>,
    skill_base_dirs: Vec<PathBuf>,
    prompt_base_dirs: Vec<PathBuf>,
}

/// Append the resource sources that are specific to a reload after the
/// conventional project/global directories. Keep this order aligned with the
/// initial build: explicit CLI paths must remain available after `/reload`,
/// while discovered and package resources retain their lower precedence.
fn append_reload_resource_paths(
    mut paths: Vec<PathBuf>,
    explicit: &[PathBuf],
    discovered: &[String],
    js_paths: &[PathBuf],
    package_paths: &[PathBuf],
) -> Vec<PathBuf> {
    paths.extend(explicit.iter().cloned());
    paths.extend(discovered.iter().map(PathBuf::from));
    paths.extend(js_paths.iter().cloned());
    paths.extend(package_paths.iter().cloned());
    paths
}

/// Re-run extension + resource discovery and push the rebuilt state into the
/// live `harness` via the B5d setters. The old `ExtensionSession` +
/// `ActionBridge` are invalidated + swapped in [`ReloadContext`]'s cells. This
/// is the single routine both `/reload` (TUI) and a plugin's
/// `runtime_action(Reload)` drive (the latter via the mailbox signal).
///
/// Returns a [`ReloadOutcome`] for the transcript. Best-effort: a failure in
/// one channel (e.g. a plugin that fails to reload) does not abort the others —
/// the reload completes with whatever loaded, mirroring pi's per-plugin
/// skip-on-error. A hard failure (e.g. the harness is closed) surfaces as an
/// error summary.
pub async fn reload_extension_resources(
    harness: &AgentHarness,
    ctx: &ReloadContext,
) -> ReloadOutcome {
    reload_extension_resources_inner(harness, ctx, || {}).await
}

/// Re-read the model/provider catalog from disk and swap it into the live
/// harness — **without** reloading extensions.
///
/// This is the cheap counterpart of [`reload_extension_resources`]. That
/// routine rebuilds the whole `ExtensionSession`, which invalidates the old
/// `RegistrySnapshot`; the **in-flight run** captured its emitter at run start
/// (`agent_harness.rs`, `snap.agent_emitter.clone()`), so from that moment
/// every plugin event is dropped by the staleness guard in
/// `dispatch_to_handlers`. The TUI keeps rendering (it uses the broadcast
/// emitter), so the session *looks* alive while its plugin event stream is
/// dead until the next run.
///
/// This routine therefore touches only:
///
/// 1. the model catalog (`ctx.catalog` + the action host's copy), and
/// 2. the provider runtime bindings (`harness.set_models`).
///
/// The extension registry, the emitter and the event snapshot are **not**
/// touched, so a running turn keeps receiving plugin events.
///
/// `models.json` is re-read from disk by the resolver
/// (`resolve_with_extensions` → `load_models_config`), so a catalog edited
/// out-of-band is picked up here.
pub async fn reload_model_catalog_only(
    harness: &AgentHarness,
    ctx: &ReloadContext,
) -> Result<usize, String> {
    let project_trusted = resolve_project_trust(&ctx.args, &ctx.cwd);

    // The **current** extension session is reused (not reloaded): its provider
    // packages are still registered, so their models stay in the catalog.
    let session_guard = ctx.extension_session.lock().unwrap();
    let extension_session = session_guard.clone();
    drop(session_guard);

    // Re-resolve from disk: `resolve_with_extensions` re-reads `models.json`
    // and re-reads credentials. Keep the currently selected provider/model so
    // a catalog refresh does not silently switch the user to another model.
    let mut args = ctx.args.clone();
    let current = harness.lane("main").get_model().await.ok();
    if let Some(model) = current.as_ref() {
        args.provider = Some(model.provider.clone());
        args.model = Some(model.id.clone());
        // ⚠️ 只在 **provider 变了** 时清掉 CLI 覆盖的密钥 / endpoint。
        //
        // `api_key` / `base_url` 是 CLI 级的覆盖，resolver 只在 provider/model
        // 身份选定**之后**才应用它们（见 `rpi-runtime/src/provider.rs`）。
        // 无条件清掉会让「用 `--api-key` 起的会话」在刷新后回落到
        // `models.json` / 环境变量 —— 凭据来源静默变了。
        //
        // 上游的 `reload_extension_resources` 用的是同一条判据
        // （`current.provider != ctx.resolved_model.provider`）：换了 provider
        // 才说明原来的 CLI 覆盖不再适用。
        if ctx.resolved_model.provider != model.provider {
            args.api_key = None;
            args.base_url = None;
        }
    }
    // Keep the lane's current thinking level: the catalog refresh is about
    // *which models exist*, not about changing how the current one thinks.
    args.thinking = harness.lane("main").get_thinking_level().await.ok();

    let resolved = crate::provider::resolve_with_extensions(
        &args,
        &ctx.cwd,
        project_trusted,
        &extension_session,
    )
    .map_err(|error| format!("重读模型清单失败：{error}"))?;

    let catalog = crate::provider::authenticated_catalog(&resolved);

    // Swap the runtime bindings FIRST: the harness must be able to route to a
    // provider that only exists in the new catalog before anything tries to
    // use it.
    let providers = build_models_with_extensions_for_reload(
        &resolved.providers,
        &extension_session,
        ctx.runtime.clone(),
    );
    harness
        .set_models(providers)
        .await
        .map_err(|error| format!("换模型运行时失败：{error}"))?;

    // Publish the catalog to every reader: the action host (what `set_model`
    // resolves against) and the reload context (what the TUI renders).
    if let Some(bridge) = ctx.action_bridge.lock().unwrap().as_ref() {
        bridge.clone_host().replace_model_catalog(catalog.clone());
    }
    *ctx.catalog
        .write()
        .unwrap_or_else(|error| error.into_inner()) = catalog.clone();
    *ctx.providers
        .write()
        .unwrap_or_else(|error| error.into_inner()) = resolved.providers.clone();

    // Keep the lane on its current model identity, but take the entry from the
    // NEW catalog: cloning the old object would leave the lane running on
    // stale metadata (endpoint, context window, sampling) that the catalog no
    // longer serves. Fall back to the freshly resolved model when the current
    // one is gone.
    let lane_model = current
        .as_ref()
        .and_then(|model| {
            catalog
                .iter()
                .find(|m| m.id == model.id && m.provider == model.provider)
        })
        .cloned()
        .unwrap_or_else(|| resolved.model.clone());
    harness
        .lane("main")
        .set_model(lane_model)
        .await
        .map_err(|error| format!("设置当前模型失败：{error}"))?;

    tracing::info!(
        models = catalog.len(),
        provider = %resolved.model.provider,
        "模型清单已重读（未重载扩展）"
    );
    Ok(catalog.len())
}

async fn reload_extension_resources_inner<F>(
    harness: &AgentHarness,
    ctx: &ReloadContext,
    after_prepare: F,
) -> ReloadOutcome
where
    F: FnOnce() + Send,
{
    if harness.has_exact_resources()
        || ctx.args.resource_config.is_some()
        || ctx.args.resource_transition.is_some()
    {
        return ReloadOutcome {
            summary:
                "Exact resource selection is restart-required; reload/dev hot apply unsupported"
                    .into(),
            had_warnings: true,
            details: Vec::new(),
            had_errors: true,
            project_trusted: None,
        };
    }
    // Resolve the arguments once for this reload. `rpi dev` may append its
    // freshly staged extension directory; every subsequent loader and policy
    // decision must observe that same effective set rather than falling back
    // to the pre-dev snapshot held in `ctx.args`.
    let mut effective_args = ctx.args.clone();
    if let Some(dev) = &ctx.dev_extension {
        if let Err(error) = dev
            .rebuild()
            .and_then(|_| dev.apply_to_args(&mut effective_args))
        {
            return ReloadOutcome {
                summary: format!(
                    "Extension build failed for {}: {error}. Keeping the currently loaded version.",
                    dev.package_name()
                ),
                had_warnings: true,
                details: Vec::new(),
                had_errors: true,
                project_trusted: None,
            };
        }
    }
    let cwd_str = ctx.cwd.to_string_lossy().to_string();
    let project_trusted = resolve_project_trust(&effective_args, &ctx.cwd);
    let prepared = match prepare_reload_inputs(&effective_args, &ctx.cwd, project_trusted) {
        Ok(prepared) => prepared,
        Err(error) => {
            return ReloadOutcome {
                summary: format!(
                    "Settings reload failed: {error}. Keeping the currently loaded resources."
                ),
                had_warnings: true,
                details: Vec::new(),
                had_errors: true,
                project_trusted: None,
            };
        }
    };
    after_prepare();
    let PreparedReloadInputs {
        extension_dirs,
        extension_files,
        skill_base_dirs,
        prompt_base_dirs,
    } = prepared;
    let mut warnings = false;

    // Shutdown callbacks must finish before a replacement can unload old DLLs.
    // Ordinary lifecycle dispatch can time out or stop on a veto; unload cleanup
    // instead joins every handler while keeping its module mapped.
    let (shutdown_snapshot, shutdown_keepalive) = {
        let session = ctx.extension_session.lock().unwrap();
        (session.snapshot_arc(), session.keepalive())
    };
    if let Err(reason) = shutdown_extensions_for_reload(shutdown_snapshot, shutdown_keepalive).await
    {
        return ReloadOutcome {
            summary: "Reload failed during extension shutdown".into(),
            had_warnings: true,
            details: vec![reason],
            had_errors: true,
            project_trusted: None,
        };
    }

    // ---- 1. Build a fresh ActionBridge + ExtensionSession ----
    // The fresh bridge carries the SAME `HarnessActionHost` (the host's harness
    // cell already points at this harness; the host impl is reusable across
    // reloads — only the bridge's staleness flag + reload callback differ). We
    // re-use the host by reading it off the OLD bridge (it's the same
    // `Arc<dyn RuntimeActionHost>`).
    let old_bridge = ctx.action_bridge.lock().unwrap().clone();
    let host: Arc<dyn rpi_extensions::RuntimeActionHost> = match &old_bridge {
        Some(b) => b.clone_host(),
        None => {
            // No prior bridge (no extensions ever loaded). Build a fresh host so
            // a reload that newly discovers plugins can still drive actions.
            let (action_host, _cell) = crate::extensions_actions::HarnessActionHost::new_empty(
                ctx.catalog
                    .read()
                    .unwrap_or_else(|error| error.into_inner())
                    .clone(),
                ctx.cwd.clone(),
                ctx.runtime.clone(),
                ctx.args.unknown_flags.clone(),
                ctx.tool_context.clone(),
            );
            crate::extensions_actions::HarnessActionHost::set_harness(
                &_cell,
                Arc::new(harness.clone()),
            );
            Arc::new(
                action_host
                    .with_session_dir(ctx.session_dir())
                    .with_fresh_session_only(ctx.fresh_session_only),
            )
        }
    };

    let reload_cb = rpi_extensions::reload_callback_from_mailbox(ctx.mailbox.clone());
    let publish_host = Arc::clone(&host);
    let fresh_bridge = rpi_extensions::ActionBridge::with_reload_and_ui(
        ctx.runtime.clone(),
        host,
        reload_cb,
        ctx.ui_dialog.clone(),
        ctx.ext_status.clone(),
        ctx.editor_text.clone(),
    );

    let extension_session = if effective_args.no_extensions {
        rpi_extensions::ExtensionSession::none()
    } else {
        load_extensions_from_files(
            &extension_dirs,
            &extension_files,
            Some(Arc::clone(&fresh_bridge)),
            Some(ctx.resource_leases.clone()),
        )
    };
    if let Err(error) =
        crate::native_store::validate_stdio(ctx.fresh_session_only, extension_session.owns_stdio())
    {
        return ReloadOutcome {
            summary: "Reload refused by experimental native session policy".into(),
            had_warnings: true,
            details: vec![error],
            had_errors: true,
            project_trusted: None,
        };
    }
    if extension_session.is_empty() && !effective_args.no_extensions {
        // The fresh session may be empty if no cdylibs are present — not a
        // warning per se, but note it.
    }
    if effective_args.verbose {
        if let Some(s) = extension_session.summary() {
            eprintln!("reload: {s}");
        }
        report_deferred_renderers(&extension_session);
    }

    let has_package_providers = |session: &ExtensionSession| {
        session.snapshot().is_some_and(|snapshot| {
            snapshot
                .providers()
                .iter()
                .any(|provider| snapshot.provider_manifest(&provider.provider_id).is_some())
        })
    };
    let packaged = has_package_providers(&extension_session)
        || has_package_providers(&ctx.extension_session.lock().unwrap());
    let refreshed = if packaged {
        let mut args = effective_args.clone();
        if let Ok(current) = harness.lane("main").get_model().await {
            let was_package = ctx
                .extension_session
                .lock()
                .unwrap()
                .snapshot()
                .is_some_and(|snapshot| snapshot.provider_manifest(&current.provider).is_some());
            let has_replacement = extension_session.snapshot().is_some_and(|snapshot| {
                snapshot
                    .providers()
                    .iter()
                    .any(|provider| provider.provider_id == current.provider)
            });
            if was_package && !has_replacement {
                return ReloadOutcome {
                    summary: "Reload failed while resolving provider packages".into(),
                    had_warnings: true,
                    details: vec![format!(
                        "active provider package '{}' is no longer loaded",
                        current.provider
                    )],
                    had_errors: true,
                    project_trusted: None,
                };
            }
            args.provider = Some(current.provider.clone());
            args.model = Some(current.id);
            if current.provider != ctx.resolved_model.provider {
                args.api_key = None;
                args.base_url = None;
            }
        }
        match crate::provider::resolve_with_extensions(
            &args,
            &ctx.cwd,
            ctx.project_trusted,
            &extension_session,
        ) {
            Ok(resolved) => Some(resolved),
            Err(error) => {
                return ReloadOutcome {
                    summary: "Reload failed while resolving provider packages".into(),
                    had_warnings: true,
                    details: vec![error.to_string()],
                    had_errors: true,
                    project_trusted: None,
                }
            }
        }
    } else {
        None
    };

    // ---- 2. Invalidate the old session + bridge BEFORE the swap ----
    // The old registry's `active` flag flips false so any in-flight
    // `emit_resources_discover`/event dispatch on the old snapshot no-ops; the
    // old bridge's flag flips false so in-flight `runtime_action` calls parked
    // on the old `user_data` hit the staleness guard. Invalidate under the
    // session lock before exposing the replacement to approval callbacks.
    //
    // Swap directly to the fresh session under the same lock that approvals
    // use. A temporary empty session would let a concurrent call bypass its gate.
    // In-flight approvals retain their old session until their callback exits;
    // its staleness flag prevents them from authorizing after this swap.
    let old_session = {
        let mut session_guard = ctx.extension_session.lock().unwrap();
        let old_session = std::mem::replace(&mut *session_guard, extension_session.clone());
        if let Some(old_snap) = old_session.snapshot_arc() {
            // `invalidate` is on the registry, but the snapshot shares the flag —
            // flipping the snapshot's flag invalidates the registry too (same Arc).
            // `RegistrySnapshot` exposes `active_flag()` for this.
            old_snap.active_flag().store(false, Ordering::SeqCst);
        }
        old_session
    };
    if let Some(old_b) = old_bridge {
        old_b.invalidate();
    }
    // Shutdown callbacks see stale host actions and run outside session locks.
    old_session.begin_shutdown();

    // The session was swapped atomically above; expose its matching bridge.
    *ctx.action_bridge.lock().unwrap() = Some(Arc::clone(&fresh_bridge));
    // Republish the registry so `ListCommands` reflects the commands the reload
    // just discovered — without this a remote UI keeps serving the startup set
    // until the host process restarts. Via the trait method because all we hold
    // here is `Arc<dyn RuntimeActionHost>`.
    // Publish even when the reload found nothing: a nonempty -> empty reload
    // returns `ExtensionSession::none()`, and skipping it would leave the old
    // snapshot advertised.
    publish_host.publish_registry_opt(extension_session.snapshot_arc());

    // ---- 3. resources_discover ("reload") over the fresh snapshot ----
    let discovered = extension_session
        .snapshot_arc()
        .map(|snap| rpi_extensions::emit_resources_discover(&cwd_str, "reload", &snap))
        .unwrap_or_default();
    // Publish the fresh theme paths so the TUI registry rebuild sees the same
    // set the skill/prompt loaders just consumed.
    if let Ok(mut guard) = ctx.discovered_theme_paths.lock() {
        *guard = discovered.theme_paths.clone();
    }

    // ---- 4. Re-run the Part-A loaders (same precedence + --no-* gates) ----
    let env = Arc::new(rpi_tools::OsExecutionEnv::with_cwd(ctx.cwd.clone()));
    let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();

    let mut skills: Vec<rpi_harness::types::Skill> = Vec::new();
    let mut skill_diags: Vec<rpi_harness::skills::SkillDiagnostic> = Vec::new();
    let mut resource_diags: Vec<rpi_harness::diagnostics::ResourceDiagnostic> = Vec::new();
    if !effective_args.no_skills {
        // JS discovery is backed by the session-long lazy Node host. Until
        // that host is swapped as part of a future full JS reload, preserve
        // the paths it contributed at startup across `/reload`.
        let js_paths: &[PathBuf] = &[];
        let package_paths: &[PathBuf] = &[];
        let dirs = append_reload_resource_paths(
            skill_base_dirs,
            &effective_args.skill,
            &discovered.skill_paths,
            js_paths,
            package_paths,
        );
        let result = load_skills_with_precedence(&env_dyn, &dirs).await;
        skills = result.skills;
        skill_diags = result.diagnostics;
        resource_diags.extend(result.resource_diagnostics);
    }

    let mut prompt_templates: Vec<rpi_harness::types::PromptTemplate> = Vec::new();
    let mut prompt_diags: Vec<rpi_harness::prompt_templates::PromptTemplateDiagnostic> = Vec::new();
    if !effective_args.no_prompt_templates {
        let js_paths: &[PathBuf] = &[];
        let package_paths: &[PathBuf] = &[];
        let dirs = append_reload_resource_paths(
            prompt_base_dirs,
            &effective_args.prompt_template,
            &discovered.prompt_paths,
            js_paths,
            package_paths,
        );
        let result = load_prompt_templates_with_precedence(&env_dyn, &dirs).await;
        prompt_templates = result.prompt_templates;
        prompt_diags = result.diagnostics;
        resource_diags.extend(result.resource_diagnostics);
    }

    let context_files = if effective_args.no_context_files || !project_trusted {
        Vec::new()
    } else {
        let agent_dir = crate::config::agent_dir().ok();
        let agent_dir_path = agent_dir.unwrap_or_else(|| ctx.cwd.clone());
        load_project_context_files(&env_dyn, &ctx.cwd, &agent_dir_path).await
    };
    let context_block = format_project_context(&context_files);

    let mut details: Vec<String> = Vec::new();
    if let Some(resolved) = &refreshed {
        warnings |= !resolved.compat_warnings.is_empty();
        details.extend(
            resolved
                .compat_warnings
                .iter()
                .map(|warning| format!("warning: models.json: {warning}")),
        );
    }
    if !skill_diags.is_empty() || !prompt_diags.is_empty() || !resource_diags.is_empty() {
        warnings = true;
        // Collect formatted diagnostics for the caller to surface in its UI
        // instead of writing to stderr directly: the TUI owns the terminal,
        // and a raw `eprintln!` during a run would scribble over the
        // alternate-screen input row.
        for d in &skill_diags {
            details.push(format!(
                "warning: skill {} ({}): {}",
                d.path,
                d.code.as_str(),
                d.message
            ));
        }
        for d in &prompt_diags {
            details.push(format!(
                "warning: prompt template {} ({}): {}",
                d.path,
                d.code.as_str(),
                d.message
            ));
        }
        // Structured collisions: winner (kept) · loser (shadowed).
        for d in &resource_diags {
            if let rpi_harness::diagnostics::ResourceDiagnostic::Collision(c) = d {
                details.push(format!(
                    "warning: {} name \"{}\" collision: keeping {} (shadowed {})",
                    c.resource_type.as_str(),
                    c.name,
                    c.winner_path,
                    c.loser_path
                ));
            }
        }
    }

    // ---- Re-compose the system prompt (same precedence as build) ----
    // Same working-state wording rule as `build`; a mid-session `/model` switch
    // re-composes the prompt on the next reload, which is the same staleness the
    // rest of this path already has (provider resolution is a startup concern).
    // Re-merge extension tools (a reloaded plugin may have added/removed a
    // tool). The built-in set is rebuilt from scratch + extension tools merged
    // on top, mirroring `build`. Built before the
    // prompt so the prompt lists exactly the active tools.
    let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
    let tool_ctx = rpi_tools::ExecutionToolContext::new(env_dyn.clone(), Some(mut_env))
        .with_file_io(ctx.file_io.clone());
    let mut tools = build_tools(&tool_ctx, &effective_args);
    merge_extension_tools(
        &mut tools,
        &extension_session,
        &effective_args,
        &ctx.tool_context,
    );
    let mut active = active_tool_names(&tools, &effective_args);
    active = filter_active_tool_names(active, &effective_args);
    let prompt_tools = active.clone();
    let prompt_metadata = extension_tool_prompt_metadata(&extension_session);
    let prior_reasoning_replayed = rpi_ai::model::prior_reasoning_is_replayed(&ctx.resolved_model);
    let default_prompt = || {
        default_system_prompt_with_tool_prompts(
            &cwd_str,
            prior_reasoning_replayed,
            &prompt_tools,
            &prompt_metadata,
        )
    };
    let mut context_paths = Vec::new();
    let base_prompt = match effective_args.system_prompt.as_deref() {
        Some(explicit) => explicit.to_string(),
        None if project_trusted => match discover_system_prompt_file(&ctx.cwd) {
            Some(path) => {
                read_prompt_source(&path, &mut context_paths).unwrap_or_else(default_prompt)
            }
            None => default_prompt(),
        },
        None => default_prompt(),
    };
    let mut append_texts: Vec<String> = Vec::new();
    for extra in &effective_args.append_system_prompt {
        let text = read_append_target(extra, &mut context_paths).unwrap_or_else(|| extra.clone());
        append_texts.push(text);
    }
    if effective_args.append_system_prompt.is_empty() {
        if let Some(path) = project_trusted
            .then(|| discover_append_system_prompt_file(&ctx.cwd))
            .flatten()
        {
            if let Some(text) = read_prompt_source(&path, &mut context_paths) {
                append_texts.push(text);
            }
        }
    }
    context_paths.extend(
        context_files
            .iter()
            .map(|file| file.path.to_string_lossy().into_owned()),
    );
    let append_join = if append_texts.is_empty() {
        None
    } else {
        Some(append_texts.join("\n\n"))
    };
    let system_prompt = compose_system_prompt(
        Some(&base_prompt),
        &[],
        if context_block.is_empty() {
            None
        } else {
            Some(&context_block)
        },
        append_join.as_deref(),
    );

    // ---- Rebuild the emitter (TeeEmitter over fresh ExtensionEmitter) ----
    let emitter: Arc<dyn rpi_agent::AgentEmitter> = match extension_session.snapshot_arc() {
        Some(snapshot) => {
            let ext = ExtensionEmitter::new(snapshot, extension_session.keepalive());
            Arc::new(TeeEmitter::new(vec![ctx.broadcast.clone(), Arc::new(ext)]))
        }
        None => ctx.broadcast.clone(),
    };

    // ---- 5. Push the rebuilt state into the live harness via the B5d setters ----
    let resources = AgentHarnessResources {
        context_files: context_paths,
        skills: if skills.is_empty() {
            None
        } else {
            Some(skills.clone())
        },
        prompt_templates: if prompt_templates.is_empty() {
            None
        } else {
            Some(prompt_templates.clone())
        },
    };
    let _ = harness.set_system_prompt(Some(system_prompt)).await;
    let _ = harness.set_resources(resources).await;
    let _ = harness.set_agent_emitter(Some(emitter)).await;
    let _ = harness
        .set_models(build_models_with_extensions_for_reload(
            &refreshed
                .as_ref()
                .map(|resolved| resolved.providers.clone())
                .unwrap_or_else(|| {
                    ctx.providers
                        .read()
                        .unwrap_or_else(|error| error.into_inner())
                        .clone()
                }),
            &extension_session,
            ctx.runtime.clone(),
        ))
        .await;
    let _ = harness
        .set_provider_hooks(
            rpi_extensions::ExtensionProviderHooks::from_session(&extension_session)
                .map(|h| Arc::new(h) as Arc<dyn rpi_ai::ProviderHooks>),
        )
        .await;

    if let Some(resolved) = &refreshed {
        *ctx.providers
            .write()
            .unwrap_or_else(|error| error.into_inner()) = resolved.providers.clone();
        let catalog = crate::provider::authenticated_catalog(resolved);
        fresh_bridge
            .clone_host()
            .replace_model_catalog(catalog.clone());
        *ctx.catalog
            .write()
            .unwrap_or_else(|error| error.into_inner()) = catalog;
        // Refresh the public descriptor. Credentials remain in the newly bound
        // provider instances installed above and never enter the lane model.
        let _ = harness.lane("main").set_model(resolved.model.clone()).await;
    }

    let _ = harness.set_tools(tools, Some(active)).await;

    let summary = format!(
        "Reloaded {} plugin(s), {} skill(s), {} prompt(s).",
        extension_session.loaded_paths().len(),
        skills.len(),
        prompt_templates.len(),
    );
    ReloadOutcome {
        summary,
        had_warnings: warnings,
        details,
        // Reload completed; any diagnostics here are soft (shadowing / name
        // validation / skipped optional resources), not hard failures.
        had_errors: false,
        project_trusted: Some(project_trusted),
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct ReloadSettingsFields {
    skill_dirs: Option<Vec<String>>,
    prompt_dirs: Option<Vec<String>>,
    extension_dirs: Option<Vec<String>>,
}

impl From<crate::settings::Settings> for ReloadSettingsFields {
    fn from(settings: crate::settings::Settings) -> Self {
        Self {
            skill_dirs: settings.skill_dirs,
            prompt_dirs: settings.prompt_dirs,
            extension_dirs: settings.extension_dirs,
        }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
struct ReloadSettingsSnapshot {
    global: ReloadSettingsFields,
    project: Option<ReloadSettingsFields>,
}

fn reload_reads_settings(args: &Args) -> bool {
    !args.dev_local_only && (!args.no_extensions || !args.no_skills || !args.no_prompt_templates)
}

/// Strictly read the settings fields consumed while preparing a reload. The
/// snapshot intentionally excludes UI/model fields that `/reload` does not
/// use, so an unrelated settings save does not invalidate the operation.
fn load_reload_settings_snapshot(
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
) -> Result<Option<ReloadSettingsSnapshot>, String> {
    if !reload_reads_settings(args) {
        return Ok(None);
    }
    let global = crate::settings::load_settings()
        .map_err(|error| format!("could not load global settings: {error}"))?
        .into();
    let project = if project_trusted {
        crate::settings::load_active_project_settings(cwd)
            .map_err(|error| format!("could not load project settings: {error}"))?
            .map(|(_, settings)| settings.into())
    } else {
        None
    };
    Ok(Some(ReloadSettingsSnapshot { global, project }))
}

/// Validate every settings document that this reload will consume before
/// replacing any live extension, bridge, or harness resource.
#[cfg(test)]
fn validate_settings_for_reload(
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
) -> Result<(), String> {
    load_reload_settings_snapshot(args, cwd, project_trusted).map(|_| ())
}

fn prepare_reload_inputs(
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
) -> Result<PreparedReloadInputs, String> {
    prepare_reload_inputs_inner(args, cwd, project_trusted, || {})
}

fn prepare_reload_inputs_inner<F>(
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
    before_verify: F,
) -> Result<PreparedReloadInputs, String>
where
    F: FnOnce(),
{
    let settings_before = load_reload_settings_snapshot(args, cwd, project_trusted)?;

    let mut extension_dirs = if args.no_extensions || args.dev_local_only {
        Vec::new()
    } else if args.project_only {
        project_extension_dirs(cwd)
    } else if project_trusted {
        extension_dirs(cwd)
    } else {
        global_extension_dirs()
    };
    if !args.no_extensions {
        extension_dirs.extend(args.extensions_dir.iter().cloned());
    }
    remove_global_native_package_dir(&mut extension_dirs);
    remove_project_package_dirs(cwd, &mut extension_dirs);
    let mut extension_files = if args.no_extensions {
        Vec::new()
    } else {
        args.extension.clone()
    };
    if !args.no_extensions && !args.dev_local_only {
        if project_trusted || args.project_only {
            extension_files.extend(crate::packages::active_project_package_artifacts(cwd));
        }
        if !args.project_only {
            extension_files.extend(crate::install::active_global_native_package_artifacts());
            extension_files.extend(crate::install::unregistered_global_native_package_artifacts());
        }
    }

    let skill_base_dirs = if args.no_skills {
        Vec::new()
    } else if args.dev_local_only || args.project_only {
        project_skill_dirs(cwd)
    } else if project_trusted {
        skill_dirs(cwd)
    } else {
        global_skill_dirs()
    };

    let prompt_base_dirs = if args.no_prompt_templates {
        Vec::new()
    } else if args.dev_local_only || args.project_only {
        project_prompt_template_dirs(cwd)
    } else if project_trusted {
        prompt_template_dirs(cwd)
    } else {
        global_prompt_template_dirs()
    };

    before_verify();
    let settings_after = load_reload_settings_snapshot(args, cwd, project_trusted)?;
    if settings_before != settings_after {
        return Err(
            "settings changed while reload inputs were being prepared; retry /reload".to_string(),
        );
    }

    Ok(PreparedReloadInputs {
        extension_dirs,
        extension_files,
        skill_base_dirs,
        prompt_base_dirs,
    })
}

/// `build_models_with_extensions` for the reload path: the resolved provider
/// runtimes (NOT `resolved` — the reload context carries the `Arc<dyn Provider>`
/// list directly, since provider/auth did not change) first, then one
/// `PluggableProvider` per registered extension provider in the fresh session.
fn build_models_with_extensions_for_reload(
    providers: &[Arc<dyn Provider>],
    extension_session: &ExtensionSession,
    runtime: tokio::runtime::Handle,
) -> Vec<Arc<dyn Provider>> {
    let mut models: Vec<Arc<dyn Provider>> = providers.to_vec();
    let pluggable = rpi_extensions::PluggableProvider::from_session(extension_session, runtime);
    for provider in pluggable {
        // Declared packages must enter through credential-bound factories.
        // Falling back to a raw adapter would reintroduce packages rejected by
        // auth filtering and would discard their configured model snapshot.
        if extension_session
            .snapshot()
            .is_some_and(|snapshot| snapshot.provider_manifest(provider.id()).is_some())
        {
            continue;
        }
        if !models.iter().any(|known| known.id() == provider.id()) {
            models.push(provider);
        }
    }
    models
}

/// Diagnostic for registered TUI renderers. All three renderer kinds are
/// consumed by the interactive TUI's JSON component adapter; this line remains
/// useful under `--verbose` for extension authors.
fn report_deferred_renderers(session: &ExtensionSession) {
    let Some(snap) = session.snapshot_arc() else {
        return;
    };
    let all = snap.renderers();
    let markdown = all
        .iter()
        .filter(|r| r.kind == rpi_extensions::RegisteredRendererKind::Markdown)
        .count();
    let message = all
        .iter()
        .filter(|r| r.kind == rpi_extensions::RegisteredRendererKind::Message)
        .count();
    let entry = all
        .iter()
        .filter(|r| r.kind == rpi_extensions::RegisteredRendererKind::Entry)
        .count();
    if markdown + message + entry == 0 {
        return;
    }
    eprintln!(
        "renderers: {} markdown-transform, {} message-render, {} entry-render (active)",
        markdown, message, entry
    );
}

/// A harness-build error.
#[derive(Debug, thiserror::Error)]
pub enum BuildError {
    #[error("Experimental native session policy: {0}")]
    NativeStorePolicy(String),
    #[error("Could not create the session directory: {0}")]
    SessionDir(String),
    #[error("Could not open the saved session: {0}")]
    SessionOpen(String),
    #[error("No session found for {requested} in {dir}. Start a fresh session instead (drop --continue/--resume/--session).")]
    SessionNotFound { requested: String, dir: String },
    #[error("Could not build the harness: {0}")]
    HarnessCreate(String),
}

/// Keep configured factories (including package-specific auth and catalogs)
/// first, then append legacy plugin registrations that have no runtime yet.
fn build_models_with_extensions(
    resolved: &ResolvedModel,
    extension_session: &ExtensionSession,
    runtime: tokio::runtime::Handle,
) -> Vec<Arc<dyn Provider>> {
    build_models_with_extensions_for_reload(&resolved.providers, extension_session, runtime)
}

/// Resolve project resource loading without prompting. Explicit CLI overrides
/// win, then a stored decision; projects with no decision load by default.
pub(crate) fn resolve_project_trust(args: &Args, cwd: &Path) -> bool {
    if let Some(override_value) = args.trust_override {
        return override_value;
    }
    crate::config::project_trust_decision(cwd)
        .ok()
        .flatten()
        .unwrap_or(true)
}

/// Resolve the extension dirs to scan and load the cdylib plugins, returning
/// the loaded session guard (keeps the `Library` handles alive for the harness
/// lifetime). Scan order: configured project paths, project `.rpi/extensions`,
/// configured/global conventional paths, then any
/// `--extensions-dir` flags (scanned after the defaults — `args.rs`).
/// Diagnostics are a no-op sink for now; load skips/ABI mismatches surface via
/// the `--verbose` summary.
pub(crate) fn load_extensions(
    args: &Args,
    cwd: &Path,
    project_trusted: bool,
    action_bridge: Option<Arc<rpi_extensions::ActionBridge>>,
) -> ExtensionSession {
    if args.no_extensions {
        return ExtensionSession::none();
    }
    let mut dirs = if args.dev_local_only {
        Vec::new()
    } else if args.project_only {
        project_extension_dirs(cwd)
    } else if project_trusted {
        extension_dirs(cwd)
    } else {
        global_extension_dirs()
    };
    dirs.extend(args.extensions_dir.iter().cloned());
    remove_global_native_package_dir(&mut dirs);
    remove_project_package_dirs(cwd, &mut dirs);
    let mut files = args.extension.clone();
    if !args.dev_local_only {
        if project_trusted || args.project_only {
            files.extend(crate::packages::active_project_package_artifacts(cwd));
        }
        if !args.project_only {
            files.extend(crate::install::active_global_native_package_artifacts());
            files.extend(crate::install::unregistered_global_native_package_artifacts());
        }
    }
    let leases = match crate::resource_management::host_leases(cwd) {
        Ok(leases) => leases,
        Err(error) => {
            eprintln!("warning: refusing extension loading: {error}");
            return ExtensionSession::none();
        }
    };
    load_extensions_from_files(&dirs, &files, action_bridge, Some(leases))
}

fn remove_global_native_package_dir(dirs: &mut Vec<PathBuf>) {
    let Ok(agent_dir) = crate::config::agent_dir() else {
        return;
    };
    let native_dir = agent_dir.join("extensions");
    dirs.retain(|dir| dir != &native_dir);
}

fn remove_project_package_dirs(cwd: &Path, dirs: &mut Vec<PathBuf>) {
    let package_dirs = crate::packages::project_package_dirs(cwd, "extensions");
    dirs.retain(|dir| !package_dirs.iter().any(|package_dir| dir == package_dir));
}

fn load_extensions_from_files(
    dirs: &[PathBuf],
    files: &[PathBuf],
    action_bridge: Option<Arc<rpi_extensions::ActionBridge>>,
    host_guard: Option<Arc<dyn Send + Sync>>,
) -> ExtensionSession {
    let diagnostics: Arc<dyn PluginDiagnostics> = Arc::new(NullDiagnostics);
    // B5a: the action bridge is cloned into every loaded plugin's vtable
    // `user_data` so post-register `runtime_action` calls recover the harness
    // host from any thread. Directory scanning remains for project/configured
    // extensions; installed global packages are supplied as explicit active
    // artifact paths so retired versions are never loaded accidentally.
    rpi_extensions::load_session_mixed_guarded(dirs, files, diagnostics, action_bridge, host_guard)
}

/// Merge the loaded extension tools into the built-in set. An extension tool
/// overrides a same-named built-in; first-extension-wins across plugins is
/// already guaranteed by the registry (`register_tool` keeps the prior). The
/// explicit `--tools` allowlist / `--exclude-tools` denylist apply to the
/// merged set (the built-ins were already filtered in [`build_tools`]).
fn merge_extension_tools(
    tools: &mut Vec<HarnessTool>,
    session: &ExtensionSession,
    args: &Args,
    tool_context: &rpi_extensions::ToolCallContext,
) {
    let Some(snapshot) = session.snapshot() else {
        return;
    };
    for et in snapshot.tools() {
        let name = &et.tool.name;
        if !tool_name_allowed(name, args) {
            continue;
        }
        let adapter = PluginToolAdapter::new(et.tool.clone(), et.handle(), session.keepalive())
            .with_context(tool_context.clone());
        let harness_tool = HarnessTool::new(Arc::new(adapter));
        match tools.iter_mut().find(|t| t.tool.schema().name == *name) {
            Some(slot) => *slot = harness_tool,
            None => tools.push(harness_tool),
        }
    }
    // `tool_result` transforms see every tool's result, built-in or plugin.
    // Without one, the tools stay unwrapped.
    if let Some(snapshot) = session.snapshot_arc() {
        if !snapshot.tool_result().is_empty() {
            for tool in tools.iter_mut() {
                tool.tool = rpi_extensions::with_tool_result_transform(
                    Arc::clone(&tool.tool),
                    Arc::clone(&snapshot),
                    session.keepalive(),
                );
            }
        }
    }
}

/// Build the tool list per `--tools`/`--exclude-tools`/`--no-tools`/
/// `--no-builtin-tools`. Mirrors the TS `tools`/`excludeTools`/`noTools`
/// resolution in `createAgentSession`.
/// Default bash timeout: 120s when the model doesn't pass one (prevents a
/// forgotten `timeout` from hanging the run forever — the "卡住" report).
/// `RPI_BASH_TIMEOUT` overrides; a model-supplied timeout always wins.
pub fn bash_options() -> rpi_tools::tools::bash::BashToolOptions {
    use rpi_tools::tools::bash::BashToolOptions;
    let default = std::env::var("RPI_BASH_TIMEOUT")
        .ok()
        .and_then(|v| v.parse::<f64>().ok())
        .unwrap_or(rpi_tools::tools::bash::DEFAULT_BASH_TIMEOUT_SECONDS);
    // Saved user defaults, matching how `defaultTools` is read in `build_tools`.
    // `shellCommandPrefix` is prepended to every command (upstream's key for
    // alias/`shopt` setup). Native's `shellPath` is *not* wired: neither
    // `BashToolOptions` nor `ShellCaptureOptions` carries a shell binary today, so
    // honouring it means plumbing one through pi-tools first.
    let settings = crate::settings::load_settings().unwrap_or_default();
    BashToolOptions {
        command_prefix: settings.shell_command_prefix.clone(),
        default_timeout: Some(default),
        prepare: None,
    }
}

fn build_tools(ctx: &ExecutionToolContext, args: &Args) -> Vec<HarnessTool> {
    if args.no_tools {
        return Vec::new();
    }
    // Keep the coding tools aligned with Pi and expose rpi's read-only docs
    // lookup as a default assistant capability.
    //
    // Replay policy (native `ToolDeclaration.replay`): only tools that are
    // genuinely free of side effects may be re-run when a crash leaves their
    // outcome unknown — everything else must stay `Never`, because replaying
    // `bash`/`write`/`edit` would duplicate a side effect. `HarnessTool::new`
    // already defaults to `Never`, so this is opt-in per read-only tool.
    let mut all: Vec<(&'static str, HarnessTool)> = vec![
        (
            "read",
            HarnessTool::new(create_read_tool(
                ctx,
                Some(rpi_tools::tools::read::ReadToolOptions {
                    image_processor: Some(Arc::new(
                        rpi_tools::image_processing::InlineImageProcessor,
                    )),
                    ..Default::default()
                }),
            ))
            .with_replay(ToolReplay::Safe),
        ),
        (
            "bash",
            HarnessTool::new(create_bash_tool(ctx, Some(bash_options()))),
        ),
        ("edit", HarnessTool::new(create_edit_tool(ctx))),
        ("write", HarnessTool::new(create_write_tool(ctx))),
        (
            "docs",
            HarnessTool::new(create_docs_tool()).with_replay(ToolReplay::Safe),
        ),
    ];

    // Add PowerShell tool on Windows
    #[cfg(target_os = "windows")]
    all.push(("powershell", HarnessTool::new(create_powershell_tool(ctx))));

    // Apply default_tools from settings (if no CLI override)
    if args.tools.is_none() && args.resource_config.is_none() && args.resource_transition.is_none()
    {
        if let Ok(settings) = crate::settings::load_settings() {
            if let Some(default_tools) = &settings.default_tools {
                all.retain(|(name, _)| default_tools.iter().any(|t| t == name));
            }
        }
    }

    // `--no-builtin-tools` disables the built-in set but would keep
    // extension/custom tools — v1 has none, so it's equivalent to `--no-tools`
    // here. We honor it by clearing the built-ins.
    if args.no_builtin_tools {
        all.clear();
    }

    // Allowlist (`--tools`): keep only named built-ins.
    if let Some(allow) = &args.tools {
        all.retain(|(name, _)| allow.iter().any(|a| a == name));
    }
    // Denylist (`--exclude-tools`): drop named tools.
    if let Some(deny) = &args.exclude_tools {
        all.retain(|(name, _)| !deny.iter().any(|d| d == name));
    }

    all.into_iter().map(|(_, t)| t).collect()
}

/// Resolve the active tool names from the constructed tools when no explicit
/// `--tools` allowlist was given. Mirrors the TS default: all registered tools
/// active.
fn active_tool_names(tools: &[HarnessTool], args: &Args) -> Vec<String> {
    filter_active_tool_names(
        tools.iter().map(|tool| tool.tool.schema().name.clone()),
        args,
    )
}

/// Whether a tool name survives the command-line tool policy. Keep this check
/// centralized so extension-provided tools honor the same policy as built-ins.
pub(crate) fn tool_name_allowed(name: &str, args: &Args) -> bool {
    if args.no_tools {
        return false;
    }
    if args
        .tools
        .as_ref()
        .is_some_and(|allow| !allow.iter().any(|value| value == name))
    {
        return false;
    }
    if args
        .exclude_tools
        .as_ref()
        .is_some_and(|deny| deny.iter().any(|value| value == name))
    {
        return false;
    }
    true
}

pub(crate) fn filter_active_tool_names<I>(names: I, args: &Args) -> Vec<String>
where
    I: IntoIterator<Item = String>,
{
    names
        .into_iter()
        .filter(|name| tool_name_allowed(name, args))
        .collect()
}

/// Build the `Session` facade for the chosen selection.
async fn build_session(
    selection: &SessionSelection,
    cwd: &str,
    selected: Option<&crate::resource_selection::ResolvedSelection>,
) -> Result<Session, BuildError> {
    match selection {
        SessionSelection::Ephemeral => Ok(ephemeral_session()),
        SessionSelection::ExperimentalNative { dir } => Ok(Session::new(
            backend::fresh_experimental_native(dir).await?,
            None,
        )),
        SessionSelection::New { dir, .. } => {
            // Ensure the sessions directory exists, then create a fresh JSONL
            // session file inside it.
            std::fs::create_dir_all(dir)
                .map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
            let session = create_jsonl_session(dir, cwd)
                .await
                .map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
            Ok(session)
        }
        SessionSelection::Latest { .. }
        | SessionSelection::ById { .. }
        | SessionSelection::ByExactId { .. } => restore_session(selection, cwd).await,
        SessionSelection::Fork { source, dir } => {
            fork_session_at_launch(source, dir, cwd, selected).await
        }
    }
}

/// Open an existing JSONL session for `Latest` / `ById`. Mirrors the TS
/// `SessionManager.resume`/`open` flow: list the session dir (newest-first),
/// match the request, then open the matched file and wrap it in a `Session`
/// facade. The restored transcript renders into the TUI at startup and the
/// harness continues appending to the same file.
async fn restore_session(selection: &SessionSelection, cwd: &str) -> Result<Session, BuildError> {
    // `resolve_restore_target` is newest-first; `Latest` takes the head, `ById`
    // matches the id exactly or by file-name containment (so `--session 01a02…`
    // or a partial id works, mirroring the TS id/path matching).
    let Some(meta) = resolve_restore_target(selection, cwd).await? else {
        return match selection {
            // Exact id match only (pi `--session-id`): a missing session is
            // created fresh under the selected dir instead of an error.
            SessionSelection::ByExactId { id, dir } => {
                std::fs::create_dir_all(dir)
                    .map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
                create_jsonl_session_with_id(dir, cwd, Some(id.clone()))
                    .await
                    .map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))
            }
            SessionSelection::Latest { dir } => Err(BuildError::SessionNotFound {
                requested: "the most recent session".to_string(),
                dir: dir.display().to_string(),
            }),
            SessionSelection::ById { id, dir } => Err(BuildError::SessionNotFound {
                requested: format!("session {id}"),
                dir: dir.display().to_string(),
            }),
            _ => unreachable!("resolve_restore_target only returns None for restore selections"),
        };
    };
    let sessions_root = selection_dir(selection);
    let writer_restore = StartupSpan::start(Phase::WriterRestore);
    let result = open_session(&meta, cwd, &sessions_root)
        .await
        .map(|restored| restored.session);
    if result.is_ok() {
        writer_restore.finish();
    }
    result
}

/// The sessions root a selection was resolved against.
fn selection_dir(selection: &SessionSelection) -> PathBuf {
    match selection {
        SessionSelection::Ephemeral => PathBuf::new(),
        SessionSelection::New { dir, .. }
        | SessionSelection::ExperimentalNative { dir }
        | SessionSelection::Latest { dir }
        | SessionSelection::ById { dir, .. }
        | SessionSelection::ByExactId { dir, .. }
        | SessionSelection::Fork { dir, .. } => dir.clone(),
    }
}

/// Resolve the metadata a restore selection names, without opening the file.
///
/// Shared by the startup build ([`restore_session`]) and the cwd peek
/// ([`peek_restored_session_cwd`]) so the two can never disagree about which
/// session a launch restores. Pure listing + matching: no session lock is
/// claimed and nothing is created.
async fn resolve_restore_target(
    selection: &SessionSelection,
    cwd: &str,
) -> Result<Option<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
    match selection {
        SessionSelection::Latest { dir } => {
            let metas = list_session_metadata_in_dir(cwd, dir).await?;
            Ok(first_resumable_session(&metas).cloned())
        }
        SessionSelection::ById { id, dir } => find_session_metadata(id, cwd, dir).await,
        SessionSelection::ByExactId { id, dir } => {
            let metas = list_session_metadata_in_dir(cwd, dir).await?;
            Ok(metas.into_iter().find(|m| m.id == *id))
        }
        SessionSelection::Fork { .. }
        | SessionSelection::ExperimentalNative { .. }
        | SessionSelection::New { .. }
        | SessionSelection::Ephemeral => Ok(None),
    }
}

/// The metadata of the session whose id matches exactly or by file-name
/// containment (so `--session 01a02…` / a partial id / a full file name all
/// work), or `None` when nothing matches.
async fn find_session_metadata(
    id: &str,
    cwd: &str,
    dir: &Path,
) -> Result<Option<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
    let metas = list_session_metadata_in_dir(cwd, dir).await?;
    Ok(metas
        .into_iter()
        .find(|m| m.id == id || m.path.contains(id) || id.contains(&m.id)))
}

/// What a launch that would restore a session says about the working directory.
pub(crate) struct RestoredSessionCwd {
    /// The id of the session this launch will restore. The host pins it into
    /// `args` so the later build opens exactly this session — a directory
    /// listing taken twice can disagree (another `rpi` may write a newer
    /// session in between), and adopting one session's directory while opening
    /// another is the very bug this restores.
    pub id: String,
    /// The directory recorded in the session header.
    pub recorded: PathBuf,
    /// `Some` when the recorded directory still exists and the run adopts it;
    /// `None` when the caller must fall back to the launch cwd and say so.
    pub adopted: Option<PathBuf>,
}

/// The working directory the launch's restored session was created in, resolved
/// WITHOUT opening, locking, or creating anything.
///
/// Read-only by design: this runs before trust resolution and extension
/// preparation, where a failure must not take the run down — the real open
/// happens later and surfaces its own errors. Returns `None` when the selection
/// is not a restore (fresh/ephemeral/fork) or cannot be resolved.
pub(crate) async fn peek_restored_session_cwd(
    selection: &SessionSelection,
    launch_cwd: &Path,
) -> Option<RestoredSessionCwd> {
    let cwd_str = launch_cwd.to_string_lossy().into_owned();
    let meta = resolve_restore_target(selection, &cwd_str).await.ok()??;
    let recorded = PathBuf::from(&meta.cwd);
    if recorded.as_os_str().is_empty() {
        return None;
    }
    Some(RestoredSessionCwd {
        id: meta.id,
        adopted: crate::session_cwd::resolve_session_cwd(&recorded),
        recorded,
    })
}

/// Pin a peeked restore target into `args` so the later build opens exactly the
/// session the peek inspected.
///
/// `Latest` (`-c`) re-lists the session directory when the build runs, and
/// another `rpi` may have written a newer session in between. Adopting one
/// session's working directory while opening a different conversation would put
/// the tools in a directory that conversation never used — the exact failure
/// this adoption exists to prevent. Pinning the resolved id turns the second
/// listing into an exact match instead of a fresh "newest" choice.
pub(crate) fn pin_restored_selection(args: &mut Args, restored: &RestoredSessionCwd) {
    args.continue_session = false;
    args.resume = false;
    args.session = Some(restored.id.clone());
}

/// `--fork <path|id>`: open the source session, fork it into a new JSONL
/// session (records the parent id), and start in the fork.
async fn fork_session_at_launch(
    source: &str,
    dir: &Path,
    cwd: &str,
    selected: Option<&crate::resource_selection::ResolvedSelection>,
) -> Result<Session, BuildError> {
    use rpi_harness::session::jsonl::{
        JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
    };
    use rpi_harness::session::types::{ForkOptions, SessionStorage};
    use rpi_tools::FileSystem;

    let dir = dir.to_path_buf();
    std::fs::create_dir_all(&dir)
        .map_err(|e| BuildError::SessionDir(format!("{}: {e}", dir.display())))?;
    let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
    let fs: Arc<dyn FileSystem> = env.clone();
    let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
        fs: fs.clone(),
        sessions_root: dir.to_string_lossy().into_owned(),
        clock: Arc::new(SystemClock),
        ids: Arc::new(DefaultIdGenerator::new()),
    });
    // A session file given by path forks from wherever it lives (pi parity:
    // `pi --fork <file>` takes any session file); an id or a name fragment is
    // looked up in the session directory.
    let source_meta = match session_metadata_at(Path::new(source)) {
        Some(meta) => meta,
        None => {
            let metas = repo
                .list_typed(&rpi_harness::session::jsonl::JsonlSessionListOptions::default())
                .await
                .map_err(|e| BuildError::SessionDir(format!("list sessions: {e}")))?;
            metas
                .into_iter()
                .find(|m| m.id == *source || m.path.contains(source) || source.contains(&m.id))
                .ok_or_else(|| BuildError::SessionNotFound {
                    requested: format!("--fork {source}"),
                    dir: dir.display().to_string(),
                })?
        }
    };
    let fork_storage = repo
        .fork_typed_with_value_initializer(
            &source_meta,
            &JsonlSessionCreateOptions {
                id: None,
                parent_session_id: Some(source_meta.id.clone()),
                cwd: cwd.to_string(),
                metadata: None,
            },
            &ForkOptions::default(),
            |values| crate::resource_selection::fork_initial_values_for_selection(values, selected),
        )
        .await
        .map_err(|e| BuildError::SessionDir(format!("fork {}: {e}", source_meta.path)))?;
    claim_session_lock(&fork_storage.jsonl_metadata().path).map_err(BuildError::SessionOpen)?;
    let storage_arc: Arc<dyn SessionStorage> = Arc::new(fork_storage);
    Ok(Session::new(storage_arc, None))
}

/// The metadata of the JSONL session file at `path`, or `None` when `path` is
/// not a readable session file.
fn session_metadata_at(path: &Path) -> Option<rpi_harness::session::jsonl::JsonlSessionMetadata> {
    use rpi_harness::session::jsonl::codec::{metadata_from_header, parse_header};
    use std::io::BufRead;

    let file = std::fs::File::open(path).ok()?;
    let modified_at = file
        .metadata()
        .ok()?
        .modified()
        .ok()?
        .duration_since(std::time::UNIX_EPOCH)
        .ok()?
        .as_millis() as i64;
    let mut first_line = String::new();
    std::io::BufReader::new(file)
        .read_line(&mut first_line)
        .ok()?;
    let header = parse_header(first_line.trim_end()).ok()?;
    let path = std::fs::canonicalize(path).ok()?;
    Some(metadata_from_header(
        &header,
        &path.to_string_lossy(),
        modified_at,
    ))
}

/// Errors from [`open_session_by_id`], split so the CLI can map them to
/// [`BuildError`] while the TUI can surface a friendlier note.
pub enum OpenError {
    /// No session matched the request.
    NotFound { requested: String },
    /// The match existed but could not be opened/parsed.
    Other(String),
}

impl std::fmt::Display for OpenError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            OpenError::NotFound { requested } => write!(f, "no session matches {requested}"),
            // `Other` carries the bare reason (see `open_session_by_id`); the
            // prefix lives here so the TUI and the CLI's
            // `BuildError::SessionOpen` wrapper each render it exactly once
            // (wrapping the already-prefixed `BuildError` string doubled it).
            OpenError::Other(msg) => write!(f, "Could not open the saved session: {msg}"),
        }
    }
}

/// List the JSONL session metadata under the default session dir, newest
/// first. Shared by startup restore and the TUI `/session` hot-switch.
pub async fn list_session_metadata(
    cwd: &str,
) -> Result<Vec<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
    let dir = default_session_dir(Path::new(cwd));
    list_session_metadata_in_dir(cwd, &dir).await
}

pub(crate) async fn list_session_metadata_in_dir(
    cwd: &str,
    dir: &Path,
) -> Result<Vec<rpi_harness::session::jsonl::JsonlSessionMetadata>, BuildError> {
    use rpi_harness::session::jsonl::{
        JsonlSessionListOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
    };
    use rpi_tools::FileSystem;

    let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
    let fs: Arc<dyn FileSystem> = env.clone();
    let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
        fs: fs.clone(),
        sessions_root: dir.to_string_lossy().into_owned(),
        clock: Arc::new(SystemClock),
        ids: Arc::new(DefaultIdGenerator::new()),
    });
    let metadata_lookup = StartupSpan::start(Phase::MetadataLookup);
    let result = repo
        .list_typed(&JsonlSessionListOptions::default())
        .await
        .map_err(|e| BuildError::SessionDir(format!("list sessions: {e}")));
    if result.is_ok() {
        metadata_lookup.finish();
    }
    result
}

/// A cheap, display-oriented summary of one JSONL session file.
///
/// The session pickers (`-r` and `/session`) need enough context to tell saved
/// sessions apart. The header-only [`JsonlSessionMetadata`] carries no name and
/// no transcript, so a list built from it renders as a wall of identical
/// `2026-09-25T15-05-15-847Z_…` file names — including the many header-only
/// sessions an abandoned launch leaves behind. This reads just the head (and a
/// small tail, for a late `--name`/`/name` fact) of the file to recover:
///
/// - the session display name, when one was recorded;
/// - the first user prompt, single-lined and truncated, for a readable label;
/// - whether the file is header-only (nothing to resume).
///
/// It intentionally does **not** parse the whole log: sessions can be tens of
/// megabytes, and the picker runs over every session at startup.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SessionSummary {
    /// The session display name (`--name` / `/name`), if one was recorded.
    pub name: Option<String>,
    /// The first user prompt, single-lined and trimmed to a label-sized length.
    pub preview: Option<String>,
    /// True when the file holds only its header (a session with no content).
    pub empty: bool,
    /// On-disk size in bytes (a rough content-volume hint for the picker).
    pub bytes: u64,
}

/// Bytes read from the head of a session file to recover a name + first prompt.
const SUMMARY_HEAD_BYTES: u64 = 256 * 1024;
/// Bytes read from the tail of a session file to recover a late name fact.
const SUMMARY_TAIL_BYTES: u64 = 8 * 1024;
/// Longest preview retained (characters, not bytes).
const SUMMARY_PREVIEW_CHARS: usize = 80;

/// Summarize one session file for the pickers. Best-effort: an unreadable or
/// malformed file yields a summary with `empty: false` and no label, so it is
/// still selectable rather than silently dropped.
pub fn summarize_session_file(path: &Path) -> SessionSummary {
    use std::io::{Read, Seek, SeekFrom};

    let Ok(mut file) = std::fs::File::open(path) else {
        return SessionSummary::default();
    };
    let Ok(size) = file.metadata().map(|m| m.len()) else {
        return SessionSummary::default();
    };
    if size == 0 {
        return SessionSummary {
            name: None,
            preview: None,
            empty: true,
            bytes: 0,
        };
    }

    // Head: the header line (for the empty check) + a name fact + the first user
    // prompt. The whole window is scanned; it is already in memory.
    let mut head = vec![0u8; SUMMARY_HEAD_BYTES.min(size) as usize];
    let head_len = file.read(&mut head).unwrap_or(0);
    head.truncate(head_len);
    let head = String::from_utf8_lossy(&head);
    let mut lines: Vec<&str> = head.lines().collect();
    let header_len = lines.first().map(|line| line.len()).unwrap_or(0);
    // A header-only file is exactly the header plus an optional trailing `\n`.
    let empty = (size as usize) <= header_len + 1;
    // When the window stopped short of the file end, its final line is partial:
    // drop it so a half-read message can never be mistaken for a complete one.
    if (head_len as u64) < size && !lines.is_empty() {
        lines.pop();
    }

    let mut summary = SessionSummary {
        name: None,
        preview: None,
        empty,
        bytes: size,
    };
    for line in lines {
        apply_summary_line(line, &mut summary);
    }

    // Tail: a name set late in the session lives after the head window. Read the
    // last window and scan only complete lines so a torn/partial first line is
    // ignored. Only do this when the file is larger than the head window.
    if size > SUMMARY_HEAD_BYTES
        && file
            .seek(SeekFrom::End(-(SUMMARY_TAIL_BYTES as i64)))
            .is_ok()
    {
        let mut tail = Vec::new();
        if file.read_to_end(&mut tail).is_ok() {
            let tail = String::from_utf8_lossy(&tail);
            let mut tail_lines = tail.lines();
            // Drop the possibly-partial first line of the window.
            tail_lines.next();
            for line in tail_lines {
                if let Some(name) = name_from_line(line) {
                    summary.name = name;
                }
            }
        }
    }
    summary
}

/// Extract the name from a `{"kind":"fact","fact":"name","name":…}` line.
/// Returns `Some(None)` for an explicit `name: null` (the fact was cleared).
fn name_from_line(line: &str) -> Option<Option<String>> {
    let value: serde_json::Value = serde_json::from_str(line).ok()?;
    if value.get("kind").and_then(|v| v.as_str()) != Some("fact") {
        return None;
    }
    if value.get("fact").and_then(|v| v.as_str()) != Some("name") {
        return None;
    }
    Some(match value.get("name") {
        Some(serde_json::Value::String(name)) => Some(name.clone()),
        _ => None,
    })
}

/// Fold one JSONL line into `summary`: record a name fact, or capture the first
/// user prompt as the preview.
fn apply_summary_line(line: &str, summary: &mut SessionSummary) {
    if let Some(name) = name_from_line(line) {
        summary.name = name;
        return;
    }
    if summary.preview.is_some() {
        return;
    }
    let Ok(value) = serde_json::from_str::<serde_json::Value>(line) else {
        return;
    };
    if value.get("kind").and_then(|v| v.as_str()) != Some("entry") {
        return;
    }
    if value.get("type").and_then(|v| v.as_str()) != Some("message") {
        return;
    }
    let Some(message) = value.get("message") else {
        return;
    };
    if message.get("role").and_then(|v| v.as_str()) != Some("user") {
        return;
    }
    let text = match message.get("content") {
        Some(serde_json::Value::String(text)) => Some(text.as_str()),
        Some(serde_json::Value::Array(blocks)) => blocks.iter().find_map(|block| {
            (block.get("type").and_then(|v| v.as_str()) == Some("text"))
                .then(|| block.get("text").and_then(|v| v.as_str()))
                .flatten()
        }),
        _ => None,
    };
    let Some(text) = text else { return };
    let collapsed = text
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty())
        .collect::<Vec<_>>()
        .join(" ");
    if collapsed.is_empty() {
        return;
    }
    summary.preview = Some(collapsed.chars().take(SUMMARY_PREVIEW_CHARS).collect());
}

/// The human label for a session row: its recorded name, else the first user
/// prompt, else a marker that says the file has no transcript to show. Shared
/// by the startup picker and the in-TUI `/session` selector.
pub fn session_display_label(summary: &SessionSummary) -> String {
    if let Some(name) = summary.name.as_deref().map(str::trim) {
        if !name.is_empty() {
            return name.to_string();
        }
    }
    match summary.preview.as_deref().map(str::trim) {
        Some(preview) if !preview.is_empty() => preview.to_string(),
        _ => "(no messages)".to_string(),
    }
}

/// Compact byte count (1.2MB / 34KB / 512B), shared by the startup picker and
/// the in-TUI `/session` selector.
pub fn format_session_bytes(bytes: u64) -> String {
    if bytes >= 1_048_576 {
        format!("{:.1}MB", bytes as f64 / 1_048_576.0)
    } else if bytes >= 1024 {
        format!("{}KB", bytes / 1024)
    } else {
        format!("{bytes}B")
    }
}

/// The first 8 characters of a session id, enough to disambiguate two rows that
/// share a timestamp without shouting the whole UUID.
pub fn short_session_id(id: &str) -> &str {
    &id[..id.len().min(8)]
}

/// Synchronously list every JSONL session file under `dir`, newest-first by
/// modification time.
///
/// The repo's [`list_session_metadata`] is async and scans one cwd-encoded
/// subdirectory level; the in-TUI `/session` selector runs on the blocking key
/// thread and cannot await it, so this walks the same layout directly. Both
/// `.jsonl` files directly under `dir` and one level of subdirectories are
/// accepted.
pub fn list_session_files_sync(dir: &Path) -> Vec<PathBuf> {
    let mut files: Vec<(PathBuf, std::time::SystemTime)> = Vec::new();
    let mut visit = |path: PathBuf| {
        if path.extension().and_then(|e| e.to_str()) != Some("jsonl")
            || path.to_string_lossy().ends_with(".progress.jsonl")
        {
            return;
        }
        let modified = std::fs::metadata(&path)
            .and_then(|m| m.modified())
            .unwrap_or(std::time::UNIX_EPOCH);
        files.push((path, modified));
    };
    if let Ok(entries) = std::fs::read_dir(dir) {
        for entry in entries.flatten() {
            let path = entry.path();
            if path.is_dir() {
                if let Ok(inner) = std::fs::read_dir(&path) {
                    for child in inner.flatten() {
                        visit(child.path());
                    }
                }
            } else {
                visit(path);
            }
        }
    }
    files.sort_by(|a, b| b.1.cmp(&a.1));
    files.into_iter().map(|(path, _)| path).collect()
}

/// Pick the session `--continue`/`-c` should open: the newest session that has
/// content (a header-only session resumes to an empty transcript). Falls back to
/// the newest overall when every session is empty, and `None` only when the
/// list is empty.
fn first_resumable_session(
    metas: &[rpi_harness::session::jsonl::JsonlSessionMetadata],
) -> Option<&rpi_harness::session::jsonl::JsonlSessionMetadata> {
    metas
        .iter()
        .find(|meta| !summarize_session_file(Path::new(&meta.path)).empty)
        .or_else(|| metas.first())
}

/// Open a session whose id matches exactly or by file-name containment
/// (so `--session 01a02…` / a partial id / a full file name all work). The
/// TUI `/session` hot-switch calls this with the selector's item value.
pub async fn open_session_by_id(id: &str, cwd: &str) -> Result<Session, OpenError> {
    let dir = default_session_dir(Path::new(cwd));
    open_session_by_id_in_dir(id, cwd, &dir)
        .await
        .map(|restored| restored.session)
}

pub(crate) async fn open_session_by_id_in_dir(
    id: &str,
    cwd: &str,
    dir: &Path,
) -> Result<RestoredSession, OpenError> {
    let meta = find_session_metadata(id, cwd, dir)
        .await
        .map_err(|e| OpenError::Other(e.to_string()))?;
    let Some(meta) = meta else {
        return Err(OpenError::NotFound {
            requested: format!("session {id}"),
        });
    };
    open_session(&meta, cwd, dir).await.map_err(|e| {
        OpenError::Other(match e {
            // `open_session` renders as "Could not open the saved session:
            // <reason>" via `BuildError`. Carry the bare reason so it is not
            // prefixed twice (once here, once by the CLI/TUI).
            BuildError::SessionOpen(reason) => reason,
            other => other.to_string(),
        })
    })
}

/// Publish the active session's id to plugins, on both channels that carry it.
///
/// Called on **every** activation of a session — the startup build AND every
/// in-process swap (`switch_to_session`, `fork_session`, and the plugin
/// `NewSession`/`Fork`/`SwitchSession` runtime actions all route through
/// [`set_active_session`]). Two things must stay in step:
///
/// - `ToolCallContext` — read per plugin **tool** call, which is why a stale
///   value there made a plugin's `todo`-style state follow the wrong session.
/// - `RPI_SESSION_ID` — read per **turn** by the provider hooks, which is the
///   only channel an observer gets for a pure-chat run. Process-global, which is
///   exactly the scope of "the session this process is serving".
///
/// The write overwrites unconditionally, deliberately. An inherited value is not
/// an override: `rpi` spawning `rpi` would otherwise hand the child the parent's
/// session id, and the child — which is *serving its own* session — would report
/// it forever. (Children no longer inherit the variable at all; see
/// `rpi_tools::HOST_PRIVATE_ENV`. This overwrite is the second line of defence,
/// for any other way a value can arrive pre-set.)
///
/// An embedder that genuinely wants to pin the id can revert to the old
/// behaviour by setting the variable *after* this runs; there is no supported
/// way to pin it before, because a session that does not match its own id is
/// never what the caller wants.
fn publish_session_identity(tool_context: &rpi_extensions::ToolCallContext, id: &str) {
    std::env::set_var(rpi_plugin_sdk::SESSION_ID_ENV, id);
    tool_context.set_session_id(id);
}

/// Swap the session the harness serves, republishing the identity to plugins.
///
/// The single funnel for an in-process session change. Prefer this over calling
/// `AgentHarness::set_session` directly: the harness holds no reference to the
/// plugin channels, so a bare `set_session` would leave every plugin reporting
/// the session it had before the swap.
///
/// `tool_context` is optional because not every caller has one (the harness's own
/// tests, a headless embedder); the env channel is always published, since the
/// provider hooks depend on it.
pub(crate) async fn set_active_session(
    harness: &AgentHarness,
    session: Session,
    tool_context: Option<&rpi_extensions::ToolCallContext>,
) -> Result<(), String> {
    let metadata = session.get_metadata().await.ok();
    harness
        .set_session(session)
        .await
        .map_err(|e| e.to_string())?;
    if let Some(metadata) = metadata {
        std::env::set_var(rpi_plugin_sdk::SESSION_ID_ENV, &metadata.id);
        if let Some(context) = tool_context {
            context.set_session_id(metadata.id);
        }
    }
    Ok(())
}

/// Fork the harness's current session into a new JSONL session (new id, parent
/// set to the source) and wrap it in a `Session`. Mirrors the TUI's
/// `fork_session` flow (`interactive_tui.rs`) — hoisted here so both the TUI
/// and the plugin `runtime_action(Fork)` host share one implementation.
/// Returns the new `Session` (NOT yet swapped onto the harness — the caller
/// does `harness.set_session(...)`).
pub(crate) async fn fork_session_storage(
    harness: &AgentHarness,
    cwd: &str,
    dir: &Path,
) -> Result<Session, String> {
    use rpi_harness::session::jsonl::{JsonlSessionRepo, JsonlSessionRepoOptions};
    use rpi_tools::FileSystem;

    let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
    let fs: Arc<dyn FileSystem> = env.clone();
    let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
        fs,
        sessions_root: dir.to_string_lossy().into_owned(),
        clock: Arc::new(SystemClock),
        ids: Arc::new(DefaultIdGenerator::new()),
    });
    // The fork needs the rich JSONL metadata (with the on-disk path); resolve
    // it from the session list by the current session's id.
    let id = harness.session().storage().metadata().id.clone();
    let metas = list_session_metadata_in_dir(cwd, dir)
        .await
        .map_err(|e| e.to_string())?;
    let Some(source) = metas.iter().find(|m| m.id == id) else {
        return Err(format!("current session {id} not found on disk"));
    };
    let fork_storage = repo
        .fork_typed_with_value_initializer(
            source,
            &rpi_harness::session::jsonl::JsonlSessionCreateOptions {
                id: None,
                parent_session_id: Some(source.id.clone()),
                cwd: cwd.to_string(),
                metadata: None,
            },
            &rpi_harness::session::types::ForkOptions::default(),
            crate::resource_selection::fork_initial_values,
        )
        .await
        .map_err(|e| e.to_string())?;
    claim_session_lock(&fork_storage.jsonl_metadata().path)?;
    Ok(Session::new(Arc::new(fork_storage), None))
}

/// Validate a private copy before publishing it in the repo's normal cwd
/// directory. Keep the imported identity/history and never overwrite a saved
/// session (including when the source is already inside this repository).
pub(crate) async fn import_session_file(
    source: &Path,
    cwd: &str,
    dir: &Path,
) -> Result<Session, String> {
    use rpi_harness::session::jsonl::{
        JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions, JsonlSessionStorage,
    };
    use rpi_harness::session::types::SessionStorage;
    use rpi_tools::FileSystem;

    let dir = if dir.is_absolute() {
        dir.to_path_buf()
    } else {
        Path::new(cwd).join(dir)
    };
    std::fs::create_dir_all(&dir).map_err(|e| e.to_string())?;
    // External writers must be closed. This process deliberately retains locks
    // for previously opened sessions, so its own copies additionally verify a
    // stable history around the sidecar snapshot (bind/entry before retirement).
    let source_progress = PathBuf::from(format!("{}.progress.jsonl", source.display()));
    let source_lock_path = PathBuf::from(format!("{}.lock", source.display()));
    let canonical_lock = std::fs::canonicalize(&source_lock_path).ok();
    let owned_here = SESSION_LOCK_PATHS.get().is_some_and(|paths| {
        paths.lock().unwrap().iter().any(|path| {
            path == &source_lock_path
                || canonical_lock.as_ref().is_some_and(|canonical| {
                    std::fs::canonicalize(path).ok().as_ref() == Some(canonical)
                })
        })
    });
    let _source_lock = if !owned_here && source_progress.try_exists().map_err(|e| e.to_string())? {
        let lock = std::fs::OpenOptions::new()
            .create(true)
            .write(true)
            .open(format!("{}.lock", source.display()))
            .map_err(|e| format!("lock import source: {e}"))?;
        fs2::FileExt::try_lock_exclusive(&lock).map_err(|_| {
            "close the source session before importing its recovery pair".to_owned()
        })?;
        Some(lock)
    } else {
        None
    };
    let staging = tempfile::tempdir_in(&dir).map_err(|e| e.to_string())?;
    let staged = staging.path().join("history.stage");
    let staged_progress = PathBuf::from(format!("{}.progress.jsonl", staged.display()));
    let mut attempts = 0;
    let has_progress = loop {
        std::fs::copy(source, &staged).map_err(|e| format!("copy import: {e}"))?;
        let copied_history = std::fs::read(&staged).map_err(|e| e.to_string())?;
        let has_progress = match std::fs::copy(&source_progress, &staged_progress) {
            Ok(_) => true,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                let _ = std::fs::remove_file(&staged_progress);
                false
            }
            Err(error) => return Err(format!("copy progress import: {error}")),
        };
        if copied_history == std::fs::read(source).map_err(|e| e.to_string())? {
            break has_progress;
        }
        attempts += 1;
        if attempts >= 4 {
            return Err(
                "source history changed during import; retry when the session is idle".into(),
            );
        }
        tokio::task::yield_now().await;
    };
    let fs: Arc<dyn FileSystem> = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
    let validated = JsonlSessionStorage::load(
        fs.clone(),
        &staged.to_string_lossy(),
        Arc::new(SystemClock),
        Arc::new(DefaultIdGenerator::new()),
    )
    .await
    .map_err(|e| format!("invalid session import: {e}"))?;
    validated
        .read_progress("")
        .await
        .map_err(|e| format!("invalid recovery import: {e}"))?;
    let mut metadata = validated.jsonl_metadata().clone();
    drop(validated);
    let text = std::fs::read_to_string(&staged).map_err(|e| e.to_string())?;
    let (first, rest) = text
        .split_once('\n')
        .ok_or_else(|| "import has no complete header".to_owned())?;
    let mut header =
        rpi_harness::session::jsonl::codec::parse_header(first).map_err(|e| e.to_string())?;
    if has_progress && header.progress_format.is_none() {
        return Err("history-only artifact has an undeclared recovery companion".into());
    }
    // An imported history-only artifact receives the destination's empty store.
    // Imports of live-format sessions retain their validated companion instead.
    header.progress_format = Some("assistant-progress-v1".into());
    std::fs::write(
        &staged,
        format!(
            "{}{}",
            rpi_harness::session::jsonl::codec::encode_header(&header),
            rest
        ),
    )
    .map_err(|e| e.to_string())?;
    if list_session_metadata_in_dir(cwd, &dir)
        .await
        .map_err(|e| e.to_string())?
        .iter()
        .any(|saved| saved.id == metadata.id)
    {
        return Err(format!(
            "session {} already exists in {}",
            metadata.id,
            dir.display()
        ));
    }
    let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
        fs,
        sessions_root: dir.to_string_lossy().into_owned(),
        clock: Arc::new(SystemClock),
        ids: Arc::new(DefaultIdGenerator::new()),
    });
    // Let the repository validate the id and choose its canonical destination.
    let destination = repo
        .create_typed(&JsonlSessionCreateOptions {
            id: Some(metadata.id.clone()),
            parent_session_id: metadata.parent_session_id.clone(),
            cwd: cwd.to_string(),
            metadata: metadata.metadata.clone(),
        })
        .await
        .map_err(|e| e.to_string())?;
    let path = destination.jsonl_metadata().path.clone();
    drop(destination);
    if let Err(error) = claim_session_lock(&path) {
        // Another process owns this destination; leave its file intact.
        return Err(error);
    }
    let destination_progress = format!("{path}.progress.jsonl");
    let publish = (|| -> std::io::Result<()> {
        if has_progress {
            std::fs::rename(&staged_progress, &destination_progress)?;
        }
        std::fs::rename(&staged, &path)
    })();
    if let Err(error) = publish {
        let _ = std::fs::remove_file(&path);
        let _ = std::fs::remove_file(&destination_progress);
        return Err(format!("publish import: {error}"));
    }
    metadata.path = path;
    let storage = repo
        .open_by_jsonl_metadata(&metadata)
        .await
        .map_err(|e| e.to_string())?;
    Ok(Session::new(Arc::new(storage), None))
}

/// Session logs this process has claimed, so re-claiming is a no-op.
static SESSION_LOCK_PATHS: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
/// The held lock handles. Kept alive for the process lifetime — dropping the
/// `File` would release the OS lock, so the guard is parked here.
static SESSION_LOCKS: OnceLock<Mutex<Vec<std::fs::File>>> = OnceLock::new();

/// Claim the exclusive lock for a session log.
///
/// Two rpi processes appending to one JSONL diverge their `sequence` counters
/// (duplicate / gapped seqs) and corrupt it. The lock is an OS advisory lock on
/// `<path>.lock`: the second process gets a clear "already open" error instead
/// of corrupting the log, and the OS releases it automatically when the holder
/// exits (no stale-lock cleanup). Re-claiming a path this process already holds
/// is a no-op, so the TUI's session hot-switch can re-open freely.
fn claim_session_lock(session_path: &str) -> Result<(), String> {
    claim_session_lock_inner(session_path, true)
}
fn claim_session_lock_inner(session_path: &str, allow_reentry: bool) -> Result<(), String> {
    let canonical =
        std::fs::canonicalize(session_path).unwrap_or_else(|_| PathBuf::from(session_path));
    let lock_path = PathBuf::from(format!("{}.lock", canonical.display()));
    let mut paths = SESSION_LOCK_PATHS
        .get_or_init(|| Mutex::new(HashSet::new()))
        .lock()
        .unwrap();
    if paths.contains(&lock_path) {
        return if allow_reentry {
            Ok(())
        } else {
            Err("transition session already has a writer in this process".into())
        };
    }
    let file = std::fs::OpenOptions::new()
        .create(true)
        .write(true)
        .open(&lock_path)
        .map_err(|e| format!("open session lock {}: {e}", lock_path.display()))?;
    if fs2::FileExt::try_lock_exclusive(&file).is_err() {
        return Err(format!(
            "session {session_path} is already open in another rpi process"
        ));
    }
    SESSION_LOCKS
        .get_or_init(|| Mutex::new(Vec::new()))
        .lock()
        .unwrap()
        .push(file);
    paths.insert(lock_path);
    Ok(())
}

/// A restored session together with the working directory it was created in.
///
/// The recorded cwd is data for the host's notices, never a silent adoption:
/// the launch path adopts the directory in [`app`](crate::app) before any
/// resource loads (tools, extensions, system prompt, and trust all key off it),
/// and the `/session` hot-switch does not move the running tools at all. Either
/// way the user hears about the directory instead of the process quietly
/// working somewhere unexpected.
pub(crate) struct RestoredSession {
    pub session: Session,
    /// The directory recorded in the session header, when it still exists.
    /// `None` when the record is empty or the path is gone (or is a file).
    pub recorded_cwd: Option<PathBuf>,
}

/// Wrap an opened [`JsonlSessionStorage`] in the `Session` facade (shared by
/// startup restore + TUI hot-switch).
///
/// `sessions_root` is passed in rather than re-derived from `cwd`: after a
/// restore adopts the session's directory, `default_session_dir(cwd)` would
/// name the ADOPTED project's tree, which is not where this session was found.
async fn open_session(
    meta: &rpi_harness::session::jsonl::JsonlSessionMetadata,
    cwd: &str,
    sessions_root: &Path,
) -> Result<RestoredSession, BuildError> {
    use rpi_harness::session::jsonl::{JsonlSessionRepo, JsonlSessionRepoOptions};
    use rpi_harness::session::types::SessionStorage;
    use rpi_tools::FileSystem;

    let dir = sessions_root;
    let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
    let fs: Arc<dyn FileSystem> = env.clone();
    let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
        fs: fs.clone(),
        sessions_root: dir.to_string_lossy().into_owned(),
        clock: Arc::new(SystemClock),
        ids: Arc::new(DefaultIdGenerator::new()),
    });
    claim_session_lock(&meta.path).map_err(BuildError::SessionOpen)?;
    let storage = repo
        .open_by_jsonl_metadata(meta)
        .await
        .map_err(|e| BuildError::SessionOpen(format!("open {}: {e}", meta.path)))?;
    // Native saved sessions predating progressFormat remain read-only to plain
    // storage loaders. Only this locked, not-yet-shared restore opts into writes.
    let storage = storage
        .enable_progress_under_exclusive_lock()
        .await
        .map_err(|e| BuildError::SessionOpen(format!("upgrade {}: {e}", meta.path)))?;
    let storage_arc: Arc<dyn SessionStorage> = Arc::new(storage);
    Ok(RestoredSession {
        session: Session::new(storage_arc, None),
        recorded_cwd: crate::session_cwd::resolve_session_cwd(Path::new(&meta.cwd)),
    })
}

/// A fresh ephemeral in-memory session (no persistence). Used for `--no-session`
/// and for `/new` inside such a run — an ephemeral launch keeps swapping in
/// in-memory sessions, never a JSONL file.
pub(crate) fn ephemeral_session() -> Session {
    let storage = Arc::new(InMemorySessionStorage::new(
        SessionMetadata {
            id: "ephemeral".into(),
            created_at: 0,
            parent_session_id: None,
        },
        Arc::new(SystemClock),
        Arc::new(DefaultIdGenerator::new()),
    ));
    Session::new(storage, None).with_volatile_progress()
}

/// Create a fresh JSONL session file under `dir` and wrap it in a `Session`.
///
/// Uses the `JsonlSessionRepo` over an `OsExecutionEnv`-backed `FileSystem`
/// rooted at the cwd, so paths resolve consistently with the tools. Mirrors the
/// TS `SessionManager.create` flow (header write + `JsonlSessionStorage` open).
/// Create a fresh JSONL session file under `dir` and wrap it in a `Session`.
///
/// Uses the `JsonlSessionRepo` over an `OsExecutionEnv`-backed `FileSystem`
/// rooted at the cwd, so paths resolve consistently with the tools. Mirrors the
/// TS `SessionManager.create` flow (header write + `JsonlSessionStorage` open).
pub(crate) async fn create_jsonl_session(dir: &Path, cwd: &str) -> Result<Session, String> {
    create_jsonl_session_with_id(dir, cwd, None).await
}

/// `create_jsonl_session` with an explicit id (the `--session-id` fixed-id
/// contract: the file is named with the given id so later `--session-id`
/// launches restore the same session).
pub(crate) async fn create_jsonl_session_with_id(
    dir: &Path,
    cwd: &str,
    id: Option<String>,
) -> Result<Session, String> {
    use rpi_harness::session::jsonl::{
        JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
    };
    use rpi_tools::FileSystem;

    // A dedicated OS env for session-file I/O, rooted at the cwd so the repo's
    // relative-path resolution matches the tool env.
    let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(cwd)));
    let fs: Arc<dyn FileSystem> = env.clone();

    let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
        fs: fs.clone(),
        sessions_root: dir.to_string_lossy().into_owned(),
        clock: Arc::new(SystemClock),
        ids: Arc::new(DefaultIdGenerator::new()),
    });

    let opts = JsonlSessionCreateOptions {
        id, // fresh uuidv7 when None (--session-id passes the fixed id)
        parent_session_id: None,
        cwd: cwd.to_string(),
        metadata: None,
    };
    let storage = repo
        .create_typed(&opts)
        .await
        .map_err(|e| format!("create session: {e}"))?;
    claim_session_lock(&storage.jsonl_metadata().path)
        .map_err(|e| format!("create session: {e}"))?;
    // `JsonlSessionStorage` implements `SessionStorage`; wrap in the facade.
    let storage_arc: Arc<dyn rpi_harness::session::types::SessionStorage> = Arc::new(storage);
    Ok(Session::new(storage_arc, None))
}

/// Read an `--append-system-prompt` target: if it's a readable file path, return
/// its contents; otherwise return `None` and let the caller use the literal.
fn read_append_target(target: &str, sources: &mut Vec<String>) -> Option<String> {
    let path = Path::new(target);
    if path.is_file() {
        read_prompt_source(path, sources)
    } else {
        None
    }
}

/// Record provenance at the successful read, rather than rediscovering files
/// during rendering (which could report a shadowed or unreadable file).
fn read_prompt_source(path: &Path, sources: &mut Vec<String>) -> Option<String> {
    let text = std::fs::read_to_string(path).ok()?;
    let absolute = if path.is_absolute() {
        path.to_path_buf()
    } else {
        std::env::current_dir()
            .map(|cwd| cwd.join(path))
            .unwrap_or_else(|_| path.to_path_buf())
    };
    let source = absolute.to_string_lossy().into_owned();
    if !sources.contains(&source) {
        sources.push(source);
    }
    Some(text)
}

#[cfg(test)]
mod goal_outbox_smoke_tests;
#[cfg(all(test, feature = "experimental-native-store"))]
mod native_roundtrip_tests;
#[cfg(all(test, feature = "experimental-native-store"))]
mod native_store_tests;
#[cfg(test)]
mod runtime_session_tests;

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn settlement_bridge_follows_reload_and_releases_its_session_with_the_bus() {
        use rpi_harness::events::{HarnessEvent, HarnessEventBus, RunEndEvent, RunEndOutcome};
        use std::sync::atomic::{AtomicUsize, Ordering};
        static OLD: AtomicUsize = AtomicUsize::new(0);
        static NEW: AtomicUsize = AtomicUsize::new(0);
        extern "C" fn old(_: rpi_plugin_sdk::StablePluginEvent, _: *mut std::ffi::c_void) -> i32 {
            OLD.fetch_add(1, Ordering::SeqCst);
            0
        }
        extern "C" fn new(_: rpi_plugin_sdk::StablePluginEvent, _: *mut std::ffi::c_void) -> i32 {
            NEW.fetch_add(1, Ordering::SeqCst);
            0
        }
        let make = |handler: rpi_plugin_sdk::EventHandlerFn| {
            let mut registry = rpi_extensions::ExtensionRegistry::new();
            registry.register_event_handler(
                "settlement".into(),
                rpi_plugin_sdk::EventTag::AgentSettled,
                handler,
                std::ptr::null_mut(),
            );
            ExtensionSession::from_parts(
                Arc::new(registry.snapshot()),
                rpi_extensions::PluginKeepalive::empty(),
                vec![],
                None,
            )
        };
        let sessions = Arc::new(Mutex::new(make(old)));
        let weak = Arc::downgrade(&sessions);
        let bus = HarnessEventBus::new();
        install_extension_settlement_bridge(&bus, sessions.clone());
        let ended = |outcome| {
            HarnessEvent::RunEnd(RunEndEvent {
                lane: "main".into(),
                run_id: "run".into(),
                outcome,
                leaf_id: "leaf".into(),
            })
        };
        bus.emit(&ended(RunEndOutcome::Completed));
        assert_eq!(OLD.load(Ordering::SeqCst), 1);
        *sessions.lock().unwrap() = make(new);
        for outcome in [RunEndOutcome::Aborted, RunEndOutcome::Failed] {
            bus.emit(&ended(outcome));
        }
        assert_eq!(OLD.load(Ordering::SeqCst), 1);
        assert_eq!(NEW.load(Ordering::SeqCst), 2);
        drop(sessions);
        assert!(weak.upgrade().is_some());
        drop(bus);
        assert!(
            weak.upgrade().is_none(),
            "settlement observer must not create an ownership cycle"
        );
    }

    #[tokio::test]
    async fn unauthenticated_packages_are_not_reintroduced_as_legacy_adapters() {
        extern "C" fn request(
            _: rpi_plugin_sdk::StbStringRef,
            _: *mut rpi_plugin_sdk::StbString,
            _: *mut std::ffi::c_void,
        ) -> i32 {
            1
        }
        let mut registry = rpi_extensions::ExtensionRegistry::new();
        for id in ["package", "legacy"] {
            registry.register_provider(rpi_extensions::RegisteredProvider {
                provider_id: id.into(),
                base_url: "http://localhost".into(),
                api_style: "openai-completions".into(),
                request_fn: request,
                stream_fn: None,
                plugin_free_string: rpi_plugin_sdk::callbacks::free_string,
                user_data: std::ptr::null_mut(),
            });
        }
        assert!(registry.apply_declaration_json(&serde_json::json!({"providers":[{
            "id":"package", "requiresAuth":true,
            "models":[rpi_ai::Model::new("model", "Model", rpi_ai::Api::OpenaiCompletions, "package", "http://localhost")]
        }]}).to_string()));
        let session = ExtensionSession::from_parts(
            Arc::new(registry.snapshot()),
            rpi_extensions::PluginKeepalive::empty(),
            vec![],
            None,
        );
        let providers = build_models_with_extensions_for_reload(
            &[],
            &session,
            tokio::runtime::Handle::current(),
        );
        assert_eq!(
            providers
                .iter()
                .map(|provider| provider.id())
                .collect::<Vec<_>>(),
            ["legacy"]
        );
    }

    // A catalog refresh must hand the lane the NEW entry, not the old object:
    // a successful reload that leaves the lane on stale metadata (endpoint,
    // context window) is silently wrong -- requests keep using it.
    #[test]
    fn catalog_refresh_keeps_model_identity_but_takes_new_metadata() {
        let stale = rpi_ai::Model {
            base_url: "http://localhost/old".into(),
            context_window: 8192,
            ..rpi_ai::Model::new(
                "model",
                "Model",
                rpi_ai::Api::OpenaiCompletions,
                "probe",
                "http://localhost",
            )
        };
        let catalog = vec![rpi_ai::Model {
            base_url: "http://localhost/new".into(),
            context_window: 32768,
            ..stale.clone()
        }];
        let current = Some(stale);

        let lane_model = current
            .as_ref()
            .and_then(|model| {
                catalog
                    .iter()
                    .find(|m| m.id == model.id && m.provider == model.provider)
            })
            .cloned()
            .unwrap_or_else(|| catalog[0].clone());

        assert_eq!(lane_model.base_url, "http://localhost/new");
        assert_eq!(lane_model.context_window, 32768);
    }

    use crate::args::Args;

    #[test]
    fn loaded_context_paths_track_precedence_and_reload() {
        const CHILD: &str = "RPI_CONTEXT_PATHS_REGRESSION_CHILD";
        if std::env::var_os(CHILD).is_none() {
            let dir = tempfile::tempdir().unwrap();
            let output = std::process::Command::new(std::env::current_exe().unwrap())
                .args([
                    "--exact",
                    "session::tests::loaded_context_paths_track_precedence_and_reload",
                    "--nocapture",
                ])
                .env(CHILD, "1")
                .env(crate::config::CONFIG_DIR_ENV, dir.path())
                .output()
                .unwrap();
            assert!(
                output.status.success(),
                "{}\n{}",
                String::from_utf8_lossy(&output.stdout),
                String::from_utf8_lossy(&output.stderr)
            );
            return;
        }
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(async {
                let agent = crate::config::agent_dir().unwrap();
                let cwd = agent.join("project");
                let project_config = cwd.join(".rpi");
                std::fs::create_dir_all(&project_config).unwrap();
                for (path, content) in [
                    (agent.join("SYSTEM.md"), "GLOBAL_BASE"),
                    (agent.join("APPEND_SYSTEM.md"), "GLOBAL_APPEND"),
                    (agent.join("AGENTS.md"), "GLOBAL_INSTRUCTIONS"),
                    (project_config.join("SYSTEM.md"), "PROJECT_BASE"),
                    (project_config.join("APPEND_SYSTEM.md"), "PROJECT_APPEND"),
                    (cwd.join("AGENTS.md"), "PROJECT_INSTRUCTIONS"),
                    (cwd.join("CLAUDE.md"), "SHADOWED_CLAUDE"),
                ] {
                    std::fs::write(path, content).unwrap();
                }
                let resolved = crate::provider::resolve(
                    Some("anthropic"),
                    Some(crate::provider::DEFAULT_MODEL_ID),
                    None,
                    Some("test-key"),
                    None,
                )
                .unwrap();
                let args = Args {
                    no_session: true,
                    no_extensions: true,
                    no_skills: true,
                    no_prompt_templates: true,
                    trust_override: Some(true),
                    ..Default::default()
                };
                let (harness, _events, mut context) =
                    build(&resolved, &args, &cwd, true).await.unwrap();
                let resources = harness.get_resources().await.unwrap();
                let path = |p: PathBuf| p.to_string_lossy().into_owned();
                assert_eq!(
                    &resources.context_files[..2],
                    &[
                        path(project_config.join("SYSTEM.md")),
                        path(project_config.join("APPEND_SYSTEM.md")),
                    ]
                );
                assert!(resources
                    .context_files
                    .contains(&path(agent.join("AGENTS.md"))));
                assert!(resources
                    .context_files
                    .contains(&path(cwd.join("AGENTS.md"))));
                assert!(!resources
                    .context_files
                    .contains(&path(cwd.join("CLAUDE.md"))));
                assert!(!resources
                    .context_files
                    .contains(&path(agent.join("SYSTEM.md"))));
                let prompt = harness.get_system_prompt().await.unwrap().unwrap();
                assert!(prompt.contains("PROJECT_BASE") && prompt.contains("PROJECT_APPEND"));
                assert!(!prompt.contains("GLOBAL_BASE") && !prompt.contains("SHADOWED_CLAUDE"));

                // Removing the selected project files makes reload fall back to
                // globals; removed instruction paths must disappear as well.
                std::fs::remove_file(project_config.join("SYSTEM.md")).unwrap();
                std::fs::remove_file(project_config.join("APPEND_SYSTEM.md")).unwrap();
                std::fs::remove_file(cwd.join("AGENTS.md")).unwrap();
                assert!(
                    !reload_extension_resources(&harness, &context)
                        .await
                        .had_errors
                );
                let resources = harness.get_resources().await.unwrap();
                assert_eq!(
                    &resources.context_files[..2],
                    &[
                        path(agent.join("SYSTEM.md")),
                        path(agent.join("APPEND_SYSTEM.md")),
                    ]
                );
                assert!(resources
                    .context_files
                    .contains(&path(cwd.join("CLAUDE.md"))));
                assert!(!resources
                    .context_files
                    .contains(&path(project_config.join("SYSTEM.md"))));

                // Explicit append files are listed; literal text and overridden
                // discovered base/append files are not invented as file sources.
                let explicit = agent.join("custom-append.md");
                std::fs::write(&explicit, "EXPLICIT_APPEND").unwrap();
                context.args.system_prompt = Some("EXPLICIT_BASE".into());
                context.args.append_system_prompt =
                    vec![path(explicit.clone()), "literal append".into()];
                context.args.no_context_files = true;
                assert!(
                    !reload_extension_resources(&harness, &context)
                        .await
                        .had_errors
                );
                assert_eq!(
                    harness.get_resources().await.unwrap().context_files,
                    vec![path(explicit)]
                );

                // The reload must honor the startup opt-out from project context.
                context.args.system_prompt = None;
                context.args.append_system_prompt.clear();
                context.args.no_context_files = false;
                context.args.trust_override = Some(false);
                assert!(
                    !reload_extension_resources(&harness, &context)
                        .await
                        .had_errors
                );
                assert!(harness
                    .get_resources()
                    .await
                    .unwrap()
                    .context_files
                    .is_empty());
                let prompt = harness.get_system_prompt().await.unwrap().unwrap();
                assert!(
                    !prompt.contains("SHADOWED_CLAUDE") && !prompt.contains("GLOBAL_INSTRUCTIONS")
                );
            });
    }

    #[test]
    fn prompt_source_read_failures_and_literals_do_not_create_file_provenance() {
        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("SYSTEM.md");
        let mut sources = Vec::new();
        assert!(read_prompt_source(&file, &mut sources).is_none());
        std::fs::write(&file, [0xff, 0xfe]).unwrap();
        assert!(read_prompt_source(&file, &mut sources).is_none());
        assert!(read_append_target("literal instructions", &mut sources).is_none());
        assert!(sources.is_empty());
        std::fs::write(&file, "loaded instructions").unwrap();
        assert_eq!(
            read_prompt_source(&file, &mut sources).as_deref(),
            Some("loaded instructions")
        );
        assert!(read_prompt_source(&file, &mut sources).is_some());
        assert_eq!(sources, vec![file.to_string_lossy().into_owned()]);
    }

    extern "C" fn deny_test_tool_call(
        _event: rpi_plugin_sdk::StablePluginEvent,
        _data: *mut std::ffi::c_void,
    ) -> i32 {
        rpi_plugin_sdk::EVENT_HANDLER_ABORT
    }

    async fn invoke_test_tool_hook(
        hook: &rpi_agent::BeforeToolCall,
    ) -> Option<rpi_agent::BeforeToolCallResult> {
        let message = rpi_ai::providers::faux::faux_assistant_message(
            "approval",
            rpi_ai::types::StopReason::ToolUse,
        );
        let rpi_ai::types::Content::ToolCall(call) =
            rpi_ai::types::Content::tool_call("call", "read", serde_json::json!({"path":"x"}))
        else {
            unreachable!()
        };
        hook(
            rpi_agent::BeforeToolCallContext {
                assistant_message: &message,
                tool_call: &call,
                args: &call.arguments,
                context: &rpi_agent::AgentContext::default(),
            },
            tokio_util::sync::CancellationToken::new(),
        )
        .await
    }

    #[tokio::test]
    async fn tool_call_hook_uses_reloaded_session_and_newly_discovered_gate() {
        let sessions = Arc::new(Mutex::new(ExtensionSession::none()));
        let hook =
            extension_tool_call_hook(sessions.clone(), rpi_extensions::UiDialogMailbox::new());
        assert!(invoke_test_tool_hook(&hook).await.is_none());
        for name in ["first-gate", "reloaded-gate"] {
            let mut registry = rpi_extensions::ExtensionRegistry::new();
            registry.register_event_handler(
                name.into(),
                rpi_plugin_sdk::EventTag::ToolCall,
                deny_test_tool_call,
                std::ptr::null_mut(),
            );
            let fresh = ExtensionSession::from_parts(
                Arc::new(registry.snapshot()),
                rpi_extensions::PluginKeepalive::empty(),
                Vec::new(),
                None,
            );
            {
                let mut current = sessions.lock().unwrap();
                if let Some(snapshot) = current.snapshot() {
                    snapshot.active_flag().store(false, Ordering::SeqCst);
                }
                *current = fresh;
            }
            let result = invoke_test_tool_hook(&hook).await.unwrap();
            assert!(result.block);
            assert!(result.reason.unwrap().contains(name));
        }
        *sessions.lock().unwrap() = ExtensionSession::none();
        assert!(invoke_test_tool_hook(&hook).await.is_none());
    }

    /// A unique scratch file for one summary test.
    fn scratch(name: &str) -> PathBuf {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "rpi-summary-{}-{nanos}-{name}.jsonl",
            std::process::id()
        ));
        path
    }

    #[test]
    fn summary_marks_a_header_only_session_empty() {
        let path = scratch("empty");
        std::fs::write(
            &path,
            "{\"kind\":\"header\",\"version\":4,\"id\":\"abc\",\"createdAt\":1,\"cwd\":\"x\"}\n",
        )
        .unwrap();
        let summary = summarize_session_file(&path);
        assert!(summary.empty, "a header-only file has nothing to resume");
        assert!(summary.preview.is_none());
        assert_eq!(summary.bytes, std::fs::metadata(&path).unwrap().len());
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn summary_reads_the_first_user_prompt_and_name() {
        let path = scratch("content");
        let body = concat!(
            "{\"kind\":\"header\",\"version\":4,\"id\":\"abc\",\"createdAt\":1,\"cwd\":\"x\"}\n",
            "{\"kind\":\"fact\",\"seq\":1,\"fact\":\"name\",\"name\":\"fix resume picker\"}\n",
            "{\"kind\":\"record\",\"type\":\"operation_started\",\"id\":\"r1\"}\n",
            "{\"kind\":\"entry\",\"lane\":\"main\",\"type\":\"message\",\"id\":\"e1\",\"message\":{\"kind\":\"user\",\"role\":\"user\",\"content\":\"read the rpi -r picker\"}}\n",
            "{\"kind\":\"entry\",\"lane\":\"main\",\"type\":\"message\",\"id\":\"e2\",\"message\":{\"kind\":\"assistant\",\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"second\"}]}}\n",
        );
        std::fs::write(&path, body).unwrap();
        let summary = summarize_session_file(&path);
        assert!(!summary.empty);
        assert_eq!(summary.name.as_deref(), Some("fix resume picker"));
        assert_eq!(summary.preview.as_deref(), Some("read the rpi -r picker"));
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn summary_extracts_a_block_style_user_prompt() {
        let path = scratch("blocks");
        let body = concat!(
            "{\"kind\":\"header\",\"version\":4,\"id\":\"abc\",\"createdAt\":1,\"cwd\":\"x\"}\n",
            "{\"kind\":\"entry\",\"type\":\"message\",\"id\":\"e1\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"text\",\"text\":\"line one\\nline two\"}]}}\n",
        );
        std::fs::write(&path, body).unwrap();
        let summary = summarize_session_file(&path);
        assert_eq!(summary.preview.as_deref(), Some("line one line two"));
        assert_eq!(session_display_label(&summary), "line one line two");
        std::fs::remove_file(&path).ok();
    }

    fn meta_for(path: &Path, id: &str) -> rpi_harness::session::jsonl::JsonlSessionMetadata {
        rpi_harness::session::jsonl::JsonlSessionMetadata {
            id: id.to_string(),
            created_at: 0,
            cwd: ".".to_string(),
            path: path.to_string_lossy().into_owned(),
            modified_at: 0,
            source_format: rpi_harness::session::jsonl::JsonlSourceFormat::V4,
            parent_session_id: None,
            legacy_parent_session_path: None,
            metadata: None,
        }
    }

    #[test]
    fn continue_skips_newer_empty_sessions() {
        let empty = scratch("newest-empty");
        std::fs::write(
            &empty,
            "{\"kind\":\"header\",\"version\":4,\"id\":\"empty\",\"createdAt\":1,\"cwd\":\"x\"}\n",
        )
        .unwrap();
        let content = scratch("older-content");
        std::fs::write(
            &content,
            concat!(
                "{\"kind\":\"header\",\"version\":4,\"id\":\"old\",\"createdAt\":1,\"cwd\":\"x\"}\n",
                "{\"kind\":\"entry\",\"type\":\"message\",\"id\":\"e1\",\"message\":{\"role\":\"user\",\"content\":\"hello\"}}\n",
            ),
        )
        .unwrap();

        let metas = vec![meta_for(&empty, "empty"), meta_for(&content, "old")];
        let picked = first_resumable_session(&metas).expect("a resumable session");
        assert_eq!(picked.id, "old");

        // When every session is empty, fall back to the newest rather than
        // erroring.
        let all_empty = vec![meta_for(&empty, "empty")];
        assert_eq!(first_resumable_session(&all_empty).unwrap().id, "empty");

        std::fs::remove_file(&empty).ok();
        std::fs::remove_file(&content).ok();
    }

    #[tokio::test]
    async fn import_preserves_validates_and_locks_the_recovery_pair() {
        use rpi_harness::session::jsonl::{JsonlSessionStorage, JsonlV4Header};
        use rpi_harness::session::types::{
            AssistantFrameOp, AssistantFrameRecord, RecordBase, SessionStorage,
        };
        let tmp = tempfile::tempdir().unwrap();
        let source = tmp.path().join("source.jsonl");
        let progress = tmp.path().join("source.jsonl.progress.jsonl");
        let fs: Arc<dyn rpi_tools::FileSystem> =
            Arc::new(OsExecutionEnv::with_cwd(tmp.path().into()));
        let storage = JsonlSessionStorage::create(
            fs,
            &source.to_string_lossy(),
            JsonlV4Header::new("import-pair".into(), 1, "/".into(), None, None, None),
            Arc::new(SystemClock),
            Arc::new(DefaultIdGenerator::new()),
        )
        .await
        .unwrap();
        storage
            .append_progress(vec![AssistantFrameRecord {
                base: RecordBase {
                    id: "accepted-prefix".into(),
                    seq: 0,
                    lane: "main".into(),
                    timestamp: 0,
                },
                run_id: "run".into(),
                stream_index: 0,
                op: AssistantFrameOp::Append,
                frame: Some(serde_json::json!({"payload":"keep"})),
            }])
            .await
            .unwrap();
        drop(storage);
        let original = std::fs::read(&progress).unwrap();
        let destination = tmp.path().join("imported");
        let imported = import_session_file(&source, &tmp.path().to_string_lossy(), &destination)
            .await
            .unwrap();
        assert_eq!(
            imported.read_progress("run").await.unwrap()[0].base.id,
            "accepted-prefix"
        );
        assert_eq!(
            std::fs::read(&progress).unwrap(),
            original,
            "import must not repair or mutate its source"
        );
        let lock = std::fs::OpenOptions::new()
            .create(true)
            .write(true)
            .open(tmp.path().join("source.jsonl.lock"))
            .unwrap();
        fs2::FileExt::try_lock_exclusive(&lock).unwrap();
        assert!(import_session_file(
            &source,
            &tmp.path().to_string_lossy(),
            &tmp.path().join("busy")
        )
        .await
        .is_err());
        drop(lock);
        let mut corrupt = String::from_utf8(original.clone()).unwrap();
        corrupt = corrupt.replace(
            "\"session_id\":\"import-pair\"",
            "\"session_id\":\"foreign\"",
        );
        assert_ne!(corrupt.as_bytes(), original.as_slice());
        std::fs::write(&progress, corrupt).unwrap();
        assert!(import_session_file(
            &source,
            &tmp.path().to_string_lossy(),
            &tmp.path().join("foreign")
        )
        .await
        .is_err());
        std::fs::remove_file(&progress).unwrap();
        assert!(import_session_file(
            &source,
            &tmp.path().to_string_lossy(),
            &tmp.path().join("missing")
        )
        .await
        .is_err());
    }

    #[test]
    fn list_session_files_sync_walks_the_cwd_subdirectory() {
        let root = std::env::temp_dir().join(format!("rpi-summary-{}-dir", std::process::id()));
        let nested = root.join("--proj--");
        std::fs::create_dir_all(&nested).unwrap();
        let file = nested.join("2026-01-01_a.jsonl");
        std::fs::write(&file, "{}\n").unwrap();
        let sidecar = nested.join("2026-01-01_a.jsonl.progress.jsonl");
        std::fs::write(&sidecar, "{}\n").unwrap();
        let found = list_session_files_sync(&root);
        assert!(found.iter().any(|p| p == &file));
        assert!(!found.contains(&sidecar));
        std::fs::remove_dir_all(&root).ok();
    }

    /// The default CLI tool set plus the `todo` tool its bundled plugins add.
    fn default_prompt_tools() -> Vec<String> {
        BUILTIN_TOOL_NAMES
            .iter()
            .copied()
            .chain(["todo"])
            .map(str::to_owned)
            .collect()
    }

    #[test]
    fn default_prompt_lists_only_the_active_tools() {
        let tools = ["read", "grep", "find", "ls"].map(str::to_owned);
        let p = default_system_prompt_with_tools("/tmp/proj", true, &tools);
        assert!(p.contains("- read  — Read file contents"));
        assert!(p.contains("- grep  — Search file contents"));
        for absent in ["- edit", "- write", "- bash", "- docs", "todo", "Windows"] {
            assert!(
                !p.contains(absent),
                "{absent} must not appear for a read-only run: {p}"
            );
        }
        assert!(p.contains("Re-stating the same plan without acting on it is a bug"));
        let none = default_system_prompt_with_tools("/tmp/proj", false, &[]);
        assert!(none.contains("Available tools:\n(none)"));
    }

    #[test]
    fn default_prompt_mentions_cwd_and_tools() {
        let p = default_system_prompt_with_tools("/tmp/proj", false, &default_prompt_tools());
        assert!(p.contains("/tmp/proj"));
        assert!(p.contains("read"));
        assert!(p.contains("bash"));
        assert!(p.contains("edit"));
        assert!(p.contains("write"));
        assert!(p.contains("docs"));
        assert!(!p.contains("- grep"));
        assert!(!p.contains("- find"));
        assert!(!p.contains("- ls"));
        assert!(!p.contains("- powershell"));
        assert!(p.contains("Bash/POSIX syntax, also on Windows"));
    }

    #[test]
    fn default_prompt_instructs_language_matching() {
        // A Chinese (or any non-English) user gets a same-language reply
        // without having to ask every turn.
        let p = default_system_prompt_with_tools("/tmp/proj", false, &default_prompt_tools());
        assert!(
            p.contains("same language as the user's latest message"),
            "the prompt must tell the model to mirror the user's language: {p}"
        );
    }

    #[test]
    fn default_prompt_keeps_working_state_in_the_visible_channel() {
        // Regression guard for the re-plan loop: on openai-compatible providers
        // the assistant's `thinking` blocks are dropped when the next request is
        // built (`providers/openai_completions.rs`: `Content::text_only`), so a
        // plan that lives only in reasoning is re-derived every turn. One
        // observed session paid for that 112 times in a single run, and twice
        // hallucinated the user's input ("The user is greeting me with hello",
        // with no such message). See `docs/llm-repetition-forensics.md` §八.
        let p = default_system_prompt_with_tools("/tmp/proj", false, &default_prompt_tools());
        let lower = p.to_lowercase();
        // The rule must state the mechanism, not just "be clear".
        assert!(
            lower.contains("reasoning is not") && lower.contains("carried into your next turn"),
            "the prompt must explain that reasoning does not survive the turn: {p}"
        );
        // Working state stays visible; extension-specific rules require opt-in.
        assert!(
            lower.contains("visible") && !lower.contains("todo tool"),
            "the prompt must keep state visible without inventing extension rules: {p}"
        );
        // …plus the anti-repetition rule itself.
        assert!(
            lower.contains("re-stating the same plan"),
            "the prompt must call out plan re-statement as a bug: {p}"
        );
        // And the guidance must not instruct re-reading what is already shown.
        assert!(
            lower.contains("already shown earlier"),
            "the prompt must discourage re-fetching known content: {p}"
        );
    }

    #[test]
    fn default_prompt_does_not_claim_reasoning_is_lost_when_it_is_replayed() {
        // `compat.requiresThinkingAsText` (and the native thinking channels of
        // anthropic-messages / openai-responses) DO carry the model's reasoning
        // back on the next request. Telling such a model "reasoning is not
        // carried into your next turn" is false, and the instruction to restate
        // the plan every turn is exactly the busy-work the re-plan analysis
        // measured. The wording must flip with the mechanism.
        let p = default_system_prompt_with_tools("/tmp/proj", true, &default_prompt_tools());
        let lower = p.to_lowercase();
        assert!(
            lower.contains("carried back"),
            "a replaying model must be told its reasoning survives: {p}"
        );
        assert!(
            !lower.contains("reasoning is not"),
            "the replayed variant must not claim reasoning is dropped: {p}"
        );
        assert!(
            lower.contains("do not re-derive"),
            "the replayed variant must forbid re-deriving the plan: {p}"
        );
        assert!(
            lower.contains("do not re-derive") && lower.contains("re-stating the same plan"),
            "the anti-repetition rule must survive the flip: {p}"
        );
        assert!(
            lower.contains("already shown earlier"),
            "the re-read rule must survive the flip: {p}"
        );
    }

    #[test]
    fn default_prompt_states_each_rule_once() {
        // The plan rule, the todo tool and the anti-repetition line used to be
        // spread over two guidelines that repeated each other.
        for replayed in [false, true] {
            let p =
                default_system_prompt_with_tools("/tmp/proj", replayed, &default_prompt_tools());
            for phrase in ["the same plan", "Windows", "guessing"] {
                assert_eq!(p.matches(phrase).count(), 1, "{phrase:?} repeated: {p}");
            }
            assert!(
                !p.contains("todo tool"),
                "extension rules require explicit metadata"
            );
        }
    }

    /// Only side-effect-free tools may be replayed after a crash.
    ///
    /// Regression guard for an inverted default: `ToolReplay` used to default to
    /// `Safe` and the CLI then marked *every* tool `Safe`, so the first time
    /// recovery acted on the flag it would have re-run `bash`/`write`/`edit` and
    /// duplicated their side effects. upstream defaults to `"never"`
    /// (`drive/tools.ts: tool.replay ?? "never"`).
    #[test]
    fn only_read_only_tools_are_replayable() {
        let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(".")));
        let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();
        let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
        let context = ExecutionToolContext::new(env_dyn, Some(mut_env));
        let tools = build_tools(&context, &Args::default());

        let replay_of = |name: &str| {
            tools
                .iter()
                .find(|tool| tool.tool.schema().name == name)
                .map(|tool| tool.replay)
        };

        for read_only in ["read", "docs"] {
            assert_eq!(
                replay_of(read_only),
                Some(ToolReplay::Safe),
                "{read_only} is side-effect free, so recovery may re-run it"
            );
        }
        for mutating in ["bash", "edit", "write", "powershell"] {
            if let Some(replay) = replay_of(mutating) {
                assert_eq!(
                    replay,
                    ToolReplay::Never,
                    "{mutating} must never be replayed: re-running it would duplicate its effect"
                );
            }
        }
    }

    #[test]
    fn default_tools_include_docs_lookup() {
        let env = Arc::new(OsExecutionEnv::with_cwd(PathBuf::from(".")));
        let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env.clone();
        let mut_env: Arc<dyn rpi_tools::MutatingEnv> = env.clone();
        let context = ExecutionToolContext::new(env_dyn, Some(mut_env));
        let names: Vec<String> = build_tools(&context, &Args::default())
            .iter()
            .map(|tool| tool.tool.schema().name.clone())
            .collect();

        // `docs` is a default capability; the legacy rpi-only lookup tools stay
        // unregistered. Asserted by membership (not exact equality) because
        // `powershell` is Windows-only and `settings.json#defaultTools` may
        // narrow the set in a developer's environment.
        assert!(names.contains(&"docs".to_string()));
        for legacy in ["grep", "find", "ls"] {
            assert!(
                !names.contains(&legacy.to_string()),
                "legacy tool {legacy} should not be registered by default"
            );
        }
    }

    #[test]
    fn reload_settings_preflight_respects_trust() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        let cwd = tmp.path().join("project");
        std::fs::create_dir_all(&agent).unwrap();
        std::fs::create_dir_all(cwd.join(".rpi")).unwrap();
        std::fs::write(agent.join("settings.json"), "{}").unwrap();
        std::fs::write(cwd.join(".rpi/settings.json"), "{ malformed").unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
        let args = Args::default();

        let trusted = validate_settings_for_reload(&args, &cwd, true);
        let untrusted = validate_settings_for_reload(&args, &cwd, false);
        assert!(trusted
            .unwrap_err()
            .contains("could not load project settings"));
        assert!(untrusted.is_ok());

        std::fs::write(agent.join("settings.json"), "{ malformed").unwrap();
        assert!(validate_settings_for_reload(&args, &cwd, false)
            .unwrap_err()
            .contains("could not load global settings"));

        let local_only = Args {
            dev_local_only: true,
            ..Args::default()
        };
        assert!(validate_settings_for_reload(&local_only, &cwd, true).is_ok());
    }

    #[tokio::test(flavor = "current_thread")]
    async fn reload_preserves_live_resources_when_settings_break() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        let cwd = tmp.path().join("project");
        let skill_dir = tmp.path().join("configured-skills");
        std::fs::create_dir_all(&agent).unwrap();
        std::fs::create_dir_all(&cwd).unwrap();
        std::fs::create_dir_all(&skill_dir).unwrap();
        std::fs::write(
            skill_dir.join("SKILL.md"),
            "---\nname: keep-me\ndescription: Reload sentinel\n---\nKeep this skill loaded.",
        )
        .unwrap();
        std::fs::write(
            agent.join("settings.json"),
            serde_json::json!({"skillDirs": [skill_dir]}).to_string(),
        )
        .unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);

        let resolved = crate::provider::resolve(
            Some("anthropic"),
            Some(crate::provider::DEFAULT_MODEL_ID),
            None,
            Some("test-key"),
            None,
        )
        .unwrap();
        let args = Args {
            trust_override: Some(false),
            no_session: true,
            no_extensions: true,
            no_prompt_templates: true,
            no_context_files: true,
            system_prompt: Some("stable system prompt".into()),
            ..Args::default()
        };
        let (harness, _events, context) = build(&resolved, &args, &cwd, false).await.unwrap();
        let before_resources = harness.get_resources().await.unwrap();
        assert_eq!(before_resources.skills.as_ref().unwrap().len(), 1);
        assert_eq!(before_resources.skills.as_ref().unwrap()[0].name, "keep-me");
        let before_prompt = harness.get_system_prompt().await.unwrap();
        let before_tools: Vec<String> = harness
            .get_tools()
            .await
            .unwrap()
            .iter()
            .map(|tool| tool.tool.schema().name.clone())
            .collect();
        let before_bridge = context
            .action_bridge
            .lock()
            .unwrap()
            .as_ref()
            .unwrap()
            .clone();

        std::fs::write(agent.join("settings.json"), "{ malformed").unwrap();
        let outcome = reload_extension_resources(&harness, &context).await;

        assert!(outcome.had_warnings);
        assert!(outcome.summary.contains("Settings reload failed"));
        assert!(outcome.summary.contains("could not load global settings"));
        // Hard failure keeps the previous resources; the outcome must not
        // advertise a fresh trust policy either (theme discovery keeps the
        // registry it already matches).
        assert_eq!(outcome.project_trusted, None);
        assert_eq!(harness.get_resources().await.unwrap(), before_resources);
        assert_eq!(harness.get_system_prompt().await.unwrap(), before_prompt);
        let after_tools: Vec<String> = harness
            .get_tools()
            .await
            .unwrap()
            .iter()
            .map(|tool| tool.tool.schema().name.clone())
            .collect();
        assert_eq!(after_tools, before_tools);
        let after_bridge = context
            .action_bridge
            .lock()
            .unwrap()
            .as_ref()
            .unwrap()
            .clone();
        assert!(Arc::ptr_eq(&after_bridge, &before_bridge));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn reload_outcome_carries_the_current_project_trust() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        let cwd = tmp.path().join("project");
        std::fs::create_dir_all(&agent).unwrap();
        std::fs::create_dir_all(&cwd).unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);

        // Startup snapshot: trusted.
        crate::config::set_project_trust(&cwd, Some(true)).unwrap();
        let resolved = crate::provider::resolve(
            Some("anthropic"),
            Some(crate::provider::DEFAULT_MODEL_ID),
            None,
            Some("test-key"),
            None,
        )
        .unwrap();
        let args = Args {
            no_session: true,
            no_extensions: true,
            no_prompt_templates: true,
            no_context_files: true,
            system_prompt: Some("stable system prompt".into()),
            ..Args::default()
        };
        let (harness, _events, context) = build(&resolved, &args, &cwd, true).await.unwrap();
        assert!(context.project_trusted, "startup snapshot is trusted");

        // /trust no, then /reload: the outcome must report the CURRENT
        // decision, not the immutable startup snapshot.
        crate::config::set_project_trust(&cwd, Some(false)).unwrap();
        let outcome = reload_extension_resources(&harness, &context).await;
        assert!(!outcome.had_errors, "{}", outcome.summary);
        assert_eq!(outcome.project_trusted, Some(false));

        // And the opposite transition back to trusted.
        crate::config::set_project_trust(&cwd, Some(true)).unwrap();
        let outcome = reload_extension_resources(&harness, &context).await;
        assert!(!outcome.had_errors, "{}", outcome.summary);
        assert_eq!(outcome.project_trusted, Some(true));
    }

    #[test]
    fn reload_preparation_rejects_settings_changed_during_derivation() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        let cwd = tmp.path().join("project");
        let first_skill_dir = tmp.path().join("first-skills");
        let second_skill_dir = tmp.path().join("second-skills");
        std::fs::create_dir_all(&agent).unwrap();
        std::fs::create_dir_all(&cwd).unwrap();
        std::fs::write(
            agent.join("settings.json"),
            serde_json::json!({"skillDirs": [first_skill_dir]}).to_string(),
        )
        .unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);
        let args = Args {
            trust_override: Some(false),
            no_extensions: true,
            no_prompt_templates: true,
            ..Args::default()
        };

        let result = prepare_reload_inputs_inner(&args, &cwd, false, || {
            std::fs::write(
                agent.join("settings.json"),
                serde_json::json!({"skillDirs": [second_skill_dir]}).to_string(),
            )
            .unwrap();
        });

        let error = result
            .err()
            .expect("settings mutation must fail preparation");
        assert!(error.contains("settings changed while reload inputs were being prepared"));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn reload_uses_frozen_settings_inputs_after_preparation() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        let cwd = tmp.path().join("project");
        let skill_dir = tmp.path().join("configured-skills");
        std::fs::create_dir_all(&agent).unwrap();
        std::fs::create_dir_all(&cwd).unwrap();
        std::fs::create_dir_all(&skill_dir).unwrap();
        std::fs::write(
            skill_dir.join("SKILL.md"),
            "---\nname: frozen-skill\ndescription: Reload sentinel\n---\nFrozen input.",
        )
        .unwrap();
        std::fs::write(
            agent.join("settings.json"),
            serde_json::json!({"skillDirs": [skill_dir]}).to_string(),
        )
        .unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);

        let resolved = crate::provider::resolve(
            Some("anthropic"),
            Some(crate::provider::DEFAULT_MODEL_ID),
            None,
            Some("test-key"),
            None,
        )
        .unwrap();
        let args = Args {
            trust_override: Some(false),
            no_session: true,
            no_extensions: true,
            no_prompt_templates: true,
            no_context_files: true,
            system_prompt: Some("stable system prompt".into()),
            ..Args::default()
        };
        let (harness, _events, context) = build(&resolved, &args, &cwd, false).await.unwrap();

        let outcome = reload_extension_resources_inner(&harness, &context, || {
            std::fs::write(agent.join("settings.json"), "{ malformed").unwrap();
        })
        .await;

        assert!(!outcome.summary.contains("Settings reload failed"));
        assert_eq!(
            outcome.summary,
            "Reloaded 0 plugin(s), 1 skill(s), 0 prompt(s)."
        );
        let resources = harness.get_resources().await.unwrap();
        let skills = resources.skills.as_ref().unwrap();
        assert_eq!(skills.len(), 1);
        assert_eq!(skills[0].name, "frozen-skill");
    }

    #[tokio::test]
    async fn reload_resource_paths_include_explicit_skill_and_prompt_files() {
        let tmp = tempfile::tempdir().unwrap();
        let skill_path = tmp.path().join("explicit-skill.md");
        std::fs::write(
            &skill_path,
            "---\nname: explicit-skill\ndescription: Explicit skill\n---\nSkill body",
        )
        .unwrap();
        let prompt_path = tmp.path().join("explicit-prompt.md");
        std::fs::write(
            &prompt_path,
            "---\ndescription: Explicit prompt\n---\nPrompt body",
        )
        .unwrap();

        let args = Args {
            skill: vec![skill_path.clone()],
            prompt_template: vec![prompt_path.clone()],
            ..Args::default()
        };
        let skill_paths = append_reload_resource_paths(Vec::new(), &args.skill, &[], &[], &[]);
        let prompt_paths =
            append_reload_resource_paths(Vec::new(), &args.prompt_template, &[], &[], &[]);
        let env = Arc::new(OsExecutionEnv::with_cwd(tmp.path().to_path_buf()));
        let env_dyn: Arc<dyn rpi_tools::ExecutionEnv> = env;

        let skills = load_skills_with_precedence(&env_dyn, &skill_paths).await;
        assert_eq!(skills.skills.len(), 1, "{:?}", skills.diagnostics);
        assert_eq!(skills.skills[0].name, "explicit-skill");

        let prompts = load_prompt_templates_with_precedence(&env_dyn, &prompt_paths).await;
        assert_eq!(
            prompts.prompt_templates.len(),
            1,
            "{:?}",
            prompts.diagnostics
        );
        assert_eq!(prompts.prompt_templates[0].name, "explicit-prompt");
    }

    #[test]
    fn tool_policy_applies_to_rust_and_js_active_names() {
        let names = vec![
            "read".to_string(),
            "ask_user_question".to_string(),
            "write".to_string(),
        ];

        let args = Args {
            tools: Some(vec!["read".into(), "ask_user_question".into()]),
            ..Args::default()
        };
        assert_eq!(
            filter_active_tool_names(names.clone(), &args),
            vec!["read", "ask_user_question"]
        );

        let args = Args {
            exclude_tools: Some(vec!["ask_user_question".into()]),
            ..Args::default()
        };
        assert_eq!(
            filter_active_tool_names(names.clone(), &args),
            vec!["read", "write"]
        );

        let args = Args {
            no_tools: true,
            ..Args::default()
        };
        assert!(filter_active_tool_names(names, &args).is_empty());
    }

    #[test]
    fn select_ephemeral_when_no_session() {
        let args = Args {
            no_session: true,
            ..Args::default()
        };
        let cwd = Path::new("/tmp");
        assert!(matches!(
            select_session(&args, cwd),
            SessionSelection::Ephemeral
        ));
    }

    #[test]
    fn project_resources_load_by_default_and_allow_explicit_opt_out() {
        let defaults = Args::default();
        assert!(resolve_project_trust(
            &defaults,
            Path::new("C:/definitely-not-a-project")
        ));

        let denied = Args {
            trust_override: Some(false),
            ..Args::default()
        };
        assert!(!resolve_project_trust(
            &denied,
            Path::new("C:/definitely-not-a-project")
        ));
    }

    #[tokio::test(flavor = "current_thread")]
    async fn build_preserves_the_supplied_startup_project_snapshot() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        let cwd = tmp.path().join("project");
        std::fs::create_dir_all(&agent).unwrap();
        std::fs::create_dir_all(&cwd).unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);

        let resolved = crate::provider::resolve(
            Some("anthropic"),
            Some(crate::provider::DEFAULT_MODEL_ID),
            None,
            Some("test-key"),
            None,
        )
        .unwrap();
        let args = Args {
            trust_override: Some(false),
            no_session: true,
            no_tools: true,
            no_extensions: true,
            no_skills: true,
            no_prompt_templates: true,
            no_context_files: true,
            system_prompt: Some("test prompt".into()),
            ..Args::default()
        };

        let (_, _, context) = build(&resolved, &args, &cwd, true).await.unwrap();

        assert_eq!(context.cwd, cwd);
        assert!(context.project_trusted);
    }

    #[tokio::test(flavor = "current_thread")]
    async fn build_applies_the_name_flag_to_the_session() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        let cwd = tmp.path().join("project");
        std::fs::create_dir_all(&agent).unwrap();
        std::fs::create_dir_all(&cwd).unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);

        let resolved = crate::provider::resolve(
            Some("anthropic"),
            Some(crate::provider::DEFAULT_MODEL_ID),
            None,
            Some("test-key"),
            None,
        )
        .unwrap();
        let args = Args {
            trust_override: Some(false),
            no_tools: true,
            no_extensions: true,
            no_skills: true,
            no_prompt_templates: true,
            no_context_files: true,
            system_prompt: Some("test prompt".into()),
            name: Some("smoke name".into()),
            session_dir: Some(cwd.join("sessions")),
            ..Args::default()
        };

        let (harness, _, _) = build(&resolved, &args, &cwd, true).await.unwrap();
        assert_eq!(
            harness.get_name().await.unwrap().as_deref(),
            Some("smoke name")
        );
    }

    #[test]
    fn select_latest_for_continue_and_resume() {
        let _env = crate::config::test_support::SessionDirEnv::new(None);
        let args = Args {
            continue_session: true,
            ..Args::default()
        };
        let cwd = Path::new("/tmp");
        assert!(matches!(
            select_session(&args, cwd),
            SessionSelection::Latest { .. }
        ));

        let args = Args {
            resume: true,
            ..Args::default()
        };
        assert!(matches!(
            select_session(&args, cwd),
            SessionSelection::Latest { .. }
        ));
    }

    #[test]
    fn select_by_id_for_session_flag() {
        let _env = crate::config::test_support::SessionDirEnv::new(None);
        let args = Args {
            session: Some("01a02ece".into()),
            ..Args::default()
        };
        let cwd = Path::new("/tmp");
        assert!(matches!(
            select_session(&args, cwd),
            SessionSelection::ById { id, .. } if id == "01a02ece"
        ));
    }

    #[test]
    fn select_new_with_custom_dir() {
        let _env = crate::config::test_support::SessionDirEnv::new(None);
        let args = Args {
            session_dir: Some(PathBuf::from("/tmp/sess")),
            ..Args::default()
        };
        let cwd = Path::new("/tmp");
        match select_session(&args, cwd) {
            SessionSelection::New { dir, .. } => assert_eq!(dir, PathBuf::from("/tmp/sess")),
            other => panic!("expected New, got {other:?}"),
        }
    }

    #[test]
    fn select_new_default_dir() {
        let _env = crate::config::test_support::SessionDirEnv::new(None);
        let args = Args::default();
        let cwd = Path::new("/proj");
        match select_session(&args, cwd) {
            SessionSelection::New { dir, .. } => {
                assert_eq!(dir, Path::new("/proj/.rpi/sessions"));
            }
            other => panic!("expected New, got {other:?}"),
        }
    }

    #[test]
    fn default_session_dir_only_uses_rpi() {
        let _env = crate::config::test_support::SessionDirEnv::new(None);
        let tmp = tempfile::tempdir().unwrap();
        let cwd = tmp.path();
        std::fs::create_dir_all(cwd.join(".rpi/sessions")).unwrap();
        assert_eq!(default_session_dir(cwd), cwd.join(".rpi/sessions"));
        std::fs::create_dir_all(cwd.join(".rpi/sessions")).unwrap();
        assert_eq!(default_session_dir(cwd), cwd.join(".rpi/sessions"));
    }

    #[test]
    fn default_session_dir_uses_env_var() {
        let tmp = tempfile::tempdir().unwrap();
        let cwd = tmp.path();
        let custom = tmp.path().join("custom_sessions");
        let _env = crate::config::test_support::SessionDirEnv::new(Some(&custom));
        assert_eq!(default_session_dir(cwd), custom);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn ephemeral_session_builds_roundtrips() {
        // Sanity: the ephemeral path produces a usable Session facade (the
        // harness build itself needs a provider; tested via the integration
        // path in tests/build.rs instead).
        let s = ephemeral_session();
        let leaf = s.get_leaf_id().await;
        assert!(leaf.is_ok());
    }

    // NOTE: `build_tools`/`active_tool_names` integration is exercised by the
    // `tests/build.rs` harness-build test (needs a provider + multi-thread rt).

    /// The exclusion policy, checked against the constants rather than copies.
    ///
    /// `rpi-tools` builds its exclusion from `rpi-plugin-sdk`'s declarations, so
    /// within this repo a rename is a compile error — this test is not the link
    /// between those two crates any more. What it still guards is the part the
    /// compiler cannot see:
    ///
    /// - **The wire values.** `rpi-langfuse` lives in a different repository and
    ///   agrees on the *strings*, not the constant paths. Rewriting a constant's
    ///   value would compile here and silently stop matching there, so the exact
    ///   values are pinned below. Changing one is a deliberate, breaking change to
    ///   a cross-repo contract, and it should fail here first.
    /// - **Coverage.** Every name the SDK declares must actually be excluded.
    /// - **Narrowness.** The user's own configuration must still reach the shell.
    #[test]
    fn host_private_env_matches_the_declared_contract() {
        // 1. The wire values. Pinned literally on purpose: see the doc comment.
        assert_eq!(rpi_plugin_sdk::SESSION_ID_ENV, "RPI_SESSION_ID");
        assert_eq!(
            rpi_plugin_sdk::SUBAGENT_PARENT_ENV,
            [
                "LANGFUSE_PI_PARENT_TRACE_ID",
                "LANGFUSE_PI_PARENT_SPAN_ID",
                "LANGFUSE_PI_PARENT_SESSION_ID",
                "LANGFUSE_PI_PARENT_DEPTH",
            ],
            "these strings are what rpi-langfuse (a different repo) matches; a \
             change here is a breaking cross-repo change, not a refactor"
        );

        // 2. Coverage — derived from the declarations, so a newly added family
        //    member is checked without anyone remembering to extend this test.
        for name in rpi_plugin_sdk::SUBAGENT_PARENT_ENV
            .iter()
            .chain(std::iter::once(&rpi_plugin_sdk::SESSION_ID_ENV))
        {
            assert!(
                rpi_tools::is_host_private_env(name),
                "`{name}` must not reach a spawned shell, but rpi-tools excludes only {:?}",
                rpi_tools::HOST_PRIVATE_ENV
            );
        }

        // 3. Narrowness — these name how the *user* configured the run and are
        //    exactly what a command inside the shell should still see.
        for visible in [
            crate::args::RPI_OFFLINE_ENV,
            crate::config::CONFIG_DIR_ENV,
            "PATH",
        ] {
            assert!(
                !rpi_tools::is_host_private_env(visible),
                "{visible} describes the user's configuration and must still reach the shell"
            );
        }
    }
    #[tokio::test]
    async fn reload_shutdown_joins_all_handlers_even_after_veto() {
        use rpi_plugin_sdk::{EventTag, StablePluginEvent};
        use std::sync::atomic::AtomicUsize;
        static CONTEXTS: rpi_plugin_sdk::handles::ToolHandles<Arc<AtomicUsize>> =
            rpi_plugin_sdk::handles::ToolHandles::new();
        struct ContextOwner(usize);
        impl Drop for ContextOwner {
            fn drop(&mut self) {
                drop(CONTEXTS.remove(self.0 as rpi_plugin_sdk::StepHandle));
            }
        }
        extern "C" fn stop(_: StablePluginEvent, data: *mut std::ffi::c_void) -> i32 {
            let context = CONTEXTS.get(data).expect("live test context");
            let calls = context.lock().clone();
            std::thread::sleep(std::time::Duration::from_millis(20));
            if calls.fetch_add(1, Ordering::SeqCst) == 0 {
                rpi_plugin_sdk::EVENT_HANDLER_ABORT
            } else {
                0
            }
        }
        let calls = Arc::new(AtomicUsize::new(0));
        let context = CONTEXTS.insert(calls.clone());
        let _owner = ContextOwner(context as usize);
        let mut registry = rpi_extensions::ExtensionRegistry::new();
        for name in ["first", "second"] {
            registry.register_event_handler(name.into(), EventTag::SessionShutdown, stop, context);
        }
        super::shutdown_extensions_for_reload(
            Some(Arc::new(registry.snapshot())),
            rpi_extensions::PluginKeepalive::empty(),
        )
        .await
        .unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 2);
    }

    // ---- restored-session working directory ----

    /// A restore peek must name the directory the session was created in —
    /// that is the whole point of recording it. Everything downstream (tool
    /// env, system prompt, context files, extensions) keys off the run's cwd,
    /// so a peek that falls back to the launch directory would leave the
    /// resumed agent editing files in whatever directory `rpi` happened to be
    /// started from.
    #[tokio::test]
    async fn a_restored_session_adopts_its_recorded_directory() {
        let tmp = tempfile::tempdir().unwrap();
        let project = tmp.path().join("project-a");
        let launch = tmp.path().join("launch-dir");
        let sessions = tmp.path().join("sessions");
        std::fs::create_dir_all(&project).unwrap();
        std::fs::create_dir_all(&launch).unwrap();
        std::fs::create_dir_all(&sessions).unwrap();
        create_jsonl_session_with_id(
            &sessions,
            &project.to_string_lossy(),
            Some("cwdadopt01".into()),
        )
        .await
        .unwrap();

        let selection = SessionSelection::ById {
            id: "cwdadopt01".into(),
            dir: sessions.clone(),
        };
        let peeked = peek_restored_session_cwd(&selection, &launch)
            .await
            .expect("a resumable session with a recorded cwd");
        assert_eq!(peeked.recorded, project);
        assert_eq!(peeked.adopted, Some(project));
    }

    /// A recorded directory that no longer exists must NOT be adopted: the run
    /// falls back to the launch cwd, and the peek hands back the record so the
    /// host can say so (instead of the old behaviour of eprintln-ing mid-TUI).
    #[tokio::test]
    async fn a_missing_recorded_directory_is_reported_not_adopted() {
        let tmp = tempfile::tempdir().unwrap();
        let gone = tmp.path().join("project-gone");
        let launch = tmp.path().join("launch-dir");
        let sessions = tmp.path().join("sessions");
        std::fs::create_dir_all(&launch).unwrap();
        std::fs::create_dir_all(&sessions).unwrap();
        create_jsonl_session_with_id(
            &sessions,
            &gone.to_string_lossy(),
            Some("cwdgone001".into()),
        )
        .await
        .unwrap();

        let selection = SessionSelection::ById {
            id: "cwdgone001".into(),
            dir: sessions.clone(),
        };
        let peeked = peek_restored_session_cwd(&selection, &launch)
            .await
            .expect("the session itself is resumable");
        assert_eq!(peeked.recorded, gone);
        assert_eq!(peeked.adopted, None);
        let note = crate::session_cwd::missing_session_cwd_note(&peeked.recorded, &launch);
        assert!(note.contains("no longer exists"), "{note}");
        assert!(note.contains("launch-dir"), "{note}");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn legacy_progress_upgrade_normal_open_runs_and_reopens() {
        use rpi_ai::providers::faux::{FauxProvider, FauxScript};
        use rpi_harness::agent_harness::{AgentHarness, AgentLane, HarnessRunOutcome};
        use rpi_harness::session::jsonl::{
            encode_header, JsonlSessionCreateOptions, JsonlSessionRepo, JsonlSessionRepoOptions,
            JsonlV4Header,
        };
        let tmp = tempfile::tempdir().unwrap();
        let sessions = tmp.path().join("sessions");
        std::fs::create_dir_all(&sessions).unwrap();
        let cwd = tmp.path().to_string_lossy().into_owned();
        let repo = JsonlSessionRepo::with_env_cwd(JsonlSessionRepoOptions {
            fs: Arc::new(OsExecutionEnv::with_cwd(tmp.path().into())),
            sessions_root: sessions.to_string_lossy().into_owned(),
            clock: Arc::new(SystemClock),
            ids: Arc::new(DefaultIdGenerator::new()),
        });
        let fixture = repo
            .create_typed(&JsonlSessionCreateOptions {
                id: Some("legacycli001".into()),
                parent_session_id: None,
                cwd: cwd.clone(),
                metadata: None,
            })
            .await
            .unwrap();
        let path = PathBuf::from(fixture.jsonl_metadata().path);
        drop(fixture);
        let sidecar = PathBuf::from(format!("{}.progress.jsonl", path.display()));
        std::fs::remove_file(&sidecar).unwrap();
        let header = JsonlV4Header::new("legacycli001".into(), 1, cwd.clone(), None, None, None);
        std::fs::write(&path, encode_header(&header)).unwrap();
        assert!(!sidecar.exists());
        for _ in 0..2 {
            let restored = open_session_by_id_in_dir("legacycli001", &cwd, &sessions)
                .await
                .map_err(|e| e.to_string())
                .unwrap();
            assert_eq!(restored.session.storage().metadata().id, "legacycli001");
            let provider =
                FauxProvider::new(FauxScript::new().with_text("continued legacy session"));
            let harness = AgentHarness::create(AgentHarnessOptions {
                model: provider.default_model().clone(),
                models: vec![provider],
                session: restored.session,
                allow_existing_session: true,
                ..Default::default()
            })
            .await
            .unwrap();
            let run = harness.prompt_text("continue", vec![]).await.unwrap();
            assert!(matches!(run.outcome, HarnessRunOutcome::Completed { .. }));
        }
        let content = std::fs::read_to_string(&path).unwrap();
        assert!(content.lines().next().unwrap().contains("progressFormat"));
        assert!(sidecar.exists());
    }

    /// `open_session` hands back the recorded directory as data (for the
    /// `/session` hot-switch notes) while still opening the session normally.
    #[tokio::test]
    async fn open_session_reports_the_recorded_directory() {
        let tmp = tempfile::tempdir().unwrap();
        let project = tmp.path().join("project-open");
        let launch = tmp.path().join("launch-dir");
        let sessions = tmp.path().join("sessions");
        std::fs::create_dir_all(&project).unwrap();
        std::fs::create_dir_all(&launch).unwrap();
        std::fs::create_dir_all(&sessions).unwrap();
        create_jsonl_session_with_id(
            &sessions,
            &project.to_string_lossy(),
            Some("cwdopen001".into()),
        )
        .await
        .unwrap();

        let restored =
            open_session_by_id_in_dir("cwdopen001", &launch.to_string_lossy(), &sessions)
                .await
                .map_err(|error| error.to_string())
                .unwrap();
        assert_eq!(restored.recorded_cwd, Some(project.clone()));
        assert_eq!(restored.session.storage().metadata().id, "cwdopen001");
    }

    /// The session the peek inspected must be the session the build opens.
    ///
    /// `-c` means "the most recent session", and the build lists the directory
    /// again. Another `rpi` (or a `/session` new-session) can write a newer one
    /// between the peek and the build; adopting the peeked session's directory
    /// while opening the newer conversation would run the tools in a directory
    /// that conversation never used. Pinning the resolved id makes the second
    /// listing an exact match instead of a fresh newest-first choice.
    #[tokio::test]
    async fn the_peeked_session_is_the_one_the_build_opens() {
        let tmp = tempfile::tempdir().unwrap();
        let first_project = tmp.path().join("project-first");
        let second_project = tmp.path().join("project-second");
        let launch = tmp.path().join("launch-dir");
        let sessions = tmp.path().join("sessions");
        for dir in [&first_project, &second_project, &launch, &sessions] {
            std::fs::create_dir_all(dir).unwrap();
        }
        create_jsonl_session_with_id(
            &sessions,
            &first_project.to_string_lossy(),
            Some("cwdrace001".into()),
        )
        .await
        .unwrap();

        let mut args = Args {
            continue_session: true,
            session_dir: Some(sessions.clone()),
            ..Default::default()
        };
        let selection = select_session_in_dir(&args, sessions.clone());
        let peeked = peek_restored_session_cwd(&selection, &launch)
            .await
            .expect("a resumable session");
        assert_eq!(peeked.adopted, Some(first_project.clone()));

        // A newer session appears in another project before the build runs.
        create_jsonl_session_with_id(
            &sessions,
            &second_project.to_string_lossy(),
            Some("cwdrace002".into()),
        )
        .await
        .unwrap();

        pin_restored_selection(&mut args, &peeked);
        let pinned = select_session_in_dir(&args, sessions.clone());
        let reopened = resolve_restore_target(&pinned, &launch.to_string_lossy())
            .await
            .unwrap()
            .expect("the pinned session still exists");
        assert_eq!(
            reopened.id, peeked.id,
            "the build must open the session the peek adopted the directory of"
        );
        assert_eq!(reopened.cwd, first_project.to_string_lossy());
    }

    /// Trust must be resolved against the directory it is asked about. A
    /// session recorded in another project must not inherit the launch
    /// directory's decision — that would run an untrusted project's settings,
    /// skills, and context files with trusted privileges.
    #[test]
    fn trust_is_resolved_per_directory() {
        struct RestoreConfigDir(Option<std::ffi::OsString>);

        impl Drop for RestoreConfigDir {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(value) => std::env::set_var(crate::config::CONFIG_DIR_ENV, value),
                    None => std::env::remove_var(crate::config::CONFIG_DIR_ENV),
                }
            }
        }

        let _guard = crate::config::test_support::env_lock().lock().unwrap();
        let _restore = RestoreConfigDir(std::env::var_os(crate::config::CONFIG_DIR_ENV));
        let tmp = tempfile::tempdir().unwrap();
        let agent = tmp.path().join("agent");
        std::fs::create_dir_all(&agent).unwrap();
        std::env::set_var(crate::config::CONFIG_DIR_ENV, &agent);

        let untrusted = tmp.path().join("project-untrusted");
        let launch = tmp.path().join("project-launch");
        std::fs::create_dir_all(&untrusted).unwrap();
        std::fs::create_dir_all(&launch).unwrap();
        crate::config::set_project_trust(&untrusted, Some(false)).unwrap();

        let args = Args::default();
        assert!(!resolve_project_trust(&args, &untrusted));
        // A directory with no stored decision loads by default.
        assert!(resolve_project_trust(&args, &launch));
    }
}