use std::{
io::{Read, Write},
net::{TcpListener, TcpStream},
path::{Path, PathBuf},
sync::{
Arc, Mutex,
atomic::{AtomicUsize, Ordering},
},
thread,
};
use basis::{
AllowAll, CollectingSink, ContextConfig, MemoryConfig, RunOutcome, Runtime, Workspace,
WorkspaceBuilder, hooks::HooksConfig, skills::SkillsConfig, store, templates::TemplatesConfig,
tools::declared::ToolsConfig,
};
#[cfg(unix)]
use basis::HookSpec;
use mentra::ModelSelector;
use serde_json::json;
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(),
})
.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
}
fn resolved(path: &Path) -> PathBuf {
let canonical = path.canonicalize().expect("canonical path");
dunce::simplified(&canonical).to_path_buf()
}
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");
body["tools"]
.as_array()
.expect("a tools array")
.iter()
.filter_map(|tool| tool["function"]["name"].as_str().map(str::to_string))
.collect()
}
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");
assert!(
std::ptr::eq(a.mentra_runtime(), b.mentra_runtime()),
"both workspaces must run on the very same mentra runtime"
);
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"
);
}
#[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() {
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();
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(&resolved(dir_a.path()).to_string_lossy().to_string())
&& message.contains(&resolved(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};
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);
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));
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:?}"
);
}
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";
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(),
}
}
#[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() {
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");
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"
);
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);
}
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()
}
#[tokio::test]
async fn a_departing_sibling_does_not_release_the_agent_a_live_open_took_back() {
let endpoint = ScriptedEndpoint::start(vec![
Reply::Text,
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");
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");
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);
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);
let mut live = a
.resume(&agent_id, "again")
.expect("its own workspace resumes it");
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);
}
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() {
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:?}"
);
}
}
mod declared_roster {
use super::*;
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() {
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() {
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");
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:?}"
);
}
}
mod host_roster {
use super::*;
use basis::tools::{
ParallelToolContext, RuntimeToolDescriptor, ToolDefinition, ToolExecutor, ToolResult,
};
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() {
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");
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));
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() {
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() {
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() {
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() {
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() {
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);
}
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() {
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() {
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() {
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);
}
}
#[cfg(unix)]
#[tokio::test]
async fn a_workspaces_hooks_guard_its_runs_on_a_shared_runtime() {
use std::os::unix::fs::PermissionsExt;
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"
);
}
#[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"
);
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"
);
}
#[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}"
);
drop(first);
pinned(root.path(), runtime)
.with_hooks(config(vec![allow]))
.open()
.await
.expect("a root nobody holds is free to configure");
}
#[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"),
);
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"
);
}
#[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"
);
}
#[derive(Clone)]
enum Reply {
Text,
#[cfg(any(unix, feature = "mcp"))]
ToolCall { name: String, arguments: String },
}
impl Reply {
#[cfg(unix)]
fn write_file(path: &str) -> Self {
Self::ToolCall {
name: "write".to_string(),
arguments: json!({"path": path, "content": "hi"}).to_string(),
}
}
}
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);
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()
}
}
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());
}
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",
"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()
}
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()
}
#[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());
}