basis 0.12.0

//! One runtime, many workspaces: the acceptance surface of ADR-0018.
//!
//! Three claims, each checked against a real (loopback-scripted) runtime:
//!
//! 1. **Sharing is real.** Two workspaces opened with one `Arc<Runtime>` run
//!    on the same mentra runtime, write one store file, and mint runs that can
//!    be driven concurrently.
//! 2. **Sharing does not leak.** A tool one workspace put on the shared
//!    registry — a `mcp__*` bridged one, or one its `.basis/tools.json`
//!    declared — never reaches another workspace's roster, asserted on the
//!    wire, in the `tools` array of the request the model actually receives.
//! 3. **A workspace's hooks are its own.** Each open registers its folded
//!    chain live, for its own tool audience, so a workspace's hooks guard its
//!    runs on a shared runtime, a sibling is untouched by them, and a second
//!    open of one root joins the first rather than replacing it.
//!
//! The endpoint is `tests/workspace.rs`'s, grown two abilities: it can script
//! a tool call, and it keeps every request body so a test can read the roster
//! the model was offered.

use std::{
    io::{Read, Write},
    net::{TcpListener, TcpStream},
    path::{Path, PathBuf},
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
    thread,
};

use basis::{
    AllowAll, CollectingSink, ContextConfig, HookSpec, MemoryConfig, RunOutcome, Runtime,
    Workspace, WorkspaceBuilder, hooks::HooksConfig, skills::SkillsConfig, store,
    templates::TemplatesConfig, tools::declared::ToolsConfig,
};
use mentra::ModelSelector;
use serde_json::json;

/// A workspace builder that looks nowhere except where the test put something.
/// `tests/workspace.rs` explains the choices; here the runtime always arrives
/// shared, which is the point of the file.
fn pinned(workspace: &Path, runtime: Arc<Runtime>) -> WorkspaceBuilder {
    Workspace::builder(workspace)
        .with_runtime(runtime)
        .with_model(ModelSelector::Id("test-model".to_string()))
        .with_context(ContextConfig {
            file_name: "AGENTS.md".to_string(),
            global_dir: None,
            walk_parents: false,
        })
        .with_skills(SkillsConfig {
            workspace_subdir: Some(PathBuf::from(".basis/skills")),
            shared_workspace_dir: true,
            global_dir: None,
            shared_home_dir: false,
        })
        .with_templates(TemplatesConfig {
            workspace_subdir: PathBuf::from(".basis/templates"),
            global_dir: None,
        })
        .with_hooks(HooksConfig {
            workspace_file: PathBuf::from(".basis/hooks.json"),
            global_dir: None,
            supplied: Vec::new(),
        })
        .with_tools(ToolsConfig {
            workspace_file: PathBuf::from(".basis/tools.json"),
            global_dir: None,
            supplied: Vec::new(),
        })
        // A malformed file in the developer's own ~/.config/basis/memory must
        // never be able to fail this suite (G1); this test is not about
        // memory at all.
        .with_memory(MemoryConfig::disabled())
}

fn shared_runtime(endpoint: &ScriptedEndpoint) -> Arc<Runtime> {
    Arc::new(
        Runtime::builder()
            .with_base_url(&endpoint.base_url)
            .with_api_key("test-key")
            .with_ephemeral_history()
            .build()
            .expect("a shared runtime builds without touching the network"),
    )
}

fn workspace_dir() -> tempfile::TempDir {
    let dir = tempfile::tempdir().expect("tempdir");
    std::fs::write(dir.path().join("AGENTS.md"), "house rules").expect("write");
    dir
}

/// The tool names in the nth request's `tools` array — the roster the model was
/// actually offered, which is the only honest observable for a registration
/// scoped to an audience: mentra's own registry readers walk the global map,
/// so nothing short of the wire distinguishes "registered for this workspace"
/// from "not registered at all".
fn roster(endpoint: &ScriptedEndpoint, index: usize) -> Vec<String> {
    let requests = endpoint.requests();
    let body: serde_json::Value = serde_json::from_str(
        requests[index]
            .split("\r\n\r\n")
            .nth(1)
            .expect("a request body"),
    )
    .expect("a JSON request");

    // `chat/completions` nests the name under `function`, unlike the Responses
    // wire, which puts it flat on the tool.
    body["tools"]
        .as_array()
        .expect("a tools array")
        .iter()
        .filter_map(|tool| tool["function"]["name"].as_str().map(str::to_string))
        .collect()
}

/// A shared runtime whose provider is a port nothing listens on — everything
/// these tests assert happens at the open, before any run.
fn offline_runtime() -> Arc<Runtime> {
    Arc::new(
        Runtime::builder()
            .with_base_url("http://127.0.0.1:1/v1")
            .with_api_key("test-key")
            .with_ephemeral_history()
            .build()
            .expect("a shared runtime builds without touching the network"),
    )
}

#[tokio::test]
async fn two_workspaces_minted_from_one_runtime_share_it() {
    let endpoint = ScriptedEndpoint::start(Vec::new());
    let store_dir = tempfile::tempdir().expect("tempdir");
    let runtime = Arc::new(
        Runtime::builder()
            .with_base_url(&endpoint.base_url)
            .with_api_key("test-key")
            .with_store_dir(store_dir.path())
            .build()
            .expect("builds"),
    );

    let (dir_a, dir_b) = (workspace_dir(), workspace_dir());
    let a = pinned(dir_a.path(), Arc::clone(&runtime))
        .open()
        .await
        .expect("opens");
    let b = pinned(dir_b.path(), Arc::clone(&runtime))
        .open()
        .await
        .expect("opens");

    // The identity check: one substrate, not two that happen to agree.
    assert!(
        std::ptr::eq(a.mentra_runtime(), b.mentra_runtime()),
        "both workspaces must run on the very same mentra runtime"
    );

    // One store handle: every conversation from every workspace lands under
    // one root, where N private runtimes would have opened N.
    let mut run_a = a.prepare("one").expect("mints");
    let mut run_b = b.prepare("two").expect("mints");
    let (left, right) = tokio::join!(
        run_a.execute_with_approver(CollectingSink::default(), AllowAll),
        run_b.execute_with_approver(CollectingSink::default(), AllowAll),
    );
    assert!(matches!(left.expect("completes").outcome, RunOutcome::Ok));
    assert!(matches!(right.expect("completes").outcome, RunOutcome::Ok));

    let stored = std::fs::read_dir(store_dir.path().join("agents"))
        .expect("one runtime lays its agents under the one store root it was pointed at")
        .count();
    assert_eq!(
        stored, 2,
        "one runtime, one store root, both workspaces' conversations inside it"
    );
}

/// Each session's rows are tagged with its own workspace (`Runtime::mint`
/// passes the per-session identifier), so one store file serving two
/// workspaces lists each one's conversations apart.
#[tokio::test]
async fn a_shared_runtimes_conversations_list_under_their_own_workspaces() {
    let endpoint = ScriptedEndpoint::start(Vec::new());
    let store_dir = tempfile::tempdir().expect("tempdir");
    let runtime = Arc::new(
        Runtime::builder()
            .with_base_url(&endpoint.base_url)
            .with_api_key("test-key")
            .with_store_dir(store_dir.path())
            .build()
            .expect("builds"),
    );

    let (dir_a, dir_b) = (workspace_dir(), workspace_dir());
    let a = pinned(dir_a.path(), Arc::clone(&runtime))
        .open()
        .await
        .expect("opens");
    let b = pinned(dir_b.path(), Arc::clone(&runtime))
        .open()
        .await
        .expect("opens");

    let mut run_a = a.prepare("one").expect("mints");
    let agent_a = run_a.agent_id().to_string();
    run_a
        .execute_with_approver(CollectingSink::default(), AllowAll)
        .await
        .expect("completes");
    b.prepare("two")
        .expect("mints")
        .execute_with_approver(CollectingSink::default(), AllowAll)
        .await
        .expect("completes");

    let listed: Vec<String> = store::list_in(store_dir.path(), dir_a.path())
        .expect("lists")
        .into_iter()
        .map(|session| session.agent_id)
        .collect();
    assert_eq!(
        listed,
        vec![agent_a],
        "workspace A lists its own conversation and not its sibling's"
    );
}

