use std::path::PathBuf;
use std::sync::Arc;
use parking_lot::RwLock;
use zeph_llm::any::AnyProvider;
use zeph_memory::semantic::SemanticMemory;
use zeph_skills::matcher::SkillMatcherBackend;
use zeph_skills::registry::SkillRegistry;
use zeph_tools::ErasedToolExecutor;
#[derive(Clone)]
#[allow(clippy::struct_excessive_bools)] pub(crate) struct ServeAgentDeps {
pub(crate) provider: AnyProvider,
pub(crate) embedding_provider: AnyProvider,
pub(crate) registry: Arc<RwLock<SkillRegistry>>,
pub(crate) matcher: Option<SkillMatcherBackend>,
pub(crate) max_active_skills: usize,
pub(crate) skill_disambiguation_threshold: f32,
pub(crate) skill_two_stage_matching: bool,
pub(crate) skill_confusability_threshold: f32,
pub(crate) skill_group_structured: bool,
pub(crate) skill_support_similarity_threshold: f32,
pub(crate) skill_min_injection_score: f32,
pub(crate) skill_generation_provider: String,
pub(crate) skill_disambiguate_provider: String,
pub(crate) semantic_scan: bool,
pub(crate) semantic_scan_provider: String,
pub(crate) trust_config: zeph_core::config::TrustConfig,
pub(crate) rl_routing_enabled: bool,
pub(crate) rl_learning_rate: f32,
pub(crate) rl_weight: f32,
pub(crate) rl_persist_interval: u32,
pub(crate) rl_warmup_updates: u32,
pub(crate) rl_head: Option<zeph_skills::rl_head::RoutingHead>,
pub(crate) tool_executor: Arc<dyn ErasedToolExecutor>,
pub(crate) capability_scopes_config: zeph_config::CapabilityScopesConfig,
pub(crate) permission_policy: zeph_tools::PermissionPolicy,
pub(crate) audit_logger: Option<Arc<zeph_tools::AuditLogger>>,
pub(crate) policy_gate_pieces: crate::agent_setup::PolicyGatePieces,
pub(crate) memory: Arc<SemanticMemory>,
pub(crate) history_limit: u32,
pub(crate) recall_limit: usize,
pub(crate) summarization_threshold: usize,
pub(crate) session_config: zeph_core::AgentSessionConfig,
pub(crate) session_persistence_config: zeph_config::SessionConfig,
pub(crate) resume_condenser: Arc<zeph_session::LlmCondenser>,
pub(crate) resume_token_counter: Arc<zeph_agent_context::memory_backend::TokenCounterAdapter>,
pub(crate) provider_pool: Vec<zeph_core::config::ProviderEntry>,
pub(crate) provider_config_snapshot: zeph_core::ProviderConfigSnapshot,
pub(crate) shadow_sentinel_config: zeph_config::ShadowSentinelConfig,
pub(crate) shadow_sentinel_probe_provider: AnyProvider,
pub(crate) trajectory_sentinel_config: zeph_config::TrajectorySentinelConfig,
pub(crate) quality_pipeline: Option<Arc<zeph_core::quality::SelfCheckPipeline>>,
pub(crate) safe_mode: bool,
pub(crate) allowed_paths: Vec<PathBuf>,
}
pub(crate) async fn build_serve_deps(
config_path: Option<&std::path::Path>,
vault_backend: Option<&str>,
vault_key: Option<&std::path::Path>,
vault_path: Option<&std::path::Path>,
safe_mode: bool,
) -> anyhow::Result<(ServeAgentDeps, Option<String>)> {
use crate::bootstrap::AppBuilder;
let app = AppBuilder::new(config_path, vault_backend, vault_key, vault_path, safe_mode).await?;
let auth_token = resolve_auth_token(&app).await;
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
let core = crate::acp::build_shared_core(&app, &supervisor).await?;
let deps = assemble_serve_deps(&app, &core, &supervisor).await?;
Ok((deps, auth_token))
}
#[allow(clippy::too_many_lines)]
pub(crate) async fn assemble_serve_deps(
app: &crate::bootstrap::AppBuilder,
core: &crate::acp::SharedCore,
supervisor: &zeph_common::task_supervisor::TaskSupervisor,
) -> anyhow::Result<ServeAgentDeps> {
let config = app.config();
if config.cli.safe_mode {
tracing::info!(
"safe mode active: plugins and skills are disabled for every session built from \
this serve-sessions process (serve wires no hooks or MCP tools yet)"
);
}
let (tool_executor, permission_policy, audit_logger) =
build_tool_executor(config, supervisor).await?;
let policy_gate_pieces =
crate::agent_setup::build_policy_gate_pieces(config, &core.provider).await;
if policy_gate_pieces.policy_configured && policy_gate_pieces.policy_enforcer.is_none() {
anyhow::bail!(
"[tools.policy]/[tools.authorization] failed to compile (see the preceding error \
log for details) — refusing to start serve-sessions with policy enforcement \
silently disabled; fix the policy config or disable it explicitly to proceed"
);
}
tracing::info!(
trust = "on",
policy = policy_gate_pieces.policy_enforcer.is_some(),
adversarial = policy_gate_pieces.adversarial_validator.is_some(),
"serve-sessions: gate stack active (per-session trust/policy/adversarial wrap)"
);
let capability_scopes_config = config.security.capability_scopes.clone();
if !capability_scopes_config.scopes.is_empty() {
use zeph_tools::scope::build_scoped_executor;
let (composed_for_validation, _trust_snapshot) =
crate::serve::agent_factory::compose_session_tool_tree(
Arc::clone(&tool_executor),
&core.registry,
&core.memory,
zeph_memory::ConversationId(0),
zeph_config::MemoryWriteValidationConfig::default(),
);
let registry_ids: std::collections::HashSet<String> = composed_for_validation
.tool_definitions_erased()
.into_iter()
.map(|def| {
let id = def.id.to_string();
if id.contains(':') {
id
} else {
format!("builtin:{id}")
}
})
.collect();
if let Err(e) = build_scoped_executor(
zeph_tools::DynExecutor(composed_for_validation),
&capability_scopes_config,
®istry_ids,
) {
anyhow::bail!("capability_scopes: {e}");
}
}
crate::agent_setup::spawn_memory_maintenance_loops(
app,
&core.memory,
&core.provider,
supervisor,
None,
false,
"serve",
);
let session_config = zeph_core::AgentSessionConfig::from_config(config, core.budget_tokens);
let max_active_skills = config.skills.max_active_skills.get();
let history_limit = config.memory.history_limit;
let recall_limit = config.memory.semantic.recall_limit;
let summarization_threshold = config.memory.summarization_threshold;
let session_persistence_config = config.session.clone();
let (resume_condenser, resume_token_counter) =
zeph_core::provider_factory::build_resume_condenser(config, &core.provider);
let provider_config_snapshot = zeph_core::ProviderConfigSnapshot {
claude_api_key: config
.secrets
.claude_api_key
.as_ref()
.map(|s| s.expose().to_owned()),
openai_api_key: config
.secrets
.openai_api_key
.as_ref()
.map(|s| s.expose().to_owned()),
gemini_api_key: config
.secrets
.gemini_api_key
.as_ref()
.map(|s| s.expose().to_owned()),
compatible_api_keys: config
.secrets
.compatible_api_keys
.iter()
.map(|(k, v)| (k.clone(), v.expose().to_owned()))
.collect(),
llm_request_timeout_secs: config.timeouts.llm_request_timeout_secs,
embedding_model: config.llm.embedding_model.clone(),
gonka_private_key: config
.secrets
.gonka_private_key
.as_ref()
.map(|s| zeroize::Zeroizing::new(s.expose().to_owned())),
gonka_address: config
.secrets
.gonka_address
.as_ref()
.map(|s| s.expose().to_owned()),
cocoon_access_hash: config
.secrets
.cocoon_access_hash
.as_ref()
.map(|s| s.expose().to_owned()),
};
let shadow_sentinel_config = config.security.shadow_sentinel.clone();
let shadow_sentinel_probe_provider = {
let sentinel_cfg = &shadow_sentinel_config;
let base = if sentinel_cfg.probe_provider.is_empty() {
core.provider.clone()
} else {
match crate::bootstrap::create_named_provider(
sentinel_cfg.probe_provider.as_str(),
config,
) {
Ok(p) => p,
Err(e) => {
tracing::warn!(
provider = %sentinel_cfg.probe_provider,
error = %e,
"shadow_sentinel probe provider resolution failed, using primary"
);
core.provider.clone()
}
}
};
match app.secret_registry() {
Some(registry) => base.masked(registry as Arc<dyn zeph_llm::masking::OutboundMasker>),
None => base,
}
};
let trajectory_sentinel_config = config.security.trajectory.clone();
let quality_pipeline = crate::agent_setup::build_quality_pipeline(
config,
&core.provider,
app.secret_registry().as_ref(),
);
Ok(ServeAgentDeps {
provider: core.provider.clone(),
embedding_provider: core.embedding_provider.clone(),
registry: Arc::clone(&core.registry),
matcher: core.matcher.clone(),
max_active_skills,
skill_disambiguation_threshold: config.skills.disambiguation_threshold,
skill_two_stage_matching: config.skills.two_stage_matching,
skill_confusability_threshold: config.skills.confusability_threshold,
skill_group_structured: config.skills.group_structured,
skill_support_similarity_threshold: config.skills.support_similarity_threshold,
skill_min_injection_score: config.skills.min_injection_score,
skill_generation_provider: config.skills.generation_provider.as_str().to_owned(),
skill_disambiguate_provider: config.skills.disambiguate_provider.as_str().to_owned(),
semantic_scan: config.skills.semantic_scan,
semantic_scan_provider: config.skills.semantic_scan_provider.as_str().to_owned(),
trust_config: config.skills.trust.clone(),
rl_routing_enabled: config.skills.rl_routing_enabled,
rl_learning_rate: config.skills.rl_learning_rate,
rl_weight: config.skills.rl_weight,
rl_persist_interval: config.skills.rl_persist_interval,
rl_warmup_updates: config.skills.rl_warmup_updates,
rl_head: core.rl_head.clone(),
tool_executor,
capability_scopes_config,
permission_policy,
audit_logger,
policy_gate_pieces,
memory: Arc::clone(&core.memory),
history_limit,
recall_limit,
summarization_threshold,
session_config,
session_persistence_config,
resume_condenser: Arc::new(resume_condenser),
resume_token_counter,
provider_pool: config.llm.providers.clone(),
provider_config_snapshot,
shadow_sentinel_config,
shadow_sentinel_probe_provider,
trajectory_sentinel_config,
quality_pipeline,
safe_mode: config.cli.safe_mode,
allowed_paths: config
.tools
.shell
.allowed_paths
.iter()
.map(PathBuf::from)
.collect(),
})
}
pub(crate) async fn resolve_auth_token(app: &crate::bootstrap::AppBuilder) -> Option<String> {
let key = &app.config().serve.auth_token_vault_key;
if key.is_empty() {
return None;
}
app.vault().get_secret(key).await.unwrap_or_else(|e| {
tracing::warn!(
error = %e,
key = %key,
"serve-sessions: failed to resolve auth token from vault"
);
None
})
}
async fn build_tool_executor(
config: &zeph_core::config::Config,
supervisor: &zeph_common::task_supervisor::TaskSupervisor,
) -> anyhow::Result<(
Arc<dyn ErasedToolExecutor>,
zeph_tools::PermissionPolicy,
Option<Arc<zeph_tools::AuditLogger>>,
)> {
let permission_policy =
zeph_tools::build_permission_policy(&config.tools, config.security.autonomy_level);
let filter_registry = if config.tools.filters.enabled {
zeph_tools::OutputFilterRegistry::default_filters(&config.tools.filters)
} else {
zeph_tools::OutputFilterRegistry::new(false)
};
let mut shell_executor = zeph_tools::ShellExecutor::new(&config.tools.shell)
.with_permissions(permission_policy.clone())
.with_output_filters(filter_registry)
.with_task_supervisor(supervisor.clone());
if config.tools.sandbox.enabled {
let denied_present = !config.tools.sandbox.denied_domains.is_empty();
match zeph_tools::sandbox::build_sandbox_with_policy(
config.tools.sandbox.strict,
config.tools.sandbox.fail_if_unavailable,
denied_present,
) {
Ok(backend) => {
let name = backend.name();
let policy = crate::agent_setup::sandbox_policy_from_config(&config.tools.sandbox);
shell_executor = shell_executor.with_sandbox(Arc::from(backend), policy);
tracing::info!(backend = name, "OS sandbox enabled (serve-sessions)");
}
Err(e) if config.tools.sandbox.strict || config.tools.sandbox.fail_if_unavailable => {
anyhow::bail!("sandbox initialization failed: {e}");
}
Err(e) => {
tracing::warn!("OS sandbox unavailable, running without isolation: {e}");
}
}
}
let mut scrape_executor = zeph_tools::WebScrapeExecutor::new(&config.tools.scrape)
.with_egress_config(config.tools.egress.clone());
let mut audit_logger = None;
if config.tools.audit.enabled
&& let Ok(logger) = zeph_tools::AuditLogger::from_config(&config.tools.audit, false).await
{
let logger = Arc::new(logger);
shell_executor = shell_executor.with_audit(Arc::clone(&logger));
scrape_executor = scrape_executor.with_audit(Arc::clone(&logger));
audit_logger = Some(logger);
}
let file_executor = zeph_tools::FileExecutor::new(
config
.tools
.shell
.allowed_paths
.iter()
.map(PathBuf::from)
.collect(),
);
let cwd_executor = zeph_tools::SetCwdExecutor::new(
config
.tools
.shell
.allowed_paths
.iter()
.map(PathBuf::from)
.collect(),
);
let base: Arc<dyn ErasedToolExecutor> = Arc::new(zeph_tools::CompositeExecutor::new(
file_executor,
zeph_tools::CompositeExecutor::new(
shell_executor,
zeph_tools::CompositeExecutor::new(scrape_executor, cwd_executor),
),
));
Ok((base, permission_policy, audit_logger))
}
#[cfg(test)]
mod tests {
use super::*;
async fn make_memory() -> Arc<SemanticMemory> {
Arc::new(
SemanticMemory::new(
":memory:",
"http://127.0.0.1:1",
None,
AnyProvider::Mock(zeph_llm::mock::MockProvider::default()),
"test-model",
)
.await
.unwrap(),
)
}
#[tokio::test]
async fn spawn_memory_maintenance_loops_registers_all_ten_tasks() {
let mut config = zeph_core::config::Config::default();
config.memory.compression_guidelines.enabled = true;
config.memory.tree.enabled = true;
config.memory.hebbian.enabled = true;
config.memory.episodic_consolidation.enabled = true;
config.memory.optical_forgetting.enabled = true;
let app = crate::bootstrap::AppBuilder::for_test(config);
let memory = make_memory().await;
let provider = AnyProvider::Mock(zeph_llm::mock::MockProvider::default());
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
crate::agent_setup::spawn_memory_maintenance_loops(
&app,
&memory,
&provider,
&supervisor,
None,
false,
"serve",
);
let names: std::collections::HashSet<String> = supervisor
.snapshot()
.into_iter()
.map(|s| s.name.to_string())
.collect();
for expected in [
"mem-eviction",
"mem-tier-promotion",
"mem-scene-consolidation",
"mem-consolidation",
"mem-forgetting",
"mem-guidelines",
"mem-tree-consolidation",
"mem-hebbian-consolidation",
"mem-episodic-consolidation",
"mem-optical-forgetting",
] {
assert!(
names.contains(expected),
"expected {expected} registered by spawn_memory_maintenance_loops, got {names:?}"
);
}
}
#[tokio::test]
async fn spawn_memory_maintenance_loops_gates_the_five_new_tasks_off_by_default() {
let app = crate::bootstrap::AppBuilder::for_test(zeph_core::config::Config::default());
let memory = make_memory().await;
let provider = AnyProvider::Mock(zeph_llm::mock::MockProvider::default());
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
crate::agent_setup::spawn_memory_maintenance_loops(
&app,
&memory,
&provider,
&supervisor,
None,
false,
"serve",
);
let names: std::collections::HashSet<String> = supervisor
.snapshot()
.into_iter()
.map(|s| s.name.to_string())
.collect();
for expected in [
"mem-eviction",
"mem-tier-promotion",
"mem-scene-consolidation",
"mem-consolidation",
"mem-forgetting",
] {
assert!(
names.contains(expected),
"expected {expected} registered by spawn_memory_maintenance_loops even with \
Config::default(), got {names:?}"
);
}
for absent in [
"mem-guidelines",
"mem-tree-consolidation",
"mem-hebbian-consolidation",
"mem-episodic-consolidation",
"mem-optical-forgetting",
] {
assert!(
!names.contains(absent),
"expected {absent} NOT registered by spawn_memory_maintenance_loops under \
Config::default() (all five new loops default to disabled), got {names:?}"
);
}
}
fn make_test_core(memory: Arc<SemanticMemory>) -> crate::acp::SharedCore {
let provider = AnyProvider::Mock(zeph_llm::mock::MockProvider::default());
crate::acp::SharedCore {
provider: provider.clone(),
embedding_provider: provider,
registry: Arc::new(RwLock::new(zeph_skills::registry::SkillRegistry::empty())),
matcher: None,
memory,
budget_tokens: 4096,
rl_head: None,
}
}
#[tokio::test]
async fn assemble_serve_deps_validates_but_does_not_wrap_capability_scopes() {
let mut config = zeph_core::config::Config::default();
config.security.capability_scopes = zeph_config::CapabilityScopesConfig {
default_scope: "narrow".to_owned(),
scopes: std::collections::HashMap::from([(
"narrow".to_owned(),
zeph_config::ScopeConfig {
patterns: vec!["builtin:read".to_owned()],
},
)]),
..Default::default()
};
let app = crate::bootstrap::AppBuilder::for_test(config);
let memory = make_memory().await;
let core = make_test_core(memory);
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
let deps = assemble_serve_deps(&app, &core, &supervisor)
.await
.expect("assemble_serve_deps must succeed with a valid single-pattern scope");
assert_eq!(
deps.capability_scopes_config.scopes.len(),
1,
"the validated config must be carried forward on ServeAgentDeps for the \
per-session wrap"
);
let result = deps
.tool_executor
.execute_tool_call_erased(&zeph_tools::ToolCall {
tool_id: "bash".into(),
params: serde_json::Map::new(),
caller_id: None,
context: None,
tool_call_id: String::new(),
skill_name: None,
})
.await;
assert!(
!matches!(result, Err(zeph_tools::ToolError::OutOfScope { .. })),
"deps.tool_executor must stay unscoped after assemble_serve_deps — the scope wrap \
now happens per session in agent_factory::build_agent_factory, got {result:?}"
);
}
#[tokio::test]
async fn assemble_serve_deps_succeeds_with_capability_scope_on_6046_tool() {
let mut config = zeph_core::config::Config::default();
config.security.capability_scopes = zeph_config::CapabilityScopesConfig {
default_scope: "skills-only".to_owned(),
scopes: std::collections::HashMap::from([(
"skills-only".to_owned(),
zeph_config::ScopeConfig {
patterns: vec!["builtin:invoke_skill".to_owned()],
},
)]),
..Default::default()
};
let app = crate::bootstrap::AppBuilder::for_test(config);
let memory = make_memory().await;
let core = make_test_core(memory);
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
let result = assemble_serve_deps(&app, &core, &supervisor).await;
assert!(
result.is_ok(),
"a [security.capability_scopes] pattern targeting a #6046 tool (\"builtin:invoke_skill\") \
must NOT abort serve-sessions startup — it is valid against the real per-session \
registry, got err: {:?}",
result.err().map(|e| e.to_string())
);
}
#[tokio::test]
async fn assemble_serve_deps_fails_startup_on_invalid_capability_scope_pattern() {
let mut config = zeph_core::config::Config::default();
config.security.capability_scopes = zeph_config::CapabilityScopesConfig {
default_scope: "broken".to_owned(),
scopes: std::collections::HashMap::from([(
"broken".to_owned(),
zeph_config::ScopeConfig {
patterns: vec!["[".to_owned()],
},
)]),
..Default::default()
};
let app = crate::bootstrap::AppBuilder::for_test(config);
let memory = make_memory().await;
let core = make_test_core(memory);
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
let result = assemble_serve_deps(&app, &core, &supervisor).await;
assert!(
result.is_err(),
"an invalid capability_scopes glob pattern must abort serve-sessions startup"
);
}
#[tokio::test]
async fn assemble_serve_deps_fails_startup_on_policy_compile_failure() {
let mut config = zeph_core::config::Config::default();
config.tools.policy = zeph_tools::PolicyConfig {
enabled: true,
default_effect: zeph_tools::DefaultEffect::Allow,
rules: vec![zeph_tools::PolicyRuleConfig {
effect: zeph_tools::PolicyEffect::Deny,
tool: "shell".into(),
paths: vec![],
env: vec![],
trust_level: None,
args_match: Some("(".to_owned()),
capabilities: vec![],
}],
..Default::default()
};
let app = crate::bootstrap::AppBuilder::for_test(config);
let memory = make_memory().await;
let core = make_test_core(memory);
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
let result = assemble_serve_deps(&app, &core, &supervisor).await;
assert!(
result.is_err(),
"a [tools.policy] compile failure must abort serve-sessions startup (fail-closed), \
not silently disable policy enforcement"
);
}
#[tokio::test]
async fn assemble_serve_deps_succeeds_when_policy_not_configured() {
let config = zeph_core::config::Config::default();
let app = crate::bootstrap::AppBuilder::for_test(config);
let memory = make_memory().await;
let core = make_test_core(memory);
let cancel = tokio_util::sync::CancellationToken::new();
let supervisor = zeph_common::task_supervisor::TaskSupervisor::new(cancel);
let deps = assemble_serve_deps(&app, &core, &supervisor).await.expect(
"[tools.policy] absent by config is legitimate and must stay fail-open, not \
abort startup",
);
assert!(deps.policy_gate_pieces.policy_enforcer.is_none());
}
}