use std::future::Future;
use std::pin::Pin;
use tracing::Instrument as _;
use zeph_commands::{CommandError, SessionControlAccess};
use zeph_llm::provider::LlmProvider;
use super::Agent;
use super::error;
use crate::channel::Channel;
pub(crate) fn format_overlay_section(plugins_dir: &std::path::Path) -> String {
let mut cfg = zeph_config::Config::default();
match zeph_plugins::apply_plugin_config_overlays(&mut cfg, plugins_dir) {
Err(e) => format!("overlay resolution failed: {e}"),
Ok(overlay) => {
if overlay.source_plugins.is_empty() && overlay.skipped_plugins.is_empty() {
return "No plugin overlay active.".to_owned();
}
let mut out = String::from("Active plugin overlay:\n");
if overlay.source_plugins.is_empty() {
out.push_str(" Source plugins: (none)\n");
} else {
out.push_str(" Source plugins: ");
out.push_str(&overlay.source_plugins.join(", "));
out.push('\n');
}
if overlay.skipped_plugins.is_empty() {
out.push_str(" Skipped plugins: (none)\n");
} else {
out.push_str(" Skipped plugins:\n");
for reason in &overlay.skipped_plugins {
out.push_str(" - ");
out.push_str(reason);
out.push('\n');
}
}
out.push_str(
" Note: overlay values shown against default config — run with --config for live intersection.",
);
out
}
}
}
impl<C: crate::channel::Channel> Agent<C> {
#[allow(clippy::unused_self)]
pub(super) fn handle_builtin_command(&self, _trimmed: &str) -> Option<bool> {
None
}
pub(super) async fn dispatch_slash_command(
&mut self,
trimmed: &str,
) -> Option<Result<(), error::AgentError>> {
if trimmed.starts_with('@') {
return self.dispatch_agent_command(trimmed).await;
}
if trimmed.eq_ignore_ascii_case("/subagent")
|| trimmed.to_ascii_lowercase().starts_with("/subagent ")
{
let args = trimmed.get("/subagent".len()..).unwrap_or("").trim();
return Some(self.handle_subagent_slash(args).await);
}
None
}
async fn handle_subagent_slash(&mut self, args: &str) -> Result<(), error::AgentError> {
let msg: String = if args.is_empty() {
"Usage: /subagent <subcommand>\n\nSubcommands:\n spawn <command> Spawn an ACP sub-agent process".to_owned()
} else {
let (subcmd, rest) = args.split_once(' ').unwrap_or((args, ""));
match subcmd {
"spawn" => {
let cmd = rest.trim();
let effective_mode = self.effective_delegation_mode();
if cmd.is_empty() {
"Usage: /subagent spawn <command>\n\nExample: /subagent spawn zeph --acp"
.to_owned()
} else if !effective_mode.permits_explicit() {
tracing::warn!(
mode = ?effective_mode,
"/subagent spawn rejected: delegation disabled by configuration"
);
"Sub-agent delegation is disabled by configuration \
([agents].delegation_mode = \"disabled\" or [agents].enabled = false)."
.to_owned()
} else if let Some(spawn_fn) = self.runtime.config.acp_subagent_spawn_fn.clone()
{
let cmd = cmd.to_owned();
match spawn_fn(cmd).await {
Ok(output) => output,
Err(e) => format!("Sub-agent error: {e}"),
}
} else {
"ACP sub-agent spawning is not available in this mode.\n\
Use `zeph acp run-agent --command <CMD> --prompt <TEXT>` for one-shot sessions."
.to_owned()
}
}
other => format!("Unknown /subagent subcommand: '{other}'. Available: spawn"),
}
};
let _ = self.channel.send(&msg).await;
let _ = self.channel.flush_chunks().await;
Ok(())
}
pub(super) async fn dispatch_agent_command(
&mut self,
trimmed: &str,
) -> Option<Result<(), error::AgentError>> {
let known: Vec<String> = self
.services
.orchestration
.subagent_manager
.as_ref()
.map(|m| m.definitions().iter().map(|d| d.name.clone()).collect())
.unwrap_or_default();
match zeph_subagent::AgentCommand::parse(trimmed, &known) {
Ok(cmd) => {
if let Some(msg) = self.handle_agent_command(cmd).await
&& let Err(e) = self.channel.send(&msg).await
{
return Some(Err(e.into()));
}
let _ = self.channel.flush_chunks().await;
Some(Ok(()))
}
Err(e) if trimmed.starts_with('@') => {
tracing::debug!("@mention not matched as agent: {e}");
None
}
Err(e) => {
if let Err(send_err) = self.channel.send(&e.to_string()).await {
return Some(Err(send_err.into()));
}
let _ = self.channel.flush_chunks().await;
Some(Ok(()))
}
}
}
pub(super) fn handle_status_as_string(&mut self) -> String {
use std::fmt::Write;
use zeph_llm::provider::Role;
let uptime = self.runtime.lifecycle.start_time.elapsed().as_secs();
let msg_count = self
.msg
.messages
.iter()
.filter(|m| m.role == Role::User)
.count();
let metrics = collect_status_metrics(self.runtime.metrics.metrics_tx.as_ref());
let skill_count = self.services.skill.registry.read().all_meta().len();
let mut out = String::from("Session status:\n\n");
let _ = writeln!(out, "Provider: {}", self.provider.name());
let _ = writeln!(out, "Model: {}", self.runtime.config.model_name);
let _ = writeln!(out, "Uptime: {uptime}s");
let _ = writeln!(out, "Turns: {msg_count}");
let _ = writeln!(out, "API calls: {}", metrics.api_calls);
if metrics.reasoning_tokens > 0 {
let _ = writeln!(
out,
"Tokens: {} prompt / {} completion ({} reasoning, subset of completion)",
metrics.prompt_tokens, metrics.completion_tokens, metrics.reasoning_tokens
);
} else {
let _ = writeln!(
out,
"Tokens: {} prompt / {} completion",
metrics.prompt_tokens, metrics.completion_tokens
);
}
let _ = writeln!(out, "Skills: {skill_count}");
let _ = writeln!(out, "MCP: {} server(s)", metrics.mcp_servers);
if let Some(ref tf) = self.services.tool_state.tool_schema_filter {
let _ = writeln!(
out,
"Filter: enabled (top_k={}, always_on={}, {} embeddings)",
tf.top_k(),
tf.always_on_count(),
tf.embedding_count(),
);
}
if let Some(ref adv) = self.runtime.config.adversarial_policy_info {
let provider_display = if adv.provider.is_empty() {
"default"
} else {
adv.provider.as_str()
};
let _ = writeln!(
out,
"Adv gate: enabled (provider={}, policies={}, fail_open={}, timeout_ms={})",
provider_display, adv.policy_count, adv.fail_open, adv.timeout_ms
);
}
append_cost_section(&mut out, metrics.cost_cents, &metrics.provider_breakdown);
append_orchestration_section(
&mut out,
metrics.orch_plans,
metrics.orch_tasks,
metrics.orch_completed,
metrics.orch_failed,
metrics.orch_skipped,
);
append_ensemble_section(
&mut out,
metrics.ensemble_degraded,
metrics.ensemble_agreement_ratio,
&metrics.ensemble_member_stats,
);
append_pruning_section(
&mut out,
self.context_manager.compression.pruning_strategy,
self.services.compression.subgoal_registry.subgoals.len(),
self.services.compression.subgoal_registry.active_subgoal(),
);
append_graph_recall_section(&mut out, &self.services.memory.extraction.graph_config);
out.trim_end().to_owned()
}
pub(super) fn format_guardrail_status(&self) -> String {
use std::fmt::Write;
let mut out = String::new();
if let Some(ref guardrail) = self.services.security.guardrail {
let stats = guardrail.stats();
let _ = writeln!(out, "Guardrail: enabled");
let _ = writeln!(out, "Action: {:?}", guardrail.action());
let _ = writeln!(out, "Fail strategy: {:?}", guardrail.fail_strategy());
let _ = writeln!(out, "Timeout: {}ms", guardrail.timeout_ms());
let _ = writeln!(
out,
"Tool scan: {}",
if guardrail.scan_tool_output() {
"enabled"
} else {
"disabled"
}
);
let _ = writeln!(out, "\nStats:");
let _ = writeln!(out, " Total checks: {}", stats.total_checks);
let _ = writeln!(out, " Flagged: {}", stats.flagged_count);
let _ = writeln!(out, " Errors: {}", stats.error_count);
let _ = writeln!(out, " Avg latency: {}ms", stats.avg_latency_ms());
} else {
out.push_str("Guardrail: disabled\n");
out.push_str(
"Enable with: --guardrail flag or [security.guardrail] enabled = true in config",
);
}
out.trim_end().to_owned()
}
pub(super) fn format_focus_status(&self) -> String {
use std::fmt::Write;
let mut out = String::from("Focus Agent status\n\n");
let _ = writeln!(
out,
"Enabled: {}",
self.services.focus.config.enabled
);
let _ = writeln!(out, "Active session: {}", self.services.focus.is_active());
if let Some(ref scope) = self.services.focus.active_scope {
let _ = writeln!(out, "Active scope: {scope}");
}
let _ = writeln!(
out,
"Knowledge blocks: {}",
self.services.focus.knowledge_blocks.len()
);
let _ = writeln!(
out,
"Turns since focus: {}",
self.services.focus.turns_since_focus
);
out.trim_end().to_owned()
}
pub(super) fn format_sidequest_status(&self) -> String {
use std::fmt::Write;
let mut out = String::from("SideQuest status\n\n");
let _ = writeln!(
out,
"Enabled: {}",
self.services.sidequest.config.enabled
);
let _ = writeln!(
out,
"Interval turns: {}",
self.services.sidequest.config.interval_turns
);
let _ = writeln!(
out,
"Turn counter: {}",
self.services.sidequest.turn_counter
);
let _ = writeln!(
out,
"Passes run: {}",
self.services.sidequest.passes_run
);
let _ = writeln!(
out,
"Total evicted: {} tool outputs",
self.services.sidequest.total_evicted
);
out.trim_end().to_owned()
}
#[cfg_attr(not(test), allow(dead_code))]
pub(super) fn handle_image_as_string(&mut self, path: &str) -> String {
use zeph_common::path_guard::{PathRejection, classify_relative_path};
use zeph_llm::provider::{ImageData, MessagePart};
match classify_relative_path(path) {
PathRejection::Allowed => {}
PathRejection::Absolute => {
return "Invalid image path: absolute paths are not supported, use a path \
relative to the working directory"
.to_owned();
}
PathRejection::Traversal => {
return "Invalid image path: path traversal ('..') is not allowed".to_owned();
}
}
let data = match std::fs::read(path) {
Ok(d) => d,
Err(e) => return format!("Cannot read image {path}: {e}"),
};
if data.len() > super::message_queue::MAX_IMAGE_BYTES {
return format!(
"Image {path} exceeds size limit ({} MB), skipping",
super::message_queue::MAX_IMAGE_BYTES / 1024 / 1024
);
}
let mime_type = super::message_queue::detect_image_mime(Some(path)).to_string();
self.msg
.pending_image_parts
.push(MessagePart::Image(Box::new(ImageData { data, mime_type })));
format!("Image loaded: {path}. Send your message.")
}
#[allow(clippy::needless_pass_by_value)]
pub(super) fn run_plugin_command(
args: &str,
managed_dir: Option<std::path::PathBuf>,
mcp_allowed: Vec<String>,
base_shell_allowed: Vec<String>,
ephemeral_plugin_names: Vec<String>,
reputation_cfg: &zeph_config::plugins::ReputationConfig,
) -> String {
let plugins_dir = zeph_plugins::PluginManager::default_plugins_dir();
let (subcmd, rest) = args.trim().split_once(' ').unwrap_or((args.trim(), ""));
if subcmd == "overlay" || (matches!(subcmd, "" | "list") && rest.trim() == "--overlay") {
return format_overlay_section(&plugins_dir);
}
let managed_dir = managed_dir
.unwrap_or_else(|| zeph_config::defaults::default_vault_dir().join("skills"));
let mgr = zeph_plugins::PluginManager::new(
plugins_dir,
managed_dir,
mcp_allowed,
base_shell_allowed,
)
.with_reputation_config(reputation_cfg, false);
match subcmd {
"" | "list" => match mgr.list_installed() {
Ok(plugins) if plugins.is_empty() && ephemeral_plugin_names.is_empty() => {
"No plugins installed.".to_owned()
}
Ok(plugins) => {
let mut lines: Vec<String> = plugins
.iter()
.map(|p| format!("{} v{} — {}", p.name, p.version, p.description))
.collect();
for name in &ephemeral_plugin_names {
lines.push(format!("{name} [ephemeral]"));
}
lines.join("\n")
}
Err(e) => format!("plugin list failed: {e}"),
},
"add" => {
use std::fmt::Write as _;
if rest.is_empty() {
return "Usage: /plugins add <source>".to_owned();
}
match mgr.add(rest.trim()) {
Ok(r) => {
let mut out = format!("Installed plugin \"{}\"", r.name);
if !r.installed_skills.is_empty() {
let _ = write!(out, "\n Skills: {}", r.installed_skills.join(", "));
}
if !r.mcp_server_ids.is_empty() {
let _ = write!(
out,
"\n MCP servers (restart required): {}",
r.mcp_server_ids.join(", ")
);
}
for w in &r.warnings {
let _ = write!(out, "\n warning: {w}");
}
out
}
Err(e) => format!("plugin add failed: {e}"),
}
}
"remove" => {
use std::fmt::Write as _;
if rest.is_empty() {
return "Usage: /plugins remove <name>".to_owned();
}
match mgr.remove(rest.trim()) {
Ok(r) => {
let mut out = format!("Removed plugin \"{}\"", rest.trim());
if !r.removed_skills.is_empty() {
let _ =
write!(out, "\n Removed skills: {}", r.removed_skills.join(", "));
}
out
}
Err(e) => format!("plugin remove failed: {e}"),
}
}
other => {
format!(
"Unknown /plugins subcommand: '{other}'. Available: list, list --overlay, overlay, add, remove"
)
}
}
}
#[tracing::instrument(skip_all, name = "core.agent.handle_skills")]
pub(super) async fn handle_skills_as_string(
&mut self,
subcommand: &str,
) -> Result<String, error::AgentError> {
match subcommand {
"" => self.handle_skills_command_as_string().await,
"confusability" => self.handle_skills_confusability_as_string().await,
"injection" => self.handle_skills_injection_as_string(),
"trust" => self.handle_skills_trust_as_string(),
other => Ok(format!(
"Unknown /skills subcommand: '{other}'. Available: confusability, injection, trust"
)),
}
}
#[tracing::instrument(skip_all, name = "core.agent.handle_skills_command")]
async fn handle_skills_command_as_string(&mut self) -> Result<String, error::AgentError> {
use std::collections::BTreeMap;
use std::fmt::Write;
let (all_meta, load_errors): (
Vec<zeph_skills::loader::SkillMeta>,
Vec<(std::path::PathBuf, String)>,
) = {
let reg = self.services.skill.registry.read();
(
reg.all_meta().into_iter().cloned().collect(),
reg.load_errors().to_vec(),
)
};
let memory = self.services.memory.persistence.memory.clone();
let mut trust_map: std::collections::HashMap<String, String> =
std::collections::HashMap::new();
for meta in &all_meta {
if let Some(ref memory) = memory {
let info = memory
.sqlite()
.load_skill_trust(&meta.name)
.await
.ok()
.flatten()
.map_or_else(String::new, |r| format!(" [{}]", r.trust_level));
trust_map.insert(meta.name.clone(), info);
}
}
let mut output = String::from("Available skills:\n\n");
let has_categories = all_meta.iter().any(|m| m.category.is_some());
if has_categories {
let mut by_category: BTreeMap<&str, Vec<&zeph_skills::loader::SkillMeta>> =
BTreeMap::new();
for meta in &all_meta {
let cat = meta.category.as_deref().unwrap_or("other");
by_category.entry(cat).or_default().push(meta);
}
for (cat, skills) in &by_category {
let _ = writeln!(output, "[{cat}]");
for meta in skills {
let trust_info = trust_map.get(&meta.name).map_or("", String::as_str);
let _ = writeln!(output, "- {} — {}{trust_info}", meta.name, meta.description);
}
output.push('\n');
}
} else {
for meta in &all_meta {
let trust_info = trust_map.get(&meta.name).map_or("", String::as_str);
let _ = writeln!(output, "- {} — {}{trust_info}", meta.name, meta.description);
}
}
if let Some(ref memory) = memory {
match memory.sqlite().load_skill_usage().await {
Ok(usage) if !usage.is_empty() => {
output.push_str("\nUsage statistics:\n\n");
for row in &usage {
let _ = writeln!(
output,
"- {}: {} invocations (last: {})",
row.skill_name, row.invocation_count, row.last_used_at,
);
}
}
Ok(_) => {}
Err(e) => tracing::warn!("failed to load skill usage: {e:#}"),
}
}
if !load_errors.is_empty() {
output.push_str("\nFailed to load:\n");
for (path, reason) in &load_errors {
let _ = writeln!(output, "- {}: {reason}", path.display());
}
}
Ok(output)
}
pub(crate) fn start_user_loop(&mut self, prompt: String, interval_secs: u64) {
use std::time::Duration;
use tokio::time::{Instant, MissedTickBehavior};
let period = Duration::from_secs(interval_secs);
let mut interval = tokio::time::interval_at(Instant::now() + period, period);
interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
let cancel_tx = tokio_util::sync::CancellationToken::new();
self.runtime.lifecycle.user_loop = Some(crate::agent::state::LoopState {
prompt,
iteration: 0,
interval,
cancel_tx,
});
}
pub(crate) fn stop_user_loop(&mut self) -> String {
if let Some(ls) = self.runtime.lifecycle.user_loop.take() {
let iters = ls.iteration;
ls.cancel_tx.cancel();
format!("Loop stopped after {iters} iteration(s).")
} else {
"No active loop.".to_owned()
}
}
#[tracing::instrument(skip_all, name = "core.agent.handle_skills_confusability")]
async fn handle_skills_confusability_as_string(&mut self) -> Result<String, error::AgentError> {
let threshold = self.services.skill.confusability_threshold;
if threshold <= 0.0 {
return Ok("Confusability monitoring is disabled. \
Set [skills] confusability_threshold in config (e.g. 0.85) to enable."
.to_owned());
}
let Some(matcher) = &self.services.skill.matcher else {
return Ok(
"Skill matcher not available (no embedding provider configured).".to_owned(),
);
};
let all_meta: Vec<zeph_skills::loader::SkillMeta> = self
.services
.skill
.registry
.read()
.all_meta()
.into_iter()
.cloned()
.collect();
let refs: Vec<&zeph_skills::loader::SkillMeta> = all_meta.iter().collect();
let report = matcher.confusability_report(&refs, threshold).await;
Ok(report.to_string())
}
#[tracing::instrument(skip_all, name = "core.agent.handle_skills_injection")]
fn handle_skills_injection_as_string(&self) -> Result<String, error::AgentError> {
Ok(format!(
"Skill injection config: group_structured={}, support_similarity_threshold={:.2}, min_injection_score={:.2}",
self.services.skill.group_structured,
self.services.skill.support_similarity_threshold,
self.services.skill.min_injection_score,
))
}
#[tracing::instrument(skip_all, name = "core.agent.handle_skills_trust")]
fn handle_skills_trust_as_string(&self) -> Result<String, error::AgentError> {
let trust = &self.services.skill.trust_config;
let rl_enabled = self
.services
.learning_engine
.rl_routing
.as_ref()
.is_some_and(|r| r.enabled);
Ok(format!(
"Skill trust config: default_level={:?}, local_level={:?}, bundled_level={:?}, \
hash_mismatch_level={:?} | RL routing: enabled={}, rl_head_loaded={}",
trust.default_level,
trust.local_level,
trust.bundled_level,
trust.hash_mismatch_level,
rl_enabled,
self.services.skill.rl_head.is_some(),
))
}
}
pub(crate) fn build_session_debug_registry<'ctx>()
-> zeph_commands::CommandRegistry<zeph_commands::CommandContext<'ctx>> {
use zeph_commands::CommandRegistry;
use zeph_commands::handlers::debug::{DebugDumpCommand, DumpFormatCommand, LogCommand};
use zeph_commands::handlers::help::HelpCommand;
use zeph_commands::handlers::session::{
ClearCommand, ClearQueueCommand, ExitCommand, HistoryCommand, QuitCommand, ResetCommand,
};
let mut reg = CommandRegistry::new();
reg.register(ExitCommand);
reg.register(QuitCommand);
reg.register(ClearCommand);
reg.register(ResetCommand);
reg.register(ClearQueueCommand);
reg.register(HistoryCommand);
reg.register(LogCommand);
reg.register(DebugDumpCommand);
reg.register(DumpFormatCommand);
reg.register(HelpCommand);
#[cfg(test)]
reg.register(super::test_stubs::TestErrorCommand);
reg
}
pub(crate) fn build_agent_command_registry<'ctx>()
-> zeph_commands::CommandRegistry<zeph_commands::CommandContext<'ctx>> {
use zeph_commands::CommandRegistry;
use zeph_commands::handlers::{
acp::AcpCommand,
agent_cmd::AgentCommand,
agents_fleet::AgentsFleetCommand,
caveman::CavemanCommand,
cd::CdCommand,
checkpoint::{RedoCommand, UndoCommand},
compaction::{CompactCommand, NewConversationCommand, RecapCommand},
conv::ConvCommand,
experiment::ExperimentCommand,
goal::GoalCommand,
loop_cmd::LoopCommand,
lsp::LspCommand,
mcp::McpCommand,
memory::{
GraphCommand, GuidelinesCommand, KnowledgeSlashCommand, MemoryCommand,
StoreSlashCommand,
},
misc::{CacheStatsCommand, ImageCommand, NotifyTestCommand},
model::{ModelCommand, ProviderCommand},
plan::PlanCommand,
plugins::PluginsCommand,
policy::PolicyCommand,
reasoning_effort::ReasoningEffortCommand,
scheduler::SchedulerCommand,
search::SearchCommand,
skill::{FeedbackCommand, SkillCommand, SkillsCommand},
status::{FocusCommand, GuardrailCommand, SideQuestCommand, StatusCommand},
think_tokens::ThinkTokensCommand,
trajectory::{ScopeCommand, TrajectoryCommand},
worktree::WorktreeCommand,
};
let mut agent_reg = CommandRegistry::new();
agent_reg.register(CavemanCommand);
agent_reg.register(CdCommand);
agent_reg.register(MemoryCommand);
agent_reg.register(StoreSlashCommand);
agent_reg.register(GraphCommand);
agent_reg.register(KnowledgeSlashCommand);
agent_reg.register(GuidelinesCommand);
agent_reg.register(ModelCommand);
agent_reg.register(ProviderCommand);
agent_reg.register(ThinkTokensCommand);
agent_reg.register(ReasoningEffortCommand);
agent_reg.register(SkillCommand);
agent_reg.register(SkillsCommand);
agent_reg.register(FeedbackCommand);
agent_reg.register(McpCommand);
agent_reg.register(PolicyCommand);
agent_reg.register(SchedulerCommand);
agent_reg.register(SearchCommand);
agent_reg.register(LspCommand);
agent_reg.register(CacheStatsCommand);
agent_reg.register(ImageCommand);
agent_reg.register(NotifyTestCommand);
agent_reg.register(StatusCommand);
agent_reg.register(GuardrailCommand);
agent_reg.register(FocusCommand);
agent_reg.register(SideQuestCommand);
agent_reg.register(AgentCommand);
agent_reg.register(AgentsFleetCommand);
agent_reg.register(CompactCommand);
agent_reg.register(NewConversationCommand);
agent_reg.register(RecapCommand);
agent_reg.register(ExperimentCommand);
agent_reg.register(PlanCommand);
agent_reg.register(LoopCommand);
agent_reg.register(PluginsCommand);
agent_reg.register(AcpCommand);
#[cfg(feature = "cocoon")]
agent_reg.register(zeph_commands::handlers::cocoon::CocoonCommand);
agent_reg.register(TrajectoryCommand);
agent_reg.register(ScopeCommand);
agent_reg.register(GoalCommand);
agent_reg.register(UndoCommand);
agent_reg.register(RedoCommand);
agent_reg.register(ConvCommand);
agent_reg.register(WorktreeCommand);
agent_reg
}
struct StatusMetrics {
api_calls: u64,
prompt_tokens: u64,
completion_tokens: u64,
reasoning_tokens: u64,
cost_cents: f64,
mcp_servers: usize,
orch_plans: u64,
orch_tasks: u64,
orch_completed: u64,
orch_failed: u64,
orch_skipped: u64,
ensemble_degraded: u64,
ensemble_agreement_ratio: Option<f64>,
ensemble_member_stats: Vec<(String, f64, u64)>,
provider_breakdown: Vec<(String, crate::cost::ProviderUsage)>,
}
fn collect_status_metrics(
metrics_tx: Option<&tokio::sync::watch::Sender<crate::metrics::MetricsSnapshot>>,
) -> StatusMetrics {
if let Some(tx) = metrics_tx {
let m = tx.borrow();
StatusMetrics {
api_calls: m.api_calls,
prompt_tokens: m.prompt_tokens,
completion_tokens: m.completion_tokens,
reasoning_tokens: m.reasoning_tokens,
cost_cents: m.cost_spent_cents,
mcp_servers: m.mcp_server_count,
orch_plans: m.orchestration.plans_total,
orch_tasks: m.orchestration.tasks_total,
orch_completed: m.orchestration.tasks_completed,
orch_failed: m.orchestration.tasks_failed,
orch_skipped: m.orchestration.tasks_skipped,
ensemble_degraded: m.orchestration.ensemble_degraded_total,
ensemble_agreement_ratio: m.orchestration.ensemble_last_agreement_ratio,
ensemble_member_stats: m.orchestration.ensemble_member_stats.clone(),
provider_breakdown: m.provider_cost_breakdown.clone(),
}
} else {
StatusMetrics {
api_calls: 0,
prompt_tokens: 0,
completion_tokens: 0,
reasoning_tokens: 0,
cost_cents: 0.0,
mcp_servers: 0,
orch_plans: 0,
orch_tasks: 0,
orch_completed: 0,
orch_failed: 0,
orch_skipped: 0,
ensemble_degraded: 0,
ensemble_agreement_ratio: None,
ensemble_member_stats: vec![],
provider_breakdown: vec![],
}
}
}
fn append_cost_section(
out: &mut String,
cost_cents: f64,
provider_breakdown: &[(String, crate::cost::ProviderUsage)],
) {
use std::fmt::Write;
if cost_cents > 0.0 {
let _ = writeln!(out, "Cost: ${:.4}", cost_cents / 100.0);
if !provider_breakdown.is_empty() {
let _ = writeln!(
out,
" {:<16} {:>8} {:>8} {:>8}",
"Provider", "Requests", "Tokens", "Cost"
);
for (name, usage) in provider_breakdown {
let total_tokens = usage.input_tokens + usage.output_tokens;
let _ = writeln!(
out,
" {:<16} {:>8} {:>8} {:>8}",
name,
usage.request_count,
total_tokens,
format!("${:.4}", usage.cost_cents / 100.0),
);
}
}
}
}
fn append_orchestration_section(
out: &mut String,
orch_plans: u64,
orch_tasks: u64,
orch_completed: u64,
orch_failed: u64,
orch_skipped: u64,
) {
use std::fmt::Write;
if orch_plans > 0 {
let _ = writeln!(out);
let _ = writeln!(out, "Orchestration:");
let _ = writeln!(out, " Plans: {orch_plans}");
let _ = writeln!(out, " Tasks: {orch_completed}/{orch_tasks} completed");
if orch_failed > 0 {
let _ = writeln!(out, " Failed: {orch_failed}");
}
if orch_skipped > 0 {
let _ = writeln!(out, " Skipped: {orch_skipped}");
}
}
}
fn append_ensemble_section(
out: &mut String,
ensemble_degraded: u64,
ensemble_agreement_ratio: Option<f64>,
ensemble_member_stats: &[(String, f64, u64)],
) {
use std::fmt::Write;
if ensemble_member_stats.is_empty() && ensemble_degraded == 0 {
return;
}
let _ = writeln!(out);
let _ = writeln!(out, "Ensemble verify:");
if let Some(ratio) = ensemble_agreement_ratio {
let _ = writeln!(out, " Last agreement: {:.0}%", ratio * 100.0);
}
if ensemble_degraded > 0 {
let _ = writeln!(out, " Degraded: {ensemble_degraded} (quorum fallback)");
}
for (member, score, observations) in ensemble_member_stats {
let _ = writeln!(out, " {member:<16} ema={score:.2} (n={observations})");
}
}
fn append_pruning_section(
out: &mut String,
pruning_strategy: crate::config::PruningStrategy,
subgoal_count: usize,
active_subgoal: Option<&zeph_agent_context::compaction::Subgoal>,
) {
use crate::config::PruningStrategy;
use std::fmt::Write;
if matches!(
pruning_strategy,
PruningStrategy::Subgoal | PruningStrategy::SubgoalMig
) {
let _ = writeln!(out);
let _ = writeln!(
out,
"Pruning: {}",
match pruning_strategy {
PruningStrategy::SubgoalMig => "subgoal_mig",
_ => "subgoal",
}
);
let _ = writeln!(out, "Subgoals: {subgoal_count} tracked");
if let Some(active) = active_subgoal {
let _ = writeln!(out, "Active: \"{}\"", active.description);
} else {
let _ = writeln!(out, "Active: (none yet)");
}
}
}
fn append_graph_recall_section(out: &mut String, gc: &zeph_config::memory::GraphConfig) {
use std::fmt::Write;
if gc.enabled {
let _ = writeln!(out);
if gc.spreading_activation.enabled {
let _ = writeln!(
out,
"Graph recall: spreading activation (lambda={:.2}, hops={})",
gc.spreading_activation.decay_lambda, gc.spreading_activation.max_hops,
);
} else {
let _ = writeln!(out, "Graph recall: BFS (hops={})", gc.max_hops);
}
}
}
impl<C: Channel> Agent<C> {
async fn handle_conv_resume(&mut self, id: &str) -> Result<String, CommandError> {
if id.is_empty() {
return Ok("Usage: /conv resume <id>".to_owned());
}
if let Some(sink) = &self.services.session.session_sink
&& sink.session_id().as_str() == id
{
return Ok(format!("Already in session '{id}'."));
}
let Some(memory) = self.services.memory.persistence.memory.clone() else {
return Ok(
"Conversation-session persistence requires memory to be enabled ([memory] enabled = true)."
.to_owned(),
);
};
let store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
let Some(metadata) = store
.get(id)
.await
.map_err(|e| CommandError::new(e.to_string()))?
else {
return Ok(format!("Session '{id}' not found."));
};
let conversation_id = if let Some(cid) = metadata.conversation_id {
zeph_memory::ConversationId(cid)
} else {
let cid = memory
.sqlite()
.create_conversation()
.await
.map_err(|e| CommandError::new(e.to_string()))?;
store
.link_conversation(id, cid.0)
.await
.map_err(|e| CommandError::new(e.to_string()))?;
cid
};
let session_id = zeph_common::SessionId::new(id);
self.load_and_resume_conversation(&session_id, conversation_id)
.await
.map_err(|e| CommandError::new(e.to_string()))?;
Ok(format!(
"Resumed session {id} ({} event(s) replayed).",
metadata.event_count
))
}
async fn handle_conv_fork(&mut self, id: &str) -> Result<String, CommandError> {
if id.is_empty() {
return Ok("Usage: /conv fork <id>".to_owned());
}
let Some(memory) = self.services.memory.persistence.memory.clone() else {
return Ok(
"Conversation-session persistence requires memory to be enabled ([memory] enabled = true)."
.to_owned(),
);
};
let Some(session_persistence_config) =
self.services.session.session_persistence_config.clone()
else {
return Ok(
"Conversation-session persistence is not enabled ([session] enabled = true)."
.to_owned(),
);
};
let store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
let data_dir = std::path::PathBuf::from(&session_persistence_config.data_dir);
let new_id = zeph_common::SessionId::generate();
let fork_result =
zeph_session::ForkEngine::fork(&data_dir, id, new_id.as_str(), None, &store, None)
.await
.map_err(|e| CommandError::new(e.to_string()))?;
let conversation_id = memory
.sqlite()
.create_conversation()
.await
.map_err(|e| CommandError::new(e.to_string()))?;
self.load_and_resume_conversation(&new_id, conversation_id)
.await
.map_err(|e| CommandError::new(e.to_string()))?;
Ok(format!(
"Forked session {id} -> {} ({} event(s) copied); now the active conversation.",
fork_result.new_session_id, fork_result.events_copied
))
}
}
async fn handle_conv_list(store: &zeph_session::SessionStore) -> Result<String, CommandError> {
use std::fmt::Write as _;
let sessions = store
.list(&zeph_session::SessionFilter::default())
.await
.map_err(|e| CommandError::new(format!("failed to list sessions: {e}")))?;
if sessions.is_empty() {
return Ok("No conversation-sessions found.".to_owned());
}
let mut out = format!(
"{:<38} {:<30} {:<9} {:>6} {:<24}\n",
"ID", "TITLE", "STATUS", "EVENTS", "UPDATED"
);
out.push_str(&"-".repeat(110));
out.push('\n');
for s in &sessions {
let title = s.title.as_deref().unwrap_or("(untitled)");
let _ = writeln!(
out,
"{:<38} {:<30} {:<9} {:>6} {:<24}",
s.session_id,
crate::text::truncate_to_chars(title, 30),
s.status.as_str(),
s.event_count,
s.updated_at
);
}
Ok(out.trim_end().to_owned())
}
async fn handle_conv_show(
store: &zeph_session::SessionStore,
id: &str,
) -> Result<String, CommandError> {
if id.is_empty() {
return Ok("Usage: /conv show <id>".to_owned());
}
let metadata = store
.get(id)
.await
.map_err(|e| CommandError::new(format!("failed to read session metadata: {e}")))?;
let Some(m) = metadata else {
return Ok(format!("Session '{id}' not found."));
};
Ok(format!(
"Session {}\n title: {}\n status: {}\n events: {} (last_seq={})\n forked_from: {}\n created: {}\n updated: {}",
m.session_id,
m.title.as_deref().unwrap_or("(untitled)"),
m.status.as_str(),
m.event_count,
m.last_seq,
m.forked_from.as_deref().unwrap_or("-"),
m.created_at,
m.updated_at
))
}
impl<C: Channel + Send + 'static> SessionControlAccess for Agent<C> {
fn session_recap<'a>(
&'a mut self,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
Box::pin(
async move {
match self.build_recap().await {
Ok(text) => Ok(text),
Err(e) => {
tracing::warn!("session recap command: {}", e.0);
Ok("Recap unavailable — see logs for details".to_string())
}
}
}
.instrument(tracing::info_span!("core.agent_access.session_recap")),
)
}
fn compact_context<'a>(
&'a mut self,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
Box::pin(
self.compact_context_command()
.instrument(tracing::info_span!("core.agent_access.compact_context")),
)
}
fn reset_conversation<'a>(
&'a mut self,
keep_plan: bool,
no_digest: bool,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
Box::pin(async move {
match self.reset_conversation(keep_plan, no_digest).await {
Ok((old_id, new_id)) => {
let old = old_id.map_or_else(|| "none".to_string(), |id| id.0.to_string());
let new = new_id.map_or_else(|| "none".to_string(), |id| id.0.to_string());
let keep_note = if keep_plan { " (plan preserved)" } else { "" };
Ok(format!(
"New conversation started. Previous: {old} → Current: {new}{keep_note}"
))
}
Err(e) => Ok(format!("Failed to start new conversation: {e}")),
}
})
}
fn cache_stats(&self) -> String {
self.tool_orchestrator.cache_stats()
}
fn session_status<'a>(
&'a mut self,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
Box::pin(async move { Ok(self.handle_status_as_string()) })
}
fn guardrail_status(&self) -> String {
self.format_guardrail_status()
}
fn focus_status(&self) -> String {
self.format_focus_status()
}
fn sidequest_status(&self) -> String {
self.format_sidequest_status()
}
fn load_image<'a>(
&'a mut self,
path: &'a str,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
use zeph_common::path_guard::{PathRejection, classify_relative_path};
use zeph_llm::provider::{ImageData, MessagePart};
match classify_relative_path(path) {
PathRejection::Allowed => {}
PathRejection::Absolute => {
return Box::pin(async move {
Ok(
"Invalid image path: absolute paths are not supported, use a path \
relative to the working directory"
.to_owned(),
)
});
}
PathRejection::Traversal => {
return Box::pin(async move {
Ok("Invalid image path: path traversal ('..') is not allowed".to_owned())
});
}
}
let path_owned = path.to_owned();
Box::pin(async move {
let path_for_task = path_owned.clone();
let read_result = tokio::task::spawn_blocking(move || std::fs::read(&path_for_task))
.await
.map_err(|e| CommandError::new(format!("spawn_blocking join error: {e}")))?;
let data = match read_result {
Ok(d) => d,
Err(e) => return Ok(format!("Cannot read image {path_owned}: {e}")),
};
if data.len() > crate::agent::message_queue::MAX_IMAGE_BYTES {
return Ok(format!(
"Image {path_owned} exceeds size limit ({} MB), skipping",
crate::agent::message_queue::MAX_IMAGE_BYTES / 1024 / 1024
));
}
let mime_type =
crate::agent::message_queue::detect_image_mime(Some(&path_owned)).to_string();
self.msg
.pending_image_parts
.push(MessagePart::Image(Box::new(ImageData { data, mime_type })));
Ok(format!("Image loaded: {path_owned}. Send your message."))
})
}
fn handle_undo<'a>(
&'a mut self,
args: &'a str,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
let executor = std::sync::Arc::clone(&self.tool_executor);
let args_owned = args.trim().to_owned();
Box::pin(async move {
if args_owned == "list" {
let result = executor.checkpoint_list_erased();
if !result.supported {
return Ok(
"Checkpoints are not enabled. Set `[tools.shell] checkpoints_enabled = true` in config.".to_owned()
);
}
if result.entries.is_empty() {
return Ok("Undo stack is empty.".to_owned());
}
let mut lines = vec![format!("Undo stack ({} entries):", result.entries.len())];
for e in &result.entries {
lines.push(format!(
" [{}] {} ({} file(s))",
e.index, e.command, e.file_count
));
}
if result.redo_depth > 0 {
lines.push(format!("Redo depth: {}", result.redo_depth));
}
return Ok(lines.join("\n"));
}
let n: usize = if args_owned.is_empty() {
1
} else {
match args_owned.parse::<usize>() {
Ok(v) if v > 0 => v,
_ => {
return Err(CommandError::new(format!(
"Invalid argument: expected a positive integer or 'list', got '{args_owned}'"
)));
}
}
};
let result = tokio::task::spawn_blocking(move || executor.checkpoint_undo_erased(n))
.await
.map_err(|e| CommandError::new(format!("undo task panicked: {e}")))?;
if !result.supported {
return Ok(
"Checkpoints are not enabled. Set `[tools.shell] checkpoints_enabled = true` in config.".to_owned()
);
}
Ok(result.message)
})
}
fn handle_redo<'a>(
&'a mut self,
args: &'a str,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
let _ = args;
let executor = std::sync::Arc::clone(&self.tool_executor);
Box::pin(async move {
let result = tokio::task::spawn_blocking(move || executor.checkpoint_redo_erased())
.await
.map_err(|e| CommandError::new(format!("redo task panicked: {e}")))?;
if !result.supported {
return Ok(
"Checkpoints are not enabled. Set `[tools.shell] checkpoints_enabled = true` in config.".to_owned()
);
}
Ok(result.message)
})
}
fn handle_conv<'a>(
&'a mut self,
args: &'a str,
) -> Pin<Box<dyn Future<Output = Result<String, CommandError>> + Send + 'a>> {
let args_owned = args.trim().to_owned();
Box::pin(async move {
if let Some(id) = args_owned.strip_prefix("resume ") {
return self.handle_conv_resume(id.trim()).await;
}
if let Some(id) = args_owned.strip_prefix("fork ") {
return self.handle_conv_fork(id.trim()).await;
}
let Some(memory) = self.services.memory.persistence.memory.clone() else {
return Ok(
"Conversation-session persistence requires memory to be enabled ([memory] enabled = true)."
.to_owned(),
);
};
let store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
if let Some(id) = args_owned.strip_prefix("show ") {
return handle_conv_show(&store, id.trim()).await;
}
if args_owned.is_empty() || args_owned == "list" {
return handle_conv_list(&store).await;
}
Ok(format!(
"Unknown /conv subcommand '{args_owned}'. Usage: /conv [list | show <id> | resume <id> | fork <id>]"
))
})
}
}
#[cfg(test)]
mod tests {
use super::super::agent_tests::{
MockChannel, MockToolExecutor, create_test_registry, mock_provider,
};
use super::*;
use zeph_memory::semantic::SemanticMemory;
async fn memory_without_qdrant() -> SemanticMemory {
SemanticMemory::new(
":memory:",
"http://127.0.0.1:1",
None,
zeph_llm::any::AnyProvider::Mock(zeph_llm::mock::MockProvider::default()),
"test-model",
)
.await
.unwrap()
}
#[test]
fn subagent_is_excluded_from_is_recognized_command() {
assert!(!zeph_commands::is_recognized_command("/subagent"));
assert!(!zeph_commands::is_recognized_command(
"/subagent spawn zeph --acp"
));
}
#[test]
fn format_overlay_section_empty_dir() {
let tmp = tempfile::tempdir().unwrap();
let out = format_overlay_section(tmp.path());
assert_eq!(out, "No plugin overlay active.");
}
#[test]
fn format_overlay_section_with_source_plugin() {
let tmp = tempfile::tempdir().unwrap();
let plugin_dir = tmp.path().join("myplugin");
std::fs::create_dir_all(&plugin_dir).unwrap();
let manifest = r#"
[plugin]
name = "myplugin"
version = "0.1.0"
description = "test"
[config.tools.shell]
blocked_commands = ["curl"]
"#;
std::fs::write(plugin_dir.join(".plugin.toml"), manifest).unwrap();
let out = format_overlay_section(tmp.path());
assert!(out.contains("Active plugin overlay:"));
assert!(out.contains("myplugin"));
assert!(out.contains("Source plugins:"));
assert!(out.contains("Note:"));
}
#[test]
fn run_plugin_command_overlay_subcommand() {
let tmp = tempfile::tempdir().unwrap();
let out = format_overlay_section(tmp.path());
assert_eq!(out, "No plugin overlay active.");
}
#[test]
fn format_overlay_section_skipped_plugin_shows_reason() {
let tmp = tempfile::tempdir().unwrap();
let plugin_dir = tmp.path().join("badplugin");
std::fs::create_dir_all(&plugin_dir).unwrap();
std::fs::write(
plugin_dir.join(".plugin.toml"),
b"not valid toml at all {{{{",
)
.unwrap();
let out = format_overlay_section(tmp.path());
assert!(out.contains("No plugin overlay active.") || out.contains("badplugin"));
}
#[tokio::test]
async fn handle_conv_resume_same_session_short_circuits() {
let memory = memory_without_qdrant().await;
let cid = memory.sqlite().create_conversation().await.unwrap();
let dir = tempfile::tempdir().unwrap();
let data_dir = dir.path().to_path_buf();
let session_id = zeph_common::SessionId::new("s1");
let session_path = zeph_session::session_dir(&data_dir, session_id.as_str());
let log = zeph_session::SessionEventLog::open_exclusive(&session_path)
.await
.unwrap();
let store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
let sink = zeph_agent_persistence::SessionSink::new(
std::sync::Arc::new(log),
store,
session_id.clone(),
);
let session_config = zeph_config::SessionConfig {
enabled: true,
data_dir: data_dir.to_string_lossy().into_owned(),
..Default::default()
};
let mut agent = Agent::new(
mock_provider(vec![]),
MockChannel::new(vec![]),
create_test_registry(),
None,
5,
MockToolExecutor::no_tools(),
)
.with_memory(std::sync::Arc::new(memory), cid, 50, 5, 100)
.with_session_sink(Some(std::sync::Arc::new(sink)))
.with_session_persistence_config(Some(session_config));
let result = agent.handle_conv("resume s1").await.unwrap();
assert_eq!(
result, "Already in session 's1'.",
"resuming into the currently-active session must short-circuit, not attempt \
hydration/lock acquisition"
);
}
#[tokio::test]
async fn handle_conv_resume_different_session_still_hydrates() {
let memory = memory_without_qdrant().await;
let cid = memory.sqlite().create_conversation().await.unwrap();
let dir = tempfile::tempdir().unwrap();
let data_dir = dir.path().to_path_buf();
let active_session_id = zeph_common::SessionId::new("s1");
let active_session_path = zeph_session::session_dir(&data_dir, active_session_id.as_str());
let active_log = zeph_session::SessionEventLog::open_exclusive(&active_session_path)
.await
.unwrap();
let active_store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
let active_sink = zeph_agent_persistence::SessionSink::new(
std::sync::Arc::new(active_log),
active_store,
active_session_id,
);
let store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
store.create("s2").await.unwrap();
let session_config = zeph_config::SessionConfig {
enabled: true,
data_dir: data_dir.to_string_lossy().into_owned(),
..Default::default()
};
let mut agent = Agent::new(
mock_provider(vec![]),
MockChannel::new(vec![]),
create_test_registry(),
None,
5,
MockToolExecutor::no_tools(),
)
.with_memory(std::sync::Arc::new(memory), cid, 50, 5, 100)
.with_session_sink(Some(std::sync::Arc::new(active_sink)))
.with_session_persistence_config(Some(session_config));
let result = agent.handle_conv("resume s2").await.unwrap();
assert!(
result.starts_with("Resumed session s2"),
"resuming into a different, unlocked session must still hydrate normally, got: {result}"
);
}
#[tokio::test]
async fn handle_conv_fork_creates_child_session_and_switches_to_it() {
let memory = memory_without_qdrant().await;
let cid = memory.sqlite().create_conversation().await.unwrap();
let dir = tempfile::tempdir().unwrap();
let data_dir = dir.path().to_path_buf();
let store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
store.create("s1").await.unwrap();
let src_dir = zeph_session::session_dir(&data_dir, "s1");
let log = zeph_session::SessionEventLog::open(&src_dir).await.unwrap();
log.append(
None,
None,
zeph_session::SessionEvent::SessionStarted {
session_id: "s1".to_owned(),
cwd: "/repo".to_owned(),
provider_name: "claude".to_owned(),
model: "opus".to_owned(),
forked_from: None,
},
)
.await
.unwrap();
store
.update_seq("s1", log.last_seq().unwrap(), 1)
.await
.unwrap();
drop(log);
let session_config = zeph_config::SessionConfig {
enabled: true,
data_dir: data_dir.to_string_lossy().into_owned(),
..Default::default()
};
let mut agent = Agent::new(
mock_provider(vec![]),
MockChannel::new(vec![]),
create_test_registry(),
None,
5,
MockToolExecutor::no_tools(),
)
.with_memory(std::sync::Arc::new(memory), cid, 50, 5, 100)
.with_session_persistence_config(Some(session_config));
let result = agent.handle_conv("fork s1").await.unwrap();
assert!(
result.starts_with("Forked session s1 ->"),
"expected fork confirmation message, got: {result}"
);
assert!(
result.contains("event(s) copied"),
"expected copied-event count in confirmation, got: {result}"
);
}
#[tokio::test]
async fn handle_conv_fork_sends_resume_banner_for_non_empty_history() {
let memory = memory_without_qdrant().await;
let cid = memory.sqlite().create_conversation().await.unwrap();
let dir = tempfile::tempdir().unwrap();
let data_dir = dir.path().to_path_buf();
let store = zeph_session::SessionStore::new(memory.sqlite().pool().clone());
store.create("s1").await.unwrap();
let src_dir = zeph_session::session_dir(&data_dir, "s1");
let log = zeph_session::SessionEventLog::open(&src_dir).await.unwrap();
log.append(
None,
None,
zeph_session::SessionEvent::SessionStarted {
session_id: "s1".to_owned(),
cwd: "/repo".to_owned(),
provider_name: "claude".to_owned(),
model: "opus".to_owned(),
forked_from: None,
},
)
.await
.unwrap();
log.append(
None,
None,
zeph_session::SessionEvent::UserMessage {
text: "hello".to_owned(),
image_refs: vec![],
},
)
.await
.unwrap();
store
.update_seq("s1", log.last_seq().unwrap(), 2)
.await
.unwrap();
drop(log);
let session_config = zeph_config::SessionConfig {
enabled: true,
data_dir: data_dir.to_string_lossy().into_owned(),
..Default::default()
};
let mut agent = Agent::new(
mock_provider(vec![]),
MockChannel::new(vec![]),
create_test_registry(),
None,
5,
MockToolExecutor::no_tools(),
)
.with_memory(std::sync::Arc::new(memory), cid, 50, 5, 100)
.with_session_persistence_config(Some(session_config));
let result = agent.handle_conv("fork s1").await.unwrap();
assert!(
result.starts_with("Forked session s1 ->"),
"expected fork confirmation message, got: {result}"
);
let sent = agent.channel.sent_messages();
assert!(
sent.iter().any(|m| m.contains("Resuming session")),
"forking a session with non-empty prior history must send the resume banner \
through the channel, got sent messages: {sent:?}"
);
}
#[tokio::test]
async fn load_image_rejects_absolute_path() {
let mut agent = Agent::new(
mock_provider(vec![]),
MockChannel::new(vec![]),
create_test_registry(),
None,
5,
MockToolExecutor::no_tools(),
);
let result = SessionControlAccess::load_image(&mut agent, "/etc/passwd")
.await
.unwrap();
assert!(result.contains("absolute paths are not supported"));
}
#[tokio::test]
async fn load_image_rejects_parent_dir_traversal() {
let mut agent = Agent::new(
mock_provider(vec![]),
MockChannel::new(vec![]),
create_test_registry(),
None,
5,
MockToolExecutor::no_tools(),
);
let result = SessionControlAccess::load_image(&mut agent, "../../etc/passwd")
.await
.unwrap();
assert!(result.contains("path traversal") && result.contains("not allowed"));
}
}