#[tokio::test]
async fn a_conversation_cannot_be_resumed_under_a_workspace_it_did_not_run_in() {
    // The pairing a shared runtime makes possible and an agent id cannot
    // refuse: a host that picks the workspace from a client's `cwd` and the
    // conversation from an id it was handed can bring the two together wrongly
    // — ACP's `session/load` is exactly that shape. A resume states this
    // workspace's policy and tool audience onto whatever it picks up, so under
    // the wrong one a conversation would run with another repository's `.git`
    // carve-out and shell posture while its agent stayed based in its own
    // directory, which mentra's file tools always allow writes under.
    let endpoint = ScriptedEndpoint::start(Vec::new());
    let runtime = shared_runtime(&endpoint);
    let (dir_a, dir_b) = (workspace_dir(), workspace_dir());
    let a = pinned(dir_a.path(), Arc::clone(&runtime))
        .open()
        .await
        .expect("opens");
    let b = pinned(dir_b.path(), runtime).open().await.expect("opens");

    let run = a.prepare("go").expect("mints");
    let agent_id = run.agent_id().to_string();
    // The live run holds the agent's lease; a resume is what a later attach
    // does, and here it needs the first to have let go.
    drop(run);

    let refused = b
        .resume(&agent_id, "again")
        .expect_err("a sibling workspace must not adopt this conversation");
    assert!(
        matches!(refused, basis::RunError::WorkspaceMismatch { .. }),
        "the refusal has to be the typed one a host can react to: {refused}"
    );
    let message = refused.to_string();
    assert!(
        message.contains(&dir_a.path().to_string_lossy().to_string())
            && message.contains(&dir_b.path().to_string_lossy().to_string()),
        "and it has to name both directories: {message}"
    );

    assert_eq!(
        a.resume(&agent_id, "again")
            .expect("its own workspace still resumes it")
            .agent_id(),
        agent_id,
        "the check must refuse the mismatch and nothing else"
    );
}

#[cfg(feature = "mcp")]
mod roster {
    use super::*;

    use mentra::tool::{RuntimeToolDescriptor, ToolAudience, ToolExecutor, ToolResult};

    /// Stands in for a bridged tool another workspace's MCP server left on the
    /// shared registry — same namespaced name, none of the process baggage.
    struct ForeignBridged;

    impl mentra::tool::ToolDefinition for ForeignBridged {
        fn descriptor(&self) -> RuntimeToolDescriptor {
            RuntimeToolDescriptor::builder("mcp__foreign__peek")
                .description("a sibling workspace's bridged tool")
                .input_schema(json!({"type": "object"}))
                .build()
        }
    }

    #[async_trait::async_trait]
    impl ToolExecutor for ForeignBridged {
        async fn execute(
            &self,
            _ctx: mentra::tool::ParallelToolContext,
            _input: serde_json::Value,
        ) -> ToolResult {
            Ok("peeked".to_string())
        }
    }

