#![allow(clippy::expect_used, clippy::unwrap_used)]
#[cfg(all(feature = "session-store", feature = "memory-store"))]
mod tests {
use std::pin::Pin;
use std::sync::Arc;
use std::sync::Mutex;
use meerkat::{
AgentFactory, Config, CreateSessionRequest, FactoryAgentBuilder, PersistenceBundle,
PersistentSessionService, SessionQuery, SessionService, ToolCall, build_persistent_service,
build_persistent_service_with_runtime_adapter,
};
use meerkat_client::{LlmClient, LlmDoneOutcome, LlmError, LlmEvent, LlmRequest};
use meerkat_core::service::InitialTurnPolicy;
use meerkat_core::{
AgentToolDispatcher, AuthBindingRef, BindingId, DeferredPromptPolicy, RealmId,
ToolCallView, ToolCatalogCapabilities, ToolCatalogEntry, ToolDef, ToolProvenance,
ToolResult, ToolSourceKind, service::SessionBuildOptions,
};
use meerkat_store::MemoryBlobStore;
use serde_json::json;
fn test_auth_binding() -> AuthBindingRef {
AuthBindingRef {
realm: RealmId::parse("default").expect("default realm id"),
binding: BindingId::parse("default_openai").expect("default openai binding"),
profile: None,
}
}
fn create_request(prompt: &str) -> CreateSessionRequest {
CreateSessionRequest {
model: "gpt-5.4".to_string(),
prompt: prompt.to_string().into(),
render_metadata: None,
system_prompt: None,
max_tokens: None,
event_tx: None,
skill_references: None,
initial_turn: InitialTurnPolicy::Defer,
deferred_prompt_policy: DeferredPromptPolicy::Discard,
build: Some(SessionBuildOptions {
auth_binding: Some(test_auth_binding()),
..Default::default()
}),
labels: None,
}
}
#[derive(Default)]
struct CatalogLoadClient {
calls: Mutex<usize>,
seen_tools: Mutex<Vec<Vec<String>>>,
}
impl CatalogLoadClient {
fn seen_tools(&self) -> Vec<Vec<String>> {
self.seen_tools.lock().expect("seen tools lock").clone()
}
}
#[async_trait::async_trait]
impl LlmClient for CatalogLoadClient {
fn project_replay_messages(
&self,
messages: &[meerkat_core::Message],
) -> Result<Vec<meerkat_core::Message>, meerkat_client::LlmError> {
Ok(messages.to_vec())
}
fn stream<'a>(
&'a self,
request: &'a LlmRequest,
) -> Pin<Box<dyn futures::Stream<Item = Result<LlmEvent, LlmError>> + Send + 'a>> {
self.seen_tools.lock().expect("seen tools lock").push(
request
.tools
.iter()
.map(|tool| tool.name.to_string())
.collect(),
);
let call_index = {
let mut calls = self.calls.lock().expect("calls lock");
let call_index = *calls;
*calls += 1;
call_index
};
let events = if call_index == 0 {
vec![
Ok(LlmEvent::ToolCallComplete {
id: "load-deferred".to_string(),
name: "tool_catalog_load".to_string(),
args: json!({ "names": ["deferred_mcp_tool"] }),
meta: None,
}),
Ok(LlmEvent::Done {
outcome: LlmDoneOutcome::Success {
stop_reason: meerkat_core::StopReason::ToolUse,
},
}),
]
} else {
vec![
Ok(LlmEvent::TextDelta {
delta: "done".to_string(),
meta: None,
}),
Ok(LlmEvent::Done {
outcome: LlmDoneOutcome::Success {
stop_reason: meerkat_core::StopReason::EndTurn,
},
}),
]
};
Box::pin(futures::stream::iter(events))
}
fn provider(&self) -> &'static str {
"mock"
}
async fn health_check(&self) -> Result<(), LlmError> {
Ok(())
}
}
struct DeferredCatalogDispatcher {
tools: Arc<[Arc<ToolDef>]>,
catalog: Arc<[ToolCatalogEntry]>,
dispatched_names: Mutex<Vec<String>>,
}
impl DeferredCatalogDispatcher {
fn new() -> Self {
let deferred = deferred_tool("deferred_mcp_tool");
let deferred_two = deferred_tool("deferred_mcp_tool_two");
Self {
tools: vec![Arc::clone(&deferred), Arc::clone(&deferred_two)].into(),
catalog: vec![
ToolCatalogEntry::session_deferred(
deferred,
true,
"callback:deferred-catalog".to_string(),
),
ToolCatalogEntry::session_deferred(
deferred_two,
true,
"callback:deferred-catalog".to_string(),
),
]
.into(),
dispatched_names: Mutex::new(Vec::new()),
}
}
fn dispatched_names(&self) -> Vec<String> {
self.dispatched_names
.lock()
.expect("dispatched names lock")
.clone()
}
}
#[async_trait::async_trait]
impl AgentToolDispatcher for DeferredCatalogDispatcher {
fn tools(&self) -> Arc<[Arc<ToolDef>]> {
Arc::clone(&self.tools)
}
fn tool_catalog_capabilities(&self) -> ToolCatalogCapabilities {
ToolCatalogCapabilities {
exact_catalog: true,
may_require_catalog_control_plane: false,
}
}
fn tool_catalog(&self) -> Arc<[ToolCatalogEntry]> {
Arc::clone(&self.catalog)
}
async fn dispatch(
&self,
call: ToolCallView<'_>,
) -> Result<meerkat_core::ops::ToolDispatchOutcome, meerkat_core::ToolError> {
self.dispatched_names
.lock()
.expect("dispatched names lock")
.push(call.name.to_string());
Ok(ToolResult::new(call.id.to_string(), format!("ran {}", call.name), false).into())
}
}
fn deferred_tool(name: &str) -> Arc<ToolDef> {
Arc::new(ToolDef {
name: name.into(),
description: format!("Deferred test tool {name}"),
input_schema: json!({ "type": "object" }),
provenance: Some(deferred_catalog_provenance()),
})
}
fn deferred_catalog_provenance() -> ToolProvenance {
ToolProvenance {
kind: ToolSourceKind::Callback,
source_id: "deferred-catalog".into(),
}
}
fn catalog_route_request(
prompt: &str,
initial_turn: InitialTurnPolicy,
) -> CreateSessionRequest {
CreateSessionRequest {
model: "mock-model".to_string(),
prompt: prompt.to_string().into(),
render_metadata: None,
system_prompt: None,
max_tokens: None,
event_tx: None,
skill_references: None,
initial_turn,
deferred_prompt_policy: DeferredPromptPolicy::Discard,
build: None,
labels: None,
}
}
fn catalog_route_resume_request(session: meerkat::Session) -> CreateSessionRequest {
let mut request = catalog_route_request("", InitialTurnPolicy::Defer);
request.build = Some(SessionBuildOptions {
resume_session: Some(session),
..Default::default()
});
request
}
fn catalog_route_service(
store_path: impl Into<std::path::PathBuf>,
store: Arc<dyn meerkat::SessionStore>,
dispatcher: Arc<DeferredCatalogDispatcher>,
client: Arc<CatalogLoadClient>,
) -> PersistentSessionService<FactoryAgentBuilder> {
let factory = AgentFactory::new(store_path)
.builtins(false)
.shell(false)
.memory(false)
.schedule(false);
let mut builder = FactoryAgentBuilder::new(factory, Config::default());
let llm_client: Arc<dyn LlmClient> = client;
let tool_dispatcher: Arc<dyn AgentToolDispatcher> = dispatcher;
builder.default_llm_client = Some(llm_client);
builder.default_tool_dispatcher = Some(tool_dispatcher);
PersistentSessionService::new(builder, 4, store, None, Arc::new(MemoryBlobStore::new()))
}
#[tokio::test]
async fn persistence_contract_build_persistent_service_accepts_memory_bundle() {
let temp = tempfile::tempdir().expect("tempdir should succeed");
let session_store: Arc<dyn meerkat::SessionStore> = Arc::new(meerkat::MemoryStore::new());
let persistence = PersistenceBundle::new(
Arc::clone(&session_store),
None,
Arc::new(MemoryBlobStore::new()),
);
let factory = AgentFactory::new(temp.path().join("sessions")).builtins(false);
let mut config = Config::default();
let section = meerkat_core::RealmConfigSection::from_inline_api_keys(&[(
"openai",
"test-openai-key",
)]);
config.realm.insert("default".to_string(), section);
let service = build_persistent_service(factory, config, 4, persistence);
let created = service
.create_session(create_request("memory persistence contract"))
.await
.expect("create_session should succeed through the facade persistence bundle");
let persisted = session_store
.load(&created.session_id)
.await
.expect("load should succeed")
.expect("memory-backed bundle should persist the created session");
assert_eq!(persisted.id(), &created.session_id);
let summaries = service
.list(SessionQuery::default())
.await
.expect("list should succeed");
assert!(
summaries
.iter()
.any(|summary| summary.session_id == created.session_id),
"the facade-level persistent service should expose the memory-backed persisted session"
);
}
#[tokio::test]
async fn persistence_contract_build_persistent_runtime_adapter_shares_bundle_runtime() {
let temp = tempfile::tempdir().expect("tempdir should succeed");
let session_store: Arc<dyn meerkat::SessionStore> = Arc::new(meerkat::MemoryStore::new());
let persistence = PersistenceBundle::new(
Arc::clone(&session_store),
None,
Arc::new(MemoryBlobStore::new()),
);
let mut config = Config::default();
let section = meerkat_core::RealmConfigSection::from_inline_api_keys(&[(
"openai",
"test-openai-key",
)]);
config.realm.insert("default".to_string(), section);
let factory_a = AgentFactory::new(temp.path().join("sessions-a")).builtins(false);
let factory_b = AgentFactory::new(temp.path().join("sessions-b")).builtins(false);
let (service_a, adapter_a) = build_persistent_service_with_runtime_adapter(
factory_a,
config.clone(),
4,
persistence.clone(),
);
let (service_b, adapter_b) =
build_persistent_service_with_runtime_adapter(factory_b, config, 4, persistence);
assert!(
Arc::ptr_eq(&adapter_a, &adapter_b),
"reusing one PersistenceBundle should reuse the same runtime adapter"
);
let created = service_a
.create_session(create_request("shared runtime bundle"))
.await
.expect("create_session should succeed through the shared runtime-backed bundle");
let loaded = service_b
.load_authoritative_session(&created.session_id)
.await
.expect("authoritative load should succeed through the second service")
.expect("services built from the same PersistenceBundle should share persisted state");
assert_eq!(loaded.id(), &created.session_id);
}
#[tokio::test]
async fn deferred_catalog_load_authority_survives_persistent_restart() {
let temp = tempfile::tempdir().expect("tempdir should succeed");
let session_store: Arc<dyn meerkat::SessionStore> = Arc::new(meerkat::MemoryStore::new());
let dispatcher = Arc::new(DeferredCatalogDispatcher::new());
let client = Arc::new(CatalogLoadClient::default());
let service = catalog_route_service(
temp.path().join("sessions"),
Arc::clone(&session_store),
Arc::clone(&dispatcher),
Arc::clone(&client),
);
let created = service
.create_session(catalog_route_request(
"load the deferred catalog tool",
InitialTurnPolicy::RunImmediately,
))
.await
.expect("real catalog-load route should complete");
let id = created.session_id;
let seen_tools = client.seen_tools();
assert!(
seen_tools.len() >= 2,
"catalog-load route should continue into a post-boundary LLM request: {seen_tools:?}"
);
assert!(
seen_tools[0].iter().any(|name| name == "tool_catalog_load"),
"control load tool must be visible before deferred tools are loaded: {seen_tools:?}"
);
assert!(
!seen_tools[0].iter().any(|name| name == "deferred_mcp_tool"),
"deferred tool must stay hidden before the catalog-load authority is accepted"
);
assert!(
seen_tools[1].iter().any(|name| name == "deferred_mcp_tool"),
"real catalog-load authority should reveal the requested deferred tool at the next boundary"
);
let persisted = session_store
.load(&id)
.await
.expect("load persisted session")
.expect("persisted session should exist");
let persisted_state = persisted
.tool_visibility_state()
.expect("persisted visibility state should decode")
.expect("catalog-load run should persist visibility state");
assert!(
persisted_state
.active_requested_deferred_names
.contains("deferred_mcp_tool"),
"store projection should retain the real catalog-load requested name"
);
let persisted_witness = persisted_state
.requested_witnesses
.get("deferred_mcp_tool")
.expect("store projection must persist the catalog-derived witness");
assert_eq!(
persisted_witness.stable_owner_key.as_deref(),
Some("callback:deferred-catalog")
);
assert_eq!(
persisted_witness.last_seen_provenance.as_ref(),
Some(&deferred_catalog_provenance()),
"authority must come from the catalog entry provenance"
);
let restarted_client = Arc::new(CatalogLoadClient::default());
let restarted = catalog_route_service(
temp.path().join("sessions-restarted"),
Arc::clone(&session_store),
Arc::clone(&dispatcher),
restarted_client,
);
let resume_source = restarted
.load_authoritative_session(&id)
.await
.expect("authoritative load should succeed")
.expect("authoritative session should exist after restart");
let resumed = restarted
.create_session(catalog_route_resume_request(resume_source))
.await
.expect("resume should materialize persisted visibility state");
assert_eq!(resumed.session_id, id);
let resumed_live = restarted
.export_live_session(&id)
.await
.expect("resumed live session should export");
let resumed_state = resumed_live
.tool_visibility_state()
.expect("resumed visibility state should decode")
.expect("resumed session should carry visibility state");
assert_eq!(
resumed_state
.requested_witnesses
.get("deferred_mcp_tool")
.and_then(|witness| witness.last_seen_provenance.as_ref()),
Some(&deferred_catalog_provenance()),
"resumed live session must reactivate the persisted provenance witness"
);
let outcome = restarted
.dispatch_external_tool_call(
&id,
ToolCall::new(
"call-deferred".to_string(),
"deferred_mcp_tool".to_string(),
json!({}),
),
)
.await
.expect("restarted session should dispatch the recovered deferred tool");
assert_eq!(outcome.result.text_content(), "ran deferred_mcp_tool");
assert!(
dispatcher
.dispatched_names()
.iter()
.any(|name| name == "deferred_mcp_tool"),
"recovered visibility authority should route the actual deferred tool"
);
}
}