use std::{
collections::{BTreeMap, HashMap},
hash::{DefaultHasher, Hash, Hasher},
path::{Path, PathBuf},
sync::{Arc, Mutex},
};
use basis::{
McpConfig, McpServer, McpSseServerConfig, McpStreamableHttpServerConfig, PersistedSession,
PreparedRun, RunError, RunSpec, Runtime, RuntimeBuilder, Workspace, WorkspaceBuilder,
mcp::{McpSseLimits, McpStreamableHttpLimits, SecretString},
};
use tokio::sync::OnceCell;
use crate::{SessionSource, SessionTemplate};
pub struct ConfiguredSource {
template: SessionTemplate,
runtime: OnceCell<Arc<Runtime>>,
workspaces: Mutex<HashMap<WorkspaceKey, Arc<OnceCell<Arc<Workspace>>>>>,
}
impl ConfiguredSource {
pub fn new(template: Option<SessionTemplate>) -> Self {
Self {
template: template.unwrap_or_default(),
runtime: OnceCell::new(),
workspaces: Mutex::new(HashMap::new()),
}
}
#[cfg(test)]
pub fn on_runtime(runtime: Arc<Runtime>, template: Option<SessionTemplate>) -> Self {
let source = Self::new(template);
source
.runtime
.set(runtime)
.unwrap_or_else(|_| unreachable!("a runtime that was just constructed is empty"));
source
}
fn parts_for(&self, cwd: PathBuf, mcp: Vec<McpServer>) -> (WorkspaceBuilder, RunSpec) {
let template = &self.template;
let mut builder = Workspace::builder(cwd).with_shell(template.shell);
if let Some(model) = &template.model {
builder = builder.with_model(model.clone());
}
if let Some(system_prompt) = &template.system_prompt {
builder = builder.with_system_prompt(system_prompt.clone());
}
let discovery = &template.discovery;
let builder = builder
.with_context(discovery.context.clone())
.with_skills(discovery.skills.clone())
.with_templates(discovery.templates.clone())
.with_hooks(discovery.hooks.clone())
.with_tools(discovery.tools.clone());
let builder = builder.with_mcp(self.session_mcp(mcp));
let mut spec = RunSpec::default();
if let Some(session_name) = &template.session_name {
spec = spec.with_session_name(session_name.clone());
}
if let Some(effort) = template.effort {
spec = spec.with_effort(effort);
}
(builder, spec)
}
fn session_mcp(&self, mcp: Vec<McpServer>) -> McpConfig {
self.template.discovery.mcp.clone().with_supplied(mcp)
}
async fn workspace_for(
&self,
cwd: PathBuf,
mcp: Vec<McpServer>,
) -> Result<(Arc<Workspace>, RunSpec), RunError> {
let key = WorkspaceKey::new(&cwd, &mcp);
let (builder, spec) = self.parts_for(cwd, mcp);
Ok((self.open(key, builder).await?, spec))
}
async fn open(
&self,
key: WorkspaceKey,
builder: WorkspaceBuilder,
) -> Result<Arc<Workspace>, RunError> {
let cell = Arc::clone(self.lock().entry(key).or_default());
let workspace = cell
.get_or_try_init(|| async move {
let runtime = Arc::clone(self.runtime().await?);
Ok::<_, RunError>(Arc::new(builder.with_runtime(runtime).open().await?))
})
.await?;
Ok(Arc::clone(workspace))
}
async fn runtime(&self) -> Result<&Arc<Runtime>, RunError> {
self.runtime
.get_or_try_init(|| async { Ok(Arc::new(self.recipe().build()?)) })
.await
}
fn recipe(&self) -> RuntimeBuilder {
let template = &self.template;
let mut recipe = Runtime::builder();
if let Some(model) = &template.model {
recipe = recipe.with_model(model.clone());
}
if let Some(provider) = template.provider {
recipe = recipe.with_provider(provider);
}
if let Some(base_url) = &template.base_url {
recipe = recipe.with_base_url(base_url.clone());
}
recipe
}
fn lock(
&self,
) -> std::sync::MutexGuard<'_, HashMap<WorkspaceKey, Arc<OnceCell<Arc<Workspace>>>>> {
self.workspaces
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
#[cfg(test)]
pub fn opened(&self) -> Vec<Arc<Workspace>> {
self.lock()
.values()
.filter_map(|cell| cell.get().map(Arc::clone))
.collect()
}
}
#[async_trait::async_trait]
impl SessionSource for ConfiguredSource {
async fn create(&self, cwd: PathBuf, mcp: Vec<McpServer>) -> Result<PreparedRun, RunError> {
let (workspace, spec) = self.workspace_for(cwd, mcp).await?;
workspace.prepare(spec)
}
async fn resume(
&self,
agent_id: &str,
cwd: PathBuf,
mcp: Vec<McpServer>,
) -> Result<PreparedRun, RunError> {
let (workspace, spec) = self.workspace_for(cwd, mcp).await?;
workspace.resume(agent_id, spec)
}
fn lists_sessions(&self) -> bool {
true
}
async fn list_sessions(&self, cwd: PathBuf) -> Result<Vec<PersistedSession>, RunError> {
basis::store::list(&cwd)
}
fn deletes_sessions(&self) -> bool {
true
}
async fn delete(&self, agent_id: &str) -> Result<(), RunError> {
basis::store::forget(agent_id)
}
}
#[derive(Debug, PartialEq, Eq, Hash)]
struct WorkspaceKey {
workspace: PathBuf,
supplied: u64,
}
impl WorkspaceKey {
fn new(cwd: &Path, supplied: &[McpServer]) -> Self {
Self {
workspace: std::fs::canonicalize(cwd).unwrap_or_else(|_| cwd.to_path_buf()),
supplied: digest(supplied),
}
}
}
fn digest(servers: &[McpServer]) -> u64 {
let mut hasher = DefaultHasher::new();
for server in servers {
match server {
McpServer::Stdio(config) => {
"stdio".hash(&mut hasher);
config.name.hash(&mut hasher);
config.command.hash(&mut hasher);
config.args.hash(&mut hasher);
config.cwd.hash(&mut hasher);
let mut env: Vec<(&String, &String)> = config.env.iter().collect();
env.sort_unstable();
env.hash(&mut hasher);
}
McpServer::Sse(config) => {
let McpSseServerConfig {
name,
url,
headers,
allow_plaintext_credentials,
limits,
} = config;
hash_remote(
&mut hasher,
"sse",
name,
url,
headers,
*allow_plaintext_credentials,
);
let McpSseLimits {
connect_timeout,
initialize_timeout,
list_tools_timeout,
call_tool_timeout,
stream_idle_timeout,
max_event_bytes,
max_endpoint_bytes,
max_tool_pages,
max_tools,
} = limits;
(
connect_timeout,
initialize_timeout,
list_tools_timeout,
call_tool_timeout,
stream_idle_timeout,
)
.hash(&mut hasher);
(
max_event_bytes,
max_endpoint_bytes,
max_tool_pages,
max_tools,
)
.hash(&mut hasher);
}
McpServer::Http(config) => {
let McpStreamableHttpServerConfig {
name,
url,
headers,
allow_plaintext_credentials,
limits,
} = config;
hash_remote(
&mut hasher,
"http",
name,
url,
headers,
*allow_plaintext_credentials,
);
let McpStreamableHttpLimits {
connect_timeout,
initialize_timeout,
list_tools_timeout,
call_tool_timeout,
stream_idle_timeout,
max_event_bytes,
max_response_bytes,
max_tool_pages,
max_tools,
} = limits;
(
connect_timeout,
initialize_timeout,
list_tools_timeout,
call_tool_timeout,
stream_idle_timeout,
)
.hash(&mut hasher);
(
max_event_bytes,
max_response_bytes,
max_tool_pages,
max_tools,
)
.hash(&mut hasher);
}
server => {
std::mem::discriminant(server).hash(&mut hasher);
"unknown-transport".hash(&mut hasher);
server.name().hash(&mut hasher);
}
}
}
hasher.finish()
}
fn hash_remote(
hasher: &mut DefaultHasher,
transport: &str,
name: &str,
url: &str,
headers: &BTreeMap<String, SecretString>,
allow_plaintext_credentials: bool,
) {
transport.hash(hasher);
name.hash(hasher);
url.hash(hasher);
for (header, value) in headers {
header.hash(hasher);
value.expose_secret().hash(hasher);
}
allow_plaintext_credentials.hash(hasher);
}
#[cfg(test)]
mod tests {
use std::{
path::{Path, PathBuf},
sync::Arc,
};
use basis::{
ContextConfig, McpConfig, McpServer, Runtime, ToolsConfig, hooks::HooksConfig,
skills::SkillsConfig, templates::TemplatesConfig,
};
use mentra::{
McpServerConfig, McpSseServerConfig, McpStreamableHttpServerConfig, ModelSelector,
};
use super::{ConfiguredSource, WorkspaceKey, digest};
use crate::{Discovery, SessionSource, SessionTemplate};
#[test]
fn two_servers_differing_only_in_transport_hash_apart() {
let sse = McpServer::Sse(McpSseServerConfig::new("api", "https://example.com/mcp"));
let http = McpServer::Http(McpStreamableHttpServerConfig::new(
"api",
"https://example.com/mcp",
));
assert_ne!(digest(&[sse]), digest(&[http]));
}
#[test]
fn the_plaintext_override_reaches_the_digest() {
let base = || McpStreamableHttpServerConfig::new("api", "http://127.0.0.1/mcp");
assert_ne!(
digest(&[McpServer::Http(base())]),
digest(&[McpServer::Http(base().allowing_plaintext_credentials())]),
);
}
#[test]
fn the_clients_mcp_servers_reach_the_config() {
let source = ConfiguredSource::new(None);
let built = source.session_mcp(vec![McpServer::Stdio(McpServerConfig {
name: "fs".to_string(),
command: "/bin/mcp-fs".to_string(),
args: Vec::new(),
env: Default::default(),
cwd: None,
})]);
let supplied: Vec<&str> = built.supplied.iter().map(McpServer::name).collect();
assert_eq!(
supplied,
vec!["fs"],
"the session request is where a client says which servers it wants"
);
}
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 runtime builds without touching the network"),
)
}
fn offline_template() -> SessionTemplate {
SessionTemplate::new().with_model(ModelSelector::Id("test-model".to_string()))
}
fn offline_discovery() -> Discovery {
Discovery {
context: ContextConfig {
file_name: "AGENTS.md".to_string(),
global_dir: None,
walk_parents: false,
},
skills: SkillsConfig {
workspace_subdir: Some(PathBuf::from(".basis/skills")),
shared_workspace_dir: true,
global_dir: None,
shared_home_dir: false,
},
templates: TemplatesConfig {
workspace_subdir: PathBuf::from(".basis/templates"),
global_dir: None,
},
hooks: HooksConfig {
workspace_file: PathBuf::from(".basis/hooks.json"),
global_dir: None,
supplied: Vec::new(),
},
tools: ToolsConfig {
workspace_file: PathBuf::from(".basis/tools.json"),
global_dir: None,
supplied: Vec::new(),
},
mcp: McpConfig {
workspace_file: PathBuf::from(".mcp.json"),
global_dir: None,
supplied: Vec::new(),
},
}
}
#[tokio::test]
async fn two_sessions_on_one_workspace_share_one_runtime() {
let repository = tempfile::tempdir().expect("tempdir");
let runtime = offline_runtime();
let source = ConfiguredSource::on_runtime(
Arc::clone(&runtime),
Some(offline_template().with_discovery(offline_discovery())),
);
let first = source
.create(repository.path().to_path_buf(), Vec::new())
.await
.expect("the first session opens");
let second = source
.create(repository.path().to_path_buf(), Vec::new())
.await
.expect("the second session opens");
assert_ne!(
first.agent_id(),
second.agent_id(),
"two sessions, not one conversation handed out twice"
);
let opened = source.opened();
assert_eq!(
opened.len(),
1,
"one directory is one workspace, however many sessions were minted on it"
);
assert!(
std::ptr::eq(opened[0].mentra_runtime(), runtime.mentra_runtime()),
"and both were minted on the one runtime this process built"
);
}
#[tokio::test]
async fn a_session_this_source_opens_is_listed_for_its_own_directory() {
let store = tempfile::tempdir().expect("tempdir");
let repository = tempfile::tempdir().expect("tempdir");
let elsewhere = tempfile::tempdir().expect("tempdir");
let runtime = Arc::new(
Runtime::builder()
.with_base_url("http://127.0.0.1:1/v1")
.with_api_key("test-key")
.with_store_dir(store.path())
.build()
.expect("a runtime builds without touching the network"),
);
let source = ConfiguredSource::on_runtime(
runtime,
Some(offline_template().with_discovery(offline_discovery())),
);
let opened = source
.create(repository.path().to_path_buf(), Vec::new())
.await
.expect("the session opens");
let listed: Vec<String> = basis::store::list_in(store.path(), repository.path())
.expect("lists")
.into_iter()
.map(|session| session.agent_id)
.collect();
assert_eq!(
listed,
vec![opened.agent_id().to_string()],
"a client asking about this directory must be handed the session it opened here"
);
assert!(
basis::store::list_in(store.path(), elsewhere.path())
.expect("lists")
.is_empty(),
"and no other directory may claim it"
);
}
#[test]
fn a_session_that_asked_for_different_servers_is_a_different_workspace() {
let server = |command: &str| {
vec![McpServer::Stdio(McpServerConfig {
name: "fs".to_string(),
command: command.to_string(),
args: Vec::new(),
env: Default::default(),
cwd: None,
})]
};
let key = |mcp: Vec<McpServer>| WorkspaceKey::new(Path::new("/repo"), &mcp);
assert_eq!(
key(Vec::new()),
key(Vec::new()),
"the same directory asked about the same way is one workspace"
);
assert_ne!(
key(Vec::new()),
key(server("/bin/mcp-fs")),
"a supplied server is not none"
);
assert_ne!(
key(server("/bin/mcp-fs")),
key(server("/bin/other-fs")),
"one name, two commands: a client that named a different program must get it"
);
assert_ne!(
key(Vec::new()),
WorkspaceKey::new(Path::new("/other-repo"), &[]),
"and a directory is a key"
);
}
}