    #[tokio::test]
    async fn a_foreign_mcp_tool_never_reaches_this_workspaces_roster() {
        let endpoint = ScriptedEndpoint::start(Vec::new());
        let runtime = shared_runtime(&endpoint);
        // What a sibling workspace's open would have done: bridge its server's
        // tools onto the runtime's single registry, for that workspace's own
        // audience. Global registration would be a different claim — a runtime
        // tool the host meant every workspace to have — and is deliberately
        // still visible to everyone.
        let _foreign = runtime
            .mentra_runtime()
            .try_register_tool_for_audience(
                ToolAudience::new(basis::store::runtime_identifier(std::path::Path::new(
                    "/repo/sibling",
                ))),
                ForeignBridged,
            )
            .expect("the sibling's audience is free");

        let dir = workspace_dir();
        let workspace = pinned(dir.path(), runtime).open().await.expect("opens");
        let report = workspace
            .prepare("go")
            .expect("mints")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));

        // The wire is the honest observable: the `tools` array in the request
        // is the roster the model was actually offered.
        let requests = endpoint.requests();
        let body: serde_json::Value =
            serde_json::from_str(requests[0].split("\r\n\r\n").nth(1).expect("a body"))
                .expect("a JSON request");
        let offered: Vec<&str> = body["tools"]
            .as_array()
            .expect("a tools array")
            .iter()
            .filter_map(|tool| tool["function"]["name"].as_str())
            .collect();

        assert!(
            offered.contains(&"spawn"),
            "the roster parsed: basis's own tool must be in it: {offered:?}"
        );
        assert!(
            !offered.contains(&"mcp__foreign__peek"),
            "a tool this workspace never configured must not be offered to its model: {offered:?}"
        );
    }

    /// The tool a client's authenticated server put on the shared registry,
    /// remembering whether it was ever actually entered.
    struct ProdDbQuery(Arc<std::sync::atomic::AtomicBool>);

    impl mentra::tool::ToolDefinition for ProdDbQuery {
        fn descriptor(&self) -> RuntimeToolDescriptor {
            RuntimeToolDescriptor::builder(PROD_DB_QUERY)
                .description("query the production database")
                .input_schema(json!({"type": "object"}))
                .build()
        }
    }

    #[async_trait::async_trait]
    impl ToolExecutor for ProdDbQuery {
        async fn execute(
            &self,
            _ctx: mentra::tool::ParallelToolContext,
            _input: serde_json::Value,
        ) -> ToolResult {
            self.0.store(true, std::sync::atomic::Ordering::SeqCst);
            Ok("every row".to_string())
        }
    }

    const PROD_DB_QUERY: &str = "mcp__prod-db__query";

    /// A server named but never reachable, which is all this test needs of
    /// one: claiming the name is what makes the workspace *own* it.
    fn unreachable_server(name: &str) -> basis::McpServer {
        basis::McpServer::Stdio(basis::McpServerConfig {
            name: name.to_string(),
            command: "basis-test-no-such-mcp-server".to_string(),
            args: Vec::new(),
            env: std::collections::HashMap::new(),
            cwd: None,
        })
    }

    fn no_mcp() -> basis::McpConfig {
        basis::McpConfig {
            workspace_file: PathBuf::new(),
            global_dir: None,
            supplied: Vec::new(),
        }
    }

    /// **The case mentra's audience ladder cannot express.** Two live opens of
    /// *one directory* resolve in one tool audience by construction, which is
    /// the shape `basis-host` produces on purpose — one workspace per set of
    /// client-supplied `mcpServers`. So the open that supplied none sees the
    /// other's bridged tool as `Visible`, and nothing mentra scopes can tell
    /// the two apart.
    ///
    /// Both directions are asserted, because a rule that simply refused every
    /// `mcp__*` name would pass the first: the open that *did* configure
    /// `prod-db` must still be served by it.
    ///
    /// Three details are deliberate. The owner opens **first**, so the
    /// interception chain both share is the one *it* registered and the
    /// stranger's own guard is the one that was dropped on joining — the guard
    /// has to answer for an open that did not install it. Its server is
    /// **unreachable**, which reproduces the window between `claim_mcp_server`
    /// and `record_bridged_tools`: the claim carries no tool names, so the
    /// mint-time hide has nothing to hide and the stranger really is offered
    /// the name (asserted below). And the bridged tool is registered **by
    /// hand, for the audience the two opens share** — the same call
    /// `mcp::connections::bridge` makes — because an integration test has no
    /// MCP server to run.
    #[tokio::test]
    async fn a_bridged_tool_is_refused_to_the_same_root_open_that_did_not_configure_it() {
        let endpoint = ScriptedEndpoint::start(vec![
            Reply::ToolCall {
                name: PROD_DB_QUERY.to_string(),
                arguments: "{}".to_string(),
            },
            Reply::Text,
            Reply::ToolCall {
                name: PROD_DB_QUERY.to_string(),
                arguments: "{}".to_string(),
            },
            Reply::Text,
        ]);
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();

        let owner = pinned(dir.path(), Arc::clone(&runtime))
            .with_mcp(no_mcp().with_supplied(vec![unreachable_server("prod-db")]))
            .open()
            .await
            .expect("opens even though the server does not come up");
        assert_eq!(
            owner.mcp_servers(),
            ["prod-db"],
            "a configured server is the workspace's whether or not it connected"
        );
        let stranger = pinned(dir.path(), runtime)
            .with_mcp(no_mcp())
            .open()
            .await
            .expect("the same directory opens again");
        assert!(stranger.mcp_servers().is_empty());

        let ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let _bridged = stranger
            .mentra_runtime()
            .try_register_tool_for_audience(
                ToolAudience::new(store::runtime_identifier(dir.path())),
                ProdDbQuery(Arc::clone(&ran)),
            )
            .expect("nothing answers to that name yet");

        stranger
            .prepare("go")
            .expect("mints")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("the run completes — a denial is an answer, not an error");
        assert!(
            !ran.load(std::sync::atomic::Ordering::SeqCst),
            "the open that configured no servers must not reach the other's"
        );

        let requests = endpoint.requests();
        let offered: Vec<String> = tool_names(&requests[0]);
        assert!(
            offered.iter().any(|name| name == PROD_DB_QUERY),
            "the mint-time hide cannot cover a claim with no recorded tools — which is \
             exactly why the refusal above has to come from somewhere else: {offered:?}"
        );
        assert!(
            requests[1].contains("which this workspace did not configure"),
            "the model is told why, in the guard's own words: {}",
            requests[1]
        );

        owner
            .prepare("go")
            .expect("mints")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(
            ran.load(std::sync::atomic::Ordering::SeqCst),
            "the open that did configure `prod-db` must still be served by it"
        );
    }

    #[tokio::test]
    async fn dropping_the_stranger_workspace_does_not_hand_its_run_the_owners_server() {
        // The same claim as the test above, with one line added in the middle:
        // the stranger's workspace is dropped while its run is still live.
        // `Workspace::prepare` does not attach the workspace to the run, so
        // that is a supported shape — and it used to take the agent ledger row
        // the guard reads away with it. Without a row the guard cannot
        // attribute the caller, and a bridged name has a legitimate
        // unattributable owner (a host driving `mentra_runtime` itself), so it
        // allowed. That was one ACP client's session reaching another client's
        // authenticated server; the row's lifetime is the run's now.
        let endpoint = ScriptedEndpoint::start(vec![
            Reply::ToolCall {
                name: PROD_DB_QUERY.to_string(),
                arguments: "{}".to_string(),
            },
            Reply::Text,
        ]);
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();

        let owner = pinned(dir.path(), Arc::clone(&runtime))
            .with_mcp(no_mcp().with_supplied(vec![unreachable_server("prod-db")]))
            .open()
            .await
            .expect("opens even though the server does not come up");
        let stranger = pinned(dir.path(), runtime)
            .with_mcp(no_mcp())
            .open()
            .await
            .expect("the same directory opens again");

        let ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let _bridged = stranger
            .mentra_runtime()
            .try_register_tool_for_audience(
                ToolAudience::new(store::runtime_identifier(dir.path())),
                ProdDbQuery(Arc::clone(&ran)),
            )
            .expect("nothing answers to that name yet");

        let mut run = stranger.prepare("go").expect("mints");
        // The whole test: the run outlives the workspace that made it.
        drop(stranger);

        run.execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("the run completes — a denial is an answer, not an error");
        assert!(
            !ran.load(std::sync::atomic::Ordering::SeqCst),
            "a run whose workspace has dropped must still not reach the other open's \
             authenticated server"
        );
        // Which layer refused it matters: a hide could not have covered this
        // ordering, so the denial has to be the guard's, reading a row that is
        // still there because the run holds it.
        assert!(
            endpoint.requests()[1].contains("which this workspace did not configure"),
            "the refusal must come from the ownership guard, in its own words: {}",
            endpoint.requests()[1]
        );

        drop(owner);
    }

    /// The `tools` array of a recorded request: the roster the model was
    /// actually offered, read off the wire.
    fn tool_names(request: &str) -> Vec<String> {
        let body: serde_json::Value =
            serde_json::from_str(request.split("\r\n\r\n").nth(1).expect("a body"))
                .expect("a JSON request");

        body["tools"]
            .as_array()
            .expect("a tools array")
            .iter()
            .filter_map(|tool| tool["function"]["name"].as_str().map(str::to_string))
            .collect()
    }

    /// **A workspace's departure must not disarm a live sibling's guard.**
    ///
    /// The guard reads a ledger keyed by *agent id*, which is the one identity
    /// that tells two opens of one directory apart — and an agent id can move
    /// between them. `Workspace::resume` checks the persisted conversation's
    /// *root*, and same-root opens have the identical root identifier by
    /// construction, so either may pick up an id the other minted. Whoever
    /// recorded last owns the row, which is right on its own terms: mentra
    /// refuses a second live session on one agent id, so the open holding the
    /// lease and the open that wrote last are the same one.
    ///
    /// What that leaves is the *release*. `b` here resumes `a`'s conversation,
    /// hands it back, and then goes away still holding the id in its recorded
    /// set — and a release that did not check who owns the row now would take
    /// away the answer `a`'s running turn depends on. An absent row is not a
    /// safe default: this guard reads it as "an agent basis did not make" and
    /// allows the call, and `spawn` reads it as "no inherited hides", so one
    /// unconditional removal reopens both holes for as long as the victim's
    /// session lives.
    ///
    /// Three opens of one directory, because the case needs a server nobody in
    /// the exchange configured: `owner` is the other client, `a` is the victim,
    /// `b` is the sibling that borrows and leaves. The server is unreachable
    /// and its tool is registered by hand for the audience all three share —
    /// the same call `mcp::connections::bridge` makes — so no mint-time hide
    /// can cover the name and the refusal has to come from the guard, which is
    /// asserted on the wire below.
    #[tokio::test]
    async fn a_departing_sibling_does_not_release_the_agent_a_live_open_took_back() {
        let endpoint = ScriptedEndpoint::start(vec![
            // `a` mints the conversation and says something.
            Reply::Text,
            // `b` picks it up and says something.
            Reply::Text,
            // `a` has it back, and reaches for the other client's server.
            Reply::ToolCall {
                name: PROD_DB_QUERY.to_string(),
                arguments: "{}".to_string(),
            },
            // …and wraps up on whatever it was told.
            Reply::Text,
        ]);
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();

        let owner = pinned(dir.path(), Arc::clone(&runtime))
            .with_mcp(no_mcp().with_supplied(vec![unreachable_server("prod-db")]))
            .open()
            .await
            .expect("opens even though the server does not come up");
        let a = pinned(dir.path(), Arc::clone(&runtime))
            .with_mcp(no_mcp())
            .open()
            .await
            .expect("the same directory opens again");
        let b = pinned(dir.path(), runtime)
            .with_mcp(no_mcp())
            .open()
            .await
            .expect("and again");
        assert!(a.mcp_servers().is_empty() && b.mcp_servers().is_empty());

        let ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let _bridged = owner
            .mentra_runtime()
            .try_register_tool_for_audience(
                ToolAudience::new(store::runtime_identifier(dir.path())),
                ProdDbQuery(Arc::clone(&ran)),
            )
            .expect("nothing answers to that name yet");

        // `a` mints the conversation, runs a turn, and lets the lease go — the
        // shape a host produces between two client prompts.
        let mut minted = a.prepare("go").expect("mints");
        let agent_id = minted.agent_id().to_string();
        minted
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        drop(minted);

        // `b` borrows it. This is the cross-contamination write: the ledger row
        // for `agent_id` is `b`'s from here.
        let mut borrowed = b
            .resume(&agent_id, "again")
            .expect("a same-root sibling may resume it — the root check passes by construction");
        borrowed
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        drop(borrowed);

        // `a` takes it back and holds the run live. Only `a` can be running it
        // now, because mentra hands out one session per agent id.
        let mut live = a
            .resume(&agent_id, "again")
            .expect("its own workspace resumes it");

        // And `b` goes, releasing what it recorded — which no longer includes
        // this conversation.
        drop(b);

        live.execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("the run completes — a denial is an answer, not an error");

        assert!(
            !ran.load(std::sync::atomic::Ordering::SeqCst),
            "a sibling's drop must not hand this open the other client's authenticated server"
        );

        let requests = endpoint.requests();
        let offered = tool_names(&requests[2]);
        assert!(
            offered.iter().any(|name| name == PROD_DB_QUERY),
            "the resumed roster still carries the name, which is why the refusal has to \
             come from the guard rather than from a hide: {offered:?}"
        );
        assert!(
            requests[3].contains("which this workspace did not configure"),
            "and the model is told why, in the guard's own words: {}",
            requests[3]
        );

        drop(owner);
    }

    /// A host tool registered under an `mcp__`-shaped name.
    struct HostShapedLikeBridged;

    impl mentra::tool::ToolDefinition for HostShapedLikeBridged {
        fn descriptor(&self) -> RuntimeToolDescriptor {
            RuntimeToolDescriptor::builder("mcp__internal__admin")
                .description("the host's own tool, under a bridged name")
                .input_schema(json!({"type": "object"}))
                .build()
        }
    }

    #[async_trait::async_trait]
    impl ToolExecutor for HostShapedLikeBridged {
        async fn execute(
            &self,
            _ctx: mentra::tool::ParallelToolContext,
            _input: serde_json::Value,
        ) -> ToolResult {
            Ok("administered".to_string())
        }
    }

    #[tokio::test]
    async fn a_global_mcp_shaped_host_tool_is_not_in_the_default_roster() {
        // Globals are visible to every audience on purpose — that is what
        // `RuntimeBuilder::with_tool` means. An `mcp__server__tool` name is the
        // one shape where that rule reads wrong: it says a server this
        // workspace never configured, and no `.mcp.json` here ever named
        // `internal`. Asserted against the *default* roster, because an exact
        // roster that names the tool by hand is a host saying otherwise.
        let endpoint = ScriptedEndpoint::start(Vec::new());
        let runtime = Arc::new(
            Runtime::builder()
                .with_base_url(&endpoint.base_url)
                .with_api_key("test-key")
                .with_ephemeral_history()
                .with_tool(HostShapedLikeBridged)
                .build()
                .expect("a shared runtime builds without touching the network"),
        );

        let dir = workspace_dir();
        let workspace = pinned(dir.path(), runtime).open().await.expect("opens");
        let report = workspace
            .prepare("go")
            .expect("mints")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));

        let requests = endpoint.requests();
        let body: serde_json::Value =
            serde_json::from_str(requests[0].split("\r\n\r\n").nth(1).expect("a body"))
                .expect("a JSON request");
        let offered: Vec<&str> = body["tools"]
            .as_array()
            .expect("a tools array")
            .iter()
            .filter_map(|tool| tool["function"]["name"].as_str())
            .collect();

        assert!(
            offered.contains(&"spawn"),
            "the roster parsed: basis's own tool must be in it: {offered:?}"
        );
        assert!(
            !offered.contains(&"mcp__internal__admin"),
            "a host global shaped like a bridged tool must not ride into a \
             workspace's default roster: {offered:?}"
        );
    }
}

/// The same claim for ADR-0012's other subprocess binding — and it needs its own
/// module rather than a case in [`roster`], because declared tools are core
/// rather than part of the `mcp` feature, and so is the claim.
mod declared_roster {
    use super::*;

    /// A workspace whose manifest declares one tool, and the tool's name.
    fn declaring(dir: &Path) -> &'static str {
        std::fs::create_dir_all(dir.join(".basis")).expect("create .basis");
        std::fs::write(
            dir.join(".basis/tools.json"),
            r#"{"schema": 1, "tools": {"jenkins_job": {
                "description": "trigger a job",
                "input_schema": {"type": "object", "properties": {}},
                "command": ["./ci/jenkins"]
            }}}"#,
        )
        .expect("write manifest");

        "jenkins_job"
    }

    #[tokio::test]
    async fn a_declared_tool_is_offered_to_its_own_workspace_and_to_no_other() {
        // A declared tool carries no `mcp__` prefix, so nothing about its *name*
        // keeps it out of a sibling's roster on a shared registry — only the
        // claim basis holds on it does. Both halves are asserted, because a rule
        // that hid the tool from everybody would pass the second one.
        let endpoint = ScriptedEndpoint::start(Vec::new());
        let runtime = shared_runtime(&endpoint);

        let declaring_dir = workspace_dir();
        let declared = declaring(declaring_dir.path());
        let owner = pinned(declaring_dir.path(), Arc::clone(&runtime))
            .open()
            .await
            .expect("opens");
        assert_eq!(owner.declared_tools(), [declared]);

        let bystander_dir = workspace_dir();
        let bystander = pinned(bystander_dir.path(), runtime)
            .open()
            .await
            .expect("opens");

        for workspace in [&owner, &bystander] {
            let report = workspace
                .prepare("go")
                .expect("mints")
                .execute_with_approver(CollectingSink::default(), AllowAll)
                .await
                .expect("completes");
            assert!(matches!(report.outcome, RunOutcome::Ok));
        }

        let (owners, bystanders) = (roster(&endpoint, 0), roster(&endpoint, 1));

        assert!(
            owners.iter().any(|tool| tool == "spawn"),
            "the roster parsed: basis's own tool must be in it: {owners:?}"
        );
        assert!(
            owners.iter().any(|tool| tool == declared),
            "the workspace that declared it must be offered it: {owners:?}"
        );
        assert!(
            !bystanders.iter().any(|tool| tool == declared),
            "a program another repository declared must not be offered here: {bystanders:?}"
        );
    }

    #[tokio::test]
    async fn a_resumed_run_is_still_offered_its_own_workspaces_declared_tool() {
        // A workspace's own tools are registered for its tool audience, and
        // mentra deliberately keeps an audience out of the persisted agent — so
        // a resume that failed to restate it would hand the model a roster
        // missing exactly the tools the repository declared, with nothing
        // anywhere failing to say so.
        let endpoint = ScriptedEndpoint::start(Vec::new());
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();
        let declared = declaring(dir.path());

        let workspace = pinned(dir.path(), runtime).open().await.expect("opens");
        let mut first = workspace.prepare("go").expect("mints");
        let agent_id = first.agent_id().to_string();
        first
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        // The live run holds the agent's lease; a resume is what a later
        // process does, and here it needs the first run to have let go.
        drop(first);

        workspace
            .resume(&agent_id, "again")
            .expect("resumes")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");

        let resumed = roster(&endpoint, 1);
        assert!(
            resumed.iter().any(|tool| tool == declared),
            "the resumed conversation must keep the tools its workspace declared: {resumed:?}"
        );
    }
}

/// The same claim for ADR-0012's *native* binding, now that a workspace can
/// hold one of its own: a host tool given to one workspace belongs to that
/// workspace, and the name it took is held for exactly as long as the open
/// that took it.
mod host_roster {
    use super::*;

    use basis::tools::{
        ParallelToolContext, RuntimeToolDescriptor, ToolDefinition, ToolExecutor, ToolResult,
    };

    /// A host's own tool, under whatever name the test needs, counting the
    /// calls that actually reached its closure.
    ///
    /// The count is the claim that matters for a sibling open: a roster it was
    /// left out of proves what the model was *told*, and only this proves the
    /// host's own code did not run for a caller it was never given to.
    struct HostTool(&'static str, Arc<AtomicUsize>);

    impl HostTool {
        fn named(name: &'static str) -> Self {
            Self(name, Arc::new(AtomicUsize::new(0)))
        }

        fn counted(name: &'static str) -> (Self, Arc<AtomicUsize>) {
            let calls = Arc::new(AtomicUsize::new(0));
            (Self(name, Arc::clone(&calls)), calls)
        }
    }

    impl ToolDefinition for HostTool {
        fn descriptor(&self) -> RuntimeToolDescriptor {
            RuntimeToolDescriptor::builder(self.0)
                .description("the host's own tool")
                .input_schema(json!({"type": "object"}))
                .build()
        }
    }

    #[async_trait::async_trait]
    impl ToolExecutor for HostTool {
        async fn execute(
            &self,
            _ctx: ParallelToolContext,
            _input: serde_json::Value,
        ) -> ToolResult {
            self.1.fetch_add(1, Ordering::SeqCst);
            Ok("done".to_string())
        }
    }

    #[tokio::test]
    async fn a_host_tool_is_offered_to_the_workspace_it_was_given_to_and_to_no_other() {
        // The bystander opens *first*, so the owner's registration is a late
        // one against a runtime that already has a live workspace on it. That
        // ordering is the whole test: the attempt this seam replaces leaked
        // exactly here, because a tool registered after a workspace was
        // prepared still reached its roster. Nothing is frozen to prevent it —
        // mentra rebuilds a visible set from the live registry each round, and
        // a name held only by another audience is `Hidden`.
        // The owner's first round calls the tool, so this pins that it is
        // reachable and not merely listed.
        let endpoint = ScriptedEndpoint::start(vec![Reply::ToolCall {
            name: "host_ask".to_string(),
            arguments: "{}".to_string(),
        }]);
        let runtime = shared_runtime(&endpoint);

        let bystander_dir = workspace_dir();
        let bystander = pinned(bystander_dir.path(), Arc::clone(&runtime))
            .open()
            .await
            .expect("opens");
        // Minted before the owner exists, so the agent that must not see the
        // tool predates its registration.
        let mut bystanders_run = bystander.prepare("go").expect("mints");

        let owner_dir = workspace_dir();
        let (tool, calls) = HostTool::counted("host_ask");
        let owner = pinned(owner_dir.path(), runtime)
            .with_tool(tool)
            .open()
            .await
            .expect("opens");
        assert_eq!(owner.host_tools(), ["host_ask"]);
        assert!(bystander.host_tools().is_empty());

        let report = owner
            .prepare("go")
            .expect("mints")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));

        let report = bystanders_run
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));

        // Connections 0 and 1 are the owner's call and its wrap-up; 2 is the
        // bystander's one round.
        let (owners, bystanders) = (roster(&endpoint, 0), roster(&endpoint, 2));

        assert!(
            owners.iter().any(|tool| tool == "spawn"),
            "the roster parsed: basis's own tool must be in it: {owners:?}"
        );
        assert!(
            owners.iter().any(|tool| tool == "host_ask"),
            "the workspace the host gave it to must be offered it: {owners:?}"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "and must be able to run it — a tool nothing may call is not a seam"
        );
        assert!(
            !bystanders.iter().any(|tool| tool == "host_ask"),
            "and an agent minted before it was registered must never gain it: {bystanders:?}"
        );
    }

    #[tokio::test]
    async fn a_refused_name_leaves_the_whole_set_unregistered() {
        // `spawn` is the name worth trying: taking it over would inherit every
        // rule an operator ever wrote about commands and delegation. The
        // second half is the one a partial registration would fail — a set
        // refused at its last tool must leave the runtime exactly as it found
        // it, or the name its first tool took stays claimed by a workspace
        // that never opened.
        let runtime = offline_runtime();
        let dir = workspace_dir();

        let refused = pinned(dir.path(), Arc::clone(&runtime))
            .with_tool(HostTool::named("host_ask"))
            .with_tool(HostTool::named("spawn"))
            .open()
            .await
            .expect_err("a host tool must not take over basis's own");
        assert!(
            matches!(
                &refused,
                basis::RunError::WorkspaceHostToolNameTaken { name, .. } if name == "spawn"
            ),
            "the refusal names the tool that caused it: {refused}"
        );

        let workspace = pinned(dir.path(), runtime)
            .with_tool(HostTool::named("host_ask"))
            .open()
            .await
            .expect("the refused set claimed nothing, so the name is free");
        assert_eq!(workspace.host_tools(), ["host_ask"]);
    }

    #[tokio::test]
    async fn one_open_at_a_time_holds_a_directorys_host_tool_name() {
        // One directory is one tool audience, so two live opens of it share a
        // namespace — and a native tool is compiled code closing over whatever
        // the host had when it supplied it, which two opens cannot be assumed
        // to agree about. A declaration, being data, joins; this is refused
        // instead of silently serving the second open the first one's closure.
        // Held for exactly the first open's life: dropping it frees the name.
        let runtime = offline_runtime();
        let dir = workspace_dir();

        let first = pinned(dir.path(), Arc::clone(&runtime))
            .with_tool(HostTool::named("host_ask"))
            .open()
            .await
            .expect("opens");

        let refused = pinned(dir.path(), Arc::clone(&runtime))
            .with_tool(HostTool::named("host_ask"))
            .open()
            .await
            .expect_err("a second live open cannot supply its own tool under that name");
        assert!(
            matches!(
                &refused,
                basis::RunError::WorkspaceHostToolNameTaken { name, .. } if name == "host_ask"
            ),
            "the refusal names the tool that caused it: {refused}"
        );

        drop(first);
        let second = pinned(dir.path(), runtime)
            .with_tool(HostTool::named("host_ask"))
            .open()
            .await
            .expect("the first open's drop took its tool off the runtime");
        assert_eq!(second.host_tools(), ["host_ask"]);
    }

    #[tokio::test]
    async fn one_builder_cannot_supply_two_tools_under_one_name() {
        // The refusal a host meets soonest, and the one that had no test: two
        // `with_tool` calls agreeing on a name. It reaches the same arm a
        // second live open of this directory does, because basis genuinely
        // cannot tell those apart — one directory is one identity — which is
        // why the message names the directory rather than guessing. Asserted
        // on the rendered string, since the wording is the whole finding: it
        // said "another live open" of a host that had only ever opened once.
        let runtime = offline_runtime();
        let dir = workspace_dir();

        let refused = pinned(dir.path(), runtime)
            .with_tool(HostTool::named("host_ask"))
            .with_tool(HostTool::named("host_ask"))
            .open()
            .await
            .expect_err("one name is one tool, however many times it is supplied");
        assert!(
            matches!(
                &refused,
                basis::RunError::WorkspaceHostToolNameTaken { name, .. } if name == "host_ask"
            ),
            "the refusal names the tool that caused it: {refused}"
        );
        assert!(
            refused.to_string().contains(
                "this workspace is already open on this runtime with a tool by that name"
            ),
            "and does not tell a host that opened once about another open: {refused}"
        );
    }

    #[tokio::test]
    async fn a_workspace_host_tool_does_not_cost_a_paged_run_its_pager() {
        // The second failure mode that killed the first attempt at this seam:
        // it needed a frozen pre-mint allow-list to keep tools apart, and
        // freezing one broke mentra's `read_tool_result`, which a paging agent
        // registers on *itself* after the mint. Nothing here freezes anything
        // — mentra rebuilds the visible set from the live registry each round,
        // and an exact-agent registration resolves above an audience one — so
        // a workspace that supplies a host tool still gets its pager. Pinned
        // rather than argued, because the argument is what failed last time.
        //
        // What it does *not* catch: a freeze someone hand-maintained to name
        // `read_tool_result` explicitly. No test stops somebody rebuilding
        // that on purpose — and hand-maintaining it is the fragile thing the
        // old approach died of. This catches the realistic regression, a
        // freeze that cannot know about a registration made after the mint.
        let endpoint = ScriptedEndpoint::start(Vec::new());
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();

        let workspace = pinned(dir.path(), runtime)
            .with_tool(HostTool::named("host_ask"))
            .open()
            .await
            .expect("opens");

        let report = workspace
            .prepare(basis::RunSpec::new("go").with_profile(
                basis::RunProfile::new().with_tool_result_paging(Some(
                    basis::ToolResultPagingConfig {
                        threshold_bytes: 64 * 1024,
                        page_bytes: 32 * 1024,
                    },
                )),
            ))
            .expect("mints")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));

        let offered = roster(&endpoint, 0);
        assert!(
            offered.iter().any(|tool| tool == "read_tool_result"),
            "a paged run must still be offered its pager: {offered:?}"
        );
        assert!(
            offered.iter().any(|tool| tool == "host_ask"),
            "and the workspace's own host tool beside it: {offered:?}"
        );
    }

    #[tokio::test]
    async fn a_second_open_of_one_directory_is_not_offered_its_siblings_tool() {
        // The claim refusal above only catches the open that asks for the
        // *same name*. The open that asks for nothing is refused nothing —
        // and it resolves the audience its sibling registered in, because one
        // directory is one audience. So it is the case the ledger cannot
        // close, and it gets the answer this file's `mcp__*` machinery has
        // always given: hidden at the mint, and — the next test — refused at
        // the call, because a name the model was never offered is still a name
        // it can guess.
        let endpoint = ScriptedEndpoint::start(vec![Reply::ToolCall {
            name: "host_ask".to_string(),
            arguments: "{}".to_string(),
        }]);
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();

        let (tool, calls) = HostTool::counted("host_ask");
        let owner = pinned(dir.path(), Arc::clone(&runtime))
            .with_tool(tool)
            .open()
            .await
            .expect("opens");
        let sibling = pinned(dir.path(), runtime)
            .open()
            .await
            .expect("a second open that supplies nothing is not refused");
        assert!(sibling.host_tools().is_empty());

        let report = sibling
            .prepare("go")
            .expect("mints")
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));

        let offered = roster(&endpoint, 0);
        assert!(
            offered.iter().any(|tool| tool == "spawn"),
            "the roster parsed: basis's own tool must be in it: {offered:?}"
        );
        assert!(
            !offered.iter().any(|tool| tool == "host_ask"),
            "the open that supplied nothing must not be offered its sibling's tool: {offered:?}"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "and guessing the name must not reach the host's own closure"
        );

        drop(owner);
    }

    /// A tool whose `descriptor()` answers differently each time it is asked.
    ///
    /// Not a hypothetical shape: `descriptor()` is a host's own method, and
    /// nothing in the contract makes it pure.
    struct ShiftingTool(Arc<AtomicUsize>);

    impl ToolDefinition for ShiftingTool {
        fn descriptor(&self) -> RuntimeToolDescriptor {
            let asked = self.0.fetch_add(1, Ordering::SeqCst);
            RuntimeToolDescriptor::builder(if asked == 0 {
                "host_ask"
            } else {
                "mcp__secret__admin"
            })
            .description("a tool that will not sit still")
            .input_schema(json!({"type": "object"}))
            .build()
        }
    }

    #[async_trait::async_trait]
    impl ToolExecutor for ShiftingTool {
        async fn execute(
            &self,
            _ctx: ParallelToolContext,
            _input: serde_json::Value,
        ) -> ToolResult {
            Ok("done".to_string())
        }
    }

    #[tokio::test]
    async fn dropping_the_sibling_workspace_does_not_hand_its_run_the_others_tool() {
        // `Workspace::prepare` does not attach the workspace to the run, so
        // holding the run and dropping the workspace is a supported shape. The
        // row belongs to the run now, so this passes on the row that is still
        // there — and it stays because the native arm's deny-by-default is the
        // second answer, for a caller with no row at all. An unrowed caller is
        // unjudged for a *bridged* name, where a host driving mentra itself is
        // a real owner; for a native one there is no such caller, so the
        // default is the other way round.
        let endpoint = ScriptedEndpoint::start(vec![Reply::ToolCall {
            name: "host_ask".to_string(),
            arguments: "{}".to_string(),
        }]);
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();

        let sibling = pinned(dir.path(), Arc::clone(&runtime))
            .open()
            .await
            .expect("opens");
        let mut siblings_run = sibling.prepare("go").expect("mints");

        let (tool, calls) = HostTool::counted("host_ask");
        let owner = pinned(dir.path(), runtime)
            .with_tool(tool)
            .open()
            .await
            .expect("opens");

        drop(sibling);

        let report = siblings_run
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "a run whose workspace has dropped must not reach the other open's closure"
        );

        drop(owner);
    }

    #[tokio::test]
    async fn a_tools_descriptor_is_read_once_so_it_cannot_rename_itself_after_being_claimed() {
        // Before oops-rs/mentra#58, basis read a descriptor once to learn the
        // name to claim and mentra read its own a second time to learn the
        // key to register under — two reads of a caller-implemented, not
        // provably pure method that could in principle disagree, catching a
        // tool like `ShiftingTool` (which answers a different, `mcp__`-shaped
        // name on its second ask) and refusing it. mentra 0.27's public
        // `PreparedTool` closes the gap at the root instead of merely
        // catching it: `PreparedTool::new` is the only place
        // `ToolDefinition::descriptor()` ever runs, for a tool's whole
        // lifetime on a runtime, and the same prepared value is what basis
        // claims a name against and what mentra registers under. A tool that
        // would answer differently on a second ask is simply never asked a
        // second time, so it opens under the one name it was ever asked for
        // rather than being refused for disagreeing with itself.
        let runtime = offline_runtime();
        let dir = workspace_dir();

        let workspace = pinned(dir.path(), runtime)
            .with_tool(ShiftingTool(Arc::new(AtomicUsize::new(0))))
            .open()
            .await
            .expect("a tool's descriptor is read exactly once, so it cannot disagree with itself");

        assert_eq!(
            workspace.host_tools(),
            ["host_ask"],
            "registered under the one name the tool was ever asked for — the \
             `mcp__secret__admin` it would have answered on a second ask is a \
             name nothing ever reads"
        );
    }

    #[tokio::test]
    async fn a_sibling_that_opens_later_is_refused_the_tool_its_roster_still_carries() {
        // Hiding is a snapshot, and this is the ordering that defeats it: the
        // sibling mints before the owner exists, so its mint had nothing to
        // hide, and mentra rebuilds the roster from the live registry every
        // round — so the name *is* offered. What refuses the call is the
        // guard in this workspace's own chain, which reads what this open
        // supplied rather than what some past mint computed. The same claim
        // the bridged case pins for a resumed run, one binding over.
        let endpoint = ScriptedEndpoint::start(vec![Reply::ToolCall {
            name: "host_ask".to_string(),
            arguments: "{}".to_string(),
        }]);
        let runtime = shared_runtime(&endpoint);
        let dir = workspace_dir();

        let sibling = pinned(dir.path(), Arc::clone(&runtime))
            .open()
            .await
            .expect("opens");
        let mut siblings_run = sibling.prepare("go").expect("mints");

        let (tool, calls) = HostTool::counted("host_ask");
        let owner = pinned(dir.path(), runtime)
            .with_tool(tool)
            .open()
            .await
            .expect("opens");

        let report = siblings_run
            .execute_with_approver(CollectingSink::default(), AllowAll)
            .await
            .expect("completes");
        assert!(matches!(report.outcome, RunOutcome::Ok));

        let offered = roster(&endpoint, 0);
        assert!(
            offered.iter().any(|tool| tool == "host_ask"),
            "the roster still carries the name, which is why the refusal has to come from the \
             guard rather than from a hide: {offered:?}"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "the sibling's call must not reach the closure the other open supplied"
        );
        let refused = endpoint.requests()[1].clone();
        assert!(
            refused.contains("another open of this workspace supplied"),
            "and the model is told why, in the guard's own words: {refused}"
        );

        drop(owner);
    }
}

/// A workspace's own hooks still guard its runs when the runtime is shared —
/// the registration is a live, audience-scoped one now, the effect is unchanged
/// — and a sibling workspace without hooks is untouched by them.
#[cfg(unix)]
#[tokio::test]
async fn a_workspaces_hooks_guard_its_runs_on_a_shared_runtime() {
    use std::os::unix::fs::PermissionsExt;

    // Connections alternate per run: a tool call, then the wrap-up text.
    let endpoint = ScriptedEndpoint::start(vec![
        Reply::write_file("made.txt"),
        Reply::Text,
        Reply::write_file("made.txt"),
        Reply::Text,
    ]);
    let runtime = shared_runtime(&endpoint);

    let guarded = workspace_dir();
    let script = guarded.path().join("deny.sh");
    std::fs::write(
        &script,
        "#!/bin/sh\necho '{\"decision\":\"deny\",\"reason\":\"workspace guard\"}'\n",
    )
    .expect("script");
    std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).expect("chmod");
    std::fs::create_dir_all(guarded.path().join(".basis")).expect("dir");
    std::fs::write(
        guarded.path().join(".basis/hooks.json"),
        format!(
            r#"{{"schema": 1, "hooks": [{{"name": "guard", "command": ["{}"]}}]}}"#,
            script.display()
        ),
    )
    .expect("hooks file");
    let free = workspace_dir();

    let first = pinned(guarded.path(), Arc::clone(&runtime))
        .open()
        .await
        .expect("opens");
    let second = pinned(free.path(), runtime).open().await.expect("opens");

    first
        .prepare("write a file")
        .expect("mints")
        .execute_with_approver(CollectingSink::default(), AllowAll)
        .await
        .expect("the guarded run completes — a denial is an answer, not an error");
    assert!(
        !guarded.path().join("made.txt").exists(),
        "the guarded workspace's hook must stop the write"
    );

    second
        .prepare("write a file")
        .expect("mints")
        .execute_with_approver(CollectingSink::default(), AllowAll)
        .await
        .expect("the free run completes");
    assert!(
        second.path().join("made.txt").exists(),
        "a sibling with no hooks must be untouched by the guarded one's"
    );
}

/// One directory is one tool audience, so one directory is one chain.
///
/// A second live open of a root — the shape `basis-host` produces on purpose,
/// one workspace per set of client-supplied MCP servers — joins the
/// registration already there instead of putting a second complete chain behind
/// the same audience. Both halves are asserted, because a rule that registered
/// nothing at all would pass the first: the hook still runs, and it runs
/// *once*, which is what keeps a rewrite that is not idempotent from being fed
/// its own output.
#[cfg(unix)]
#[tokio::test]
async fn a_second_open_of_one_root_joins_the_first_ones_chain_rather_than_doubling_it() {
    use std::os::unix::fs::PermissionsExt;

    let endpoint = ScriptedEndpoint::start(vec![
        Reply::write_file("same-root.txt"),
        Reply::Text,
        Reply::write_file("same-root.txt"),
        Reply::Text,
    ]);
    let runtime = shared_runtime(&endpoint);
    let root = workspace_dir();
    let ledger = root.path().join("consulted");

    let path = root.path().join("count.sh");
    std::fs::write(
        &path,
        format!(
            "#!/bin/sh\necho ran >> '{}'\necho '{{\"decision\":\"allow\"}}'\n",
            ledger.display()
        ),
    )
    .expect("script");
    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).expect("chmod");
    let counting = HookSpec::new("count", vec![path.to_string_lossy().into_owned()]);
    let config = || HooksConfig {
        workspace_file: PathBuf::from(".basis/hooks.json"),
        global_dir: None,
        supplied: vec![counting.clone()],
    };
    let consultations = || {
        std::fs::read_to_string(&ledger)
            .map(|body| body.lines().count())
            .unwrap_or_default()
    };

    let first = pinned(root.path(), Arc::clone(&runtime))
        .with_hooks(config())
        .open()
        .await
        .expect("the chain registers");
    let second = pinned(root.path(), runtime)
        .with_hooks(config())
        .open()
        .await
        .expect("an identical second open of one root joins it");

    second
        .prepare("write a file")
        .expect("mints")
        .execute_with_approver(CollectingSink::default(), AllowAll)
        .await
        .expect("completes");
    assert_eq!(
        consultations(),
        1,
        "one directory carries one chain, however many times it is open"
    );

    // The first holder going does not take the chain with it: the second is
    // still serving.
    drop(first);
    second
        .prepare("write a file")
        .expect("mints")
        .execute_with_approver(CollectingSink::default(), AllowAll)
        .await
        .expect("completes");
    assert_eq!(
        consultations(),
        2,
        "and the surviving open's runs are still guarded"
    );
}

/// The other answer, because there is no third one.
///
/// Two live opens of a root either share one chain — which needs them to
/// configure the same chain — or would have two behind one audience, and mentra
/// would walk both for either one's calls: every hook spawned twice, and the
/// first open's sessions judged by a chain their caller never wrote. Refused by
/// name, as it was before hooks went live.
#[cfg(unix)]
#[tokio::test]
async fn a_second_open_of_one_root_with_a_different_chain_is_refused() {
    use std::os::unix::fs::PermissionsExt;

    let endpoint = ScriptedEndpoint::start(Vec::new());
    let runtime = shared_runtime(&endpoint);
    let root = workspace_dir();

    let script = |name: &str, body: &str| {
        let path = root.path().join(name);
        std::fs::write(&path, format!("#!/bin/sh\n{body}\n")).expect("script");
        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).expect("chmod");
        HookSpec::new(name, vec![path.to_string_lossy().into_owned()])
    };
    let deny = script(
        "deny-same-root.sh",
        r#"echo '{"decision":"deny","reason":"first guard stays"}'"#,
    );
    let allow = script("allow-same-root.sh", r#"echo '{"decision":"allow"}'"#);
    let config = |supplied| HooksConfig {
        workspace_file: PathBuf::from(".basis/hooks.json"),
        global_dir: None,
        supplied,
    };

    let first = pinned(root.path(), Arc::clone(&runtime))
        .with_hooks(config(vec![deny]))
        .open()
        .await
        .expect("the first chain registers");

    let refused = pinned(root.path(), Arc::clone(&runtime))
        .with_hooks(config(vec![allow.clone()]))
        .open()
        .await
        .expect_err("a second open must not be able to install a second chain");
    assert!(
        matches!(refused, basis::RunError::WorkspaceGuardConflict { .. }),
        "the refusal has to be the typed one a host can react to: {refused}"
    );

    // And the refusal is about the conflict, not about the root: once the
    // first open goes, the same configuration opens.
    drop(first);
    pinned(root.path(), runtime)
        .with_hooks(config(vec![allow]))
        .open()
        .await
        .expect("a root nobody holds is free to configure");
}

/// The host's own guards reach a session basis never minted.
///
/// `RuntimeBuilder::with_interceptor` promises runtime scope, and a runtime is
/// larger than any workspace on it: a host that builds a basis `Runtime` and
/// then drives a session of its own through `mentra_runtime` — which is what
/// `run::prepare_with_session` does — has an agent with no tool audience, so
/// every audience-scoped registration skips it. Only a global one reaches it,
/// which is where the interceptors are.
#[cfg(unix)]
#[tokio::test]
async fn a_host_interceptor_judges_a_session_no_workspace_minted() {
    use basis::{HookOutcome, HookRequest, Interceptor, InterceptorError};
    use mentra::{
        ContentBlock, ModelInfo,
        agent::{AgentConfig, WorkspaceConfig},
        provider::ProviderId,
    };
    use std::sync::atomic::AtomicUsize;

    struct Refusing(Arc<AtomicUsize>);

    #[async_trait::async_trait]
    impl Interceptor for Refusing {
        fn name(&self) -> &str {
            "host-guard"
        }

        async fn intercept(&self, _call: &HookRequest) -> Result<HookOutcome, InterceptorError> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(HookOutcome::Deny("the host says no".to_string()))
        }
    }

    let endpoint = ScriptedEndpoint::start(vec![Reply::write_file("unguarded.txt"), Reply::Text]);
    let consulted = Arc::new(AtomicUsize::new(0));
    let runtime = Arc::new(
        Runtime::builder()
            .with_base_url(&endpoint.base_url)
            .with_api_key("test-key")
            .with_ephemeral_history()
            .with_interceptor(Refusing(Arc::clone(&consulted)))
            .build()
            .expect("builds"),
    );

    // A session the host makes for itself: no workspace, no audience, based in
    // a directory a workspace could perfectly well have been opened on.
    let dir = workspace_dir();
    let mut session = runtime
        .mentra_runtime()
        .create_session_with_config(
            "the host's own",
            ModelInfo::new("test-model", ProviderId::new(runtime.provider())),
            AgentConfig {
                workspace: WorkspaceConfig {
                    base_dir: dir.path().to_path_buf(),
                    ..Default::default()
                },
                ..Default::default()
            },
        )
        .expect("the host's session is created");
    let _ = session.append_turn(vec![ContentBlock::text("go")]).await;

    assert_eq!(
        consulted.load(Ordering::SeqCst),
        1,
        "the host's interceptor must be asked about a call it did not mint"
    );
    assert!(
        !dir.path().join("unguarded.txt").exists(),
        "and its refusal must hold: the write never happened"
    );
}

/// And they still speak *first*, from a batch of their own.
///
/// The ordering `RuntimeBuilder::with_interceptor` documents — host
/// interceptors before any subprocess hook, so the host's own guard can stop a
/// program a repository chose from being spawned at all — used to be basis's
/// because one `HookRunner` held both. It is now mentra's: one chain per call
/// composed from the global batch this runtime registered at build and the
/// audience batch the workspace registered at open, in that order. Pinned end
/// to end because nothing inside basis can assert it any more.
#[cfg(unix)]
#[tokio::test]
async fn the_hosts_guard_still_speaks_before_a_workspaces_own() {
    use basis::{HookOutcome, HookRequest, Interceptor, InterceptorError};
    use std::os::unix::fs::PermissionsExt;

    struct Refusing;

    #[async_trait::async_trait]
    impl Interceptor for Refusing {
        fn name(&self) -> &str {
            "host-guard"
        }

        async fn intercept(&self, _call: &HookRequest) -> Result<HookOutcome, InterceptorError> {
            Ok(HookOutcome::Deny("my program, my rules".to_string()))
        }
    }

    let endpoint = ScriptedEndpoint::start(vec![Reply::write_file("made.txt"), Reply::Text]);
    let runtime = Arc::new(
        Runtime::builder()
            .with_base_url(&endpoint.base_url)
            .with_api_key("test-key")
            .with_ephemeral_history()
            .with_interceptor(Refusing)
            .build()
            .expect("builds"),
    );

    let dir = workspace_dir();
    let marker = dir.path().join("hook-ran");
    let script = dir.path().join("allow.sh");
    std::fs::write(
        &script,
        format!(
            "#!/bin/sh\ntouch '{}'\necho '{{\"decision\":\"allow\"}}'\n",
            marker.display()
        ),
    )
    .expect("script");
    std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).expect("chmod");
    std::fs::create_dir_all(dir.path().join(".basis")).expect("dir");
    std::fs::write(
        dir.path().join(".basis/hooks.json"),
        format!(
            r#"{{"schema": 1, "hooks": [{{"name": "repo", "command": ["{}"]}}]}}"#,
            script.display()
        ),
    )
    .expect("hooks file");

    let workspace = pinned(dir.path(), runtime).open().await.expect("opens");
    workspace
        .prepare("write a file")
        .expect("mints")
        .execute_with_approver(CollectingSink::default(), AllowAll)
        .await
        .expect("a denial is an answer, not a run failure");

    assert!(
        !marker.exists(),
        "the host's refusal must land before a program the repository chose is spawned"
    );
    assert!(
        !dir.path().join("made.txt").exists(),
        "and the call it refused must not have run"
    );
}

// ---------------------------------------------------------------------------
// The endpoint
// ---------------------------------------------------------------------------

/// What one connection answers with.
#[derive(Clone)]
enum Reply {
    /// A finished assistant message, numbered by connection.
    Text,
    /// A single tool call; the next connection is expected to wrap up.
    #[cfg(any(unix, feature = "mcp"))]
    ToolCall { name: String, arguments: String },
}

impl Reply {
    /// The write a basis runtime's roster actually offers: `write`, not the
    /// batched `files` this used to script. A runtime built through
    /// [`basis::RuntimeBuilder`] registers mentra's split file tools by
    /// default, so a scripted `files` call would name a tool that is not
    /// there and the run would prove nothing about hooks.
    #[cfg(unix)]
    fn write_file(path: &str) -> Self {
        Self::ToolCall {
            name: "write".to_string(),
            arguments: json!({"path": path, "content": "hi"}).to_string(),
        }
    }
}

/// An OpenAI-compatible endpoint on loopback speaking `chat/completions` — the
/// wire a base URL gets — that follows a per-connection script (falling back to
/// a numbered text reply) and keeps every request it was sent.
/// `tests/workspace.rs` explains why loopback is not "the network".
struct ScriptedEndpoint {
    base_url: String,
    #[cfg_attr(
        not(feature = "mcp"),
        allow(
            dead_code,
            reason = "read back only by the roster test, which is mcp-gated"
        )
    )]
    requests: Arc<Mutex<Vec<String>>>,
}

impl ScriptedEndpoint {
    fn start(script: Vec<Reply>) -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").expect("bind test endpoint");
        let address = listener.local_addr().expect("read endpoint address");
        let requests = Arc::new(Mutex::new(Vec::new()));

        let recorded = Arc::clone(&requests);
        let script = Arc::new(script);
        let turns = Arc::new(AtomicUsize::new(0));
        thread::spawn(move || {
            while let Ok((stream, _)) = listener.accept() {
                let script = Arc::clone(&script);
                let turns = Arc::clone(&turns);
                let recorded = Arc::clone(&recorded);
                // One thread per connection, so concurrent runs are answered
                // concurrently rather than queued.
                thread::spawn(move || answer(stream, &script, &turns, &recorded));
            }
        });

        Self {
            base_url: format!("http://{address}/"),
            requests,
        }
    }

    fn requests(&self) -> Vec<String> {
        self.requests.lock().expect("requests").clone()
    }
}

/// A pinned model is looked up in the provider's listing before the first
/// turn (mentra `bfe952b`), which is one `GET …/models` per run that is
/// neither a turn nor scripted. Answered with a listing that names the test
/// model, so the lookup succeeds the way a real provider's would, and never
/// counted or recorded as a turn.
fn model_listing(request: &str) -> Option<String> {
    let line = request.lines().next()?;
    let target = line.split_whitespace().nth(1)?;
    (line.starts_with("GET ") && target.ends_with("/models")).then(|| {
        let body = r#"{"object":"list","data":[{"id":"test-model","object":"model"}]}"#;
        format!(
            "HTTP/1.1 200 OK\r\nconnection: close\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\r\n{body}",
            body.len()
        )
    })
}

fn answer(
    mut stream: TcpStream,
    script: &[Reply],
    turns: &AtomicUsize,
    recorded: &Mutex<Vec<String>>,
) {
    let request = read_http_request(&mut stream);
    if let Some(listing) = model_listing(&request) {
        let _ = stream.write_all(listing.as_bytes());
        return;
    }
    let index = turns.fetch_add(1, Ordering::SeqCst) + 1;
    let reply = &script.get(index - 1).cloned().unwrap_or(Reply::Text);
    recorded.lock().expect("requests").push(request);

    let body = sse_body(index, reply);
    let response = format!(
        "HTTP/1.1 200 OK\r\nconnection: close\r\ncontent-type: text/event-stream\r\ncontent-length: {}\r\n\r\n{body}",
        body.len()
    );
    let _ = stream.write_all(response.as_bytes());
}

/// The smallest `chat/completions` stream that is a finished turn of the
/// requested shape. This wire streams flat deltas and ends at `[DONE]`; there
/// is no item to open or close, so a whole reply is one chunk.
fn sse_body(index: usize, reply: &Reply) -> String {
    let id = format!("chatcmpl_{index}");
    let mut events = Vec::new();

    match reply {
        Reply::Text => {
            events.push(json!({
                "id": id, "model": "test-model",
                "choices": [{"index": 0, "delta": {"role": "assistant", "content": format!("reply-{index}")}}]
            }));
            events.push(json!({
                "id": id,
                "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]
            }));
        }
        #[cfg(any(unix, feature = "mcp"))]
        Reply::ToolCall { name, arguments } => {
            events.push(json!({
                "id": id, "model": "test-model",
                "choices": [{"index": 0, "delta": {"role": "assistant", "tool_calls": [{
                    "index": 0, "id": format!("call_{index}"), "type": "function",
                    // Arguments are a JSON *string* on this wire, and arrive in
                    // slices; one slice is enough to be a call.
                    "function": {"name": name, "arguments": arguments}
                }]}}]
            }));
            events.push(json!({
                "id": id,
                "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}]
            }));
        }
    }

    events
        .iter()
        .map(|event| format!("data: {event}\n\n"))
        .chain(std::iter::once("data: [DONE]\n\n".to_string()))
        .collect()
}

/// Reads a request up to the end of its declared body.
///
/// Reading to end-of-stream would deadlock: the client keeps the connection
/// open waiting for the response it has not been sent yet.
fn read_http_request(stream: &mut TcpStream) -> String {
    let mut bytes = Vec::new();
    let mut buffer = [0_u8; 4096];
    let mut header_end = None;
    let mut content_length = 0_usize;

    while let Ok(read) = stream.read(&mut buffer) {
        if read == 0 {
            break;
        }
        bytes.extend_from_slice(&buffer[..read]);
        if header_end.is_none()
            && let Some(index) = bytes.windows(4).position(|window| window == b"\r\n\r\n")
        {
            let end = index + 4;
            header_end = Some(end);
            content_length = String::from_utf8_lossy(&bytes[..end])
                .lines()
                .find_map(|line| {
                    let (name, value) = line.split_once(':')?;
                    name.eq_ignore_ascii_case("content-length")
                        .then(|| value.trim().parse::<usize>().unwrap_or_default())
                })
                .unwrap_or_default();
        }
        if header_end.is_some_and(|end| bytes.len() >= end + content_length) {
            break;
        }
    }

    String::from_utf8_lossy(&bytes).into_owned()
}

/// A host that builds software needs more than two minutes.
///
/// The default suits the commands a harness usually runs and does not suit
/// `docker compose build`. Past the limit the process is killed mid-stream and
/// the caller gets truncated output with no error in it, which reads as a
/// silent failure rather than a stopped one.
#[test]
fn a_host_can_ask_for_more_command_patience() {
    let runtime = Runtime::builder()
        .with_base_url("http://127.0.0.1:1/v1")
        .with_api_key("test-key")
        .with_ephemeral_history()
        .with_command_timeout(std::time::Duration::from_secs(600))
        .build();
    assert!(
        runtime.is_ok(),
        "a longer timeout is not a reason to fail: {:?}",
        runtime.err()
    );
}

#[test]
fn asking_for_nothing_keeps_the_default() {
    let runtime = Runtime::builder()
        .with_base_url("http://127.0.0.1:1/v1")
        .with_api_key("test-key")
        .with_ephemeral_history()
        .build();
    assert!(runtime.is_ok(), "{:?}", runtime.err());
}