use std::collections::HashSet;
use crate::config::{CachePlan, Config, SteeringMode, ToolAdvertising};
use crate::error::{Error, Result};
use crate::event::AgentEvent;
use crate::message::{ChatMessage, Role};
use crate::provider::{self, ChatRequest, OpenAiProvider, Provider, ToolSchema};
use crate::reduce::rehydrate::CAP_NOTICE_MARKER;
use crate::reduce::{self, ReductionLog, ReductionPolicy};
use crate::session::Session;
use crate::sidecar::SidecarWriter;
use crate::tools::{ToolContext, ToolRegistry};
const TOOL_SEARCH: &str = "tool_search";
const EXPAND_REDUCTION: &str = "expand_reduction";
const SIDECAR_SEARCH: &str = "sidecar_search";
const SPAWN_SUBAGENT: &str = "spawn_subagent";
const CLAUDE_AGENT: &str = "Agent";
const CLAUDE_BASH: &str = "Bash";
const CLAUDE_READ: &str = "Read";
const CLAUDE_WRITE: &str = "Write";
const CLAUDE_EDIT: &str = "Edit";
const CLAUDE_GLOB: &str = "Glob";
const CLAUDE_GREP: &str = "Grep";
const CLAUDE_CRON_CREATE: &str = "CronCreate";
const CLAUDE_CRON_DELETE: &str = "CronDelete";
const CLAUDE_CRON_LIST: &str = "CronList";
const CLAUDE_SCHEDULE_WAKEUP: &str = "ScheduleWakeup";
#[derive(Default)]
pub(crate) struct SteerInbox {
queue: std::collections::VecDeque<QueuedSteer>,
accepting: bool,
}
struct QueuedSteer {
message: String,
sdk_bound: bool,
}
impl SteerInbox {
pub(crate) fn open(&mut self) {
self.queue.clear();
self.accepting = true;
}
pub(crate) fn enqueue(&mut self, message: String) -> bool {
if !self.accepting {
return false;
}
self.queue.push_back(QueuedSteer {
message,
sdk_bound: true,
});
true
}
pub(crate) fn close(&mut self) {
self.accepting = false;
self.queue.retain(|queued| !queued.sdk_bound);
}
fn drain(&mut self, mode: SteeringMode) -> Option<String> {
if self.queue.is_empty() {
return None;
}
match mode {
SteeringMode::All => Some(
self.queue
.drain(..)
.map(|queued| queued.message)
.collect::<Vec<_>>()
.join("\n\n"),
),
SteeringMode::OneAtATime => self.queue.pop_front().map(|queued| queued.message),
}
}
fn drain_or_close(&mut self, mode: SteeringMode) -> Option<String> {
if !self.queue.iter().any(|queued| queued.sdk_bound) {
self.accepting = false;
return None;
}
self.drain(mode)
}
fn queue_unchecked(&mut self, message: String) {
self.queue.push_back(QueuedSteer {
message,
sdk_bound: false,
});
}
fn len(&self) -> usize {
self.queue.len()
}
}
struct SteerTurnGuard {
inbox: std::sync::Arc<std::sync::Mutex<SteerInbox>>,
}
impl SteerTurnGuard {
fn new(inbox: std::sync::Arc<std::sync::Mutex<SteerInbox>>) -> Self {
inbox
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.accepting = true;
Self { inbox }
}
}
impl Drop for SteerTurnGuard {
fn drop(&mut self) {
self.inbox
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.close();
}
}
const SUBAGENT_STATUS: &str = "subagent_status";
const BACKGROUND_EXEC: &str = "background_exec";
const BACKGROUND_STATUS: &str = "background_status";
const BACKGROUND_LIST: &str = "background_list";
const BACKGROUND_KILL: &str = "background_kill";
enum PreparedCall {
Done((String, bool)),
Ready {
name: String,
args: serde_json::Value,
},
}
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
static SUBAGENT_ID_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
fn next_subagent_id() -> String {
let seq = SUBAGENT_ID_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
format!("agent-{:x}-{:x}", now_ms(), seq)
}
type ChildApprovalHandlerFactory = dyn Fn(
String,
std::sync::Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>,
) -> std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler>
+ Send
+ Sync;
pub struct Agent {
config: Config,
provider: std::sync::Arc<dyn Provider>,
registry: ToolRegistry,
history: Vec<ChatMessage>,
ctx: ToolContext,
total_output_tokens: u64,
activated_tools: HashSet<String>,
recorder: Option<SidecarWriter>,
reduction_policy: Option<ReductionPolicy>,
reduction_log: ReductionLog,
imported_prefix_len: Option<usize>,
span_summarizer: Option<std::sync::Arc<dyn reduce::summarize::SpanSummarizer + Send + Sync>>,
last_tool_schema_tier_signature: Option<u64>,
context_limit: Option<u64>,
requests_issued: bool,
last_cache_activity_ms: Option<i64>,
cache_established: bool,
pending_cache_turn: (bool, bool, Option<i64>),
session_titler: Option<std::sync::Arc<dyn crate::session_title::SessionTitler + Send + Sync>>,
usage_log: Vec<crate::usage_log::UsageRecord>,
turn_index: usize,
steer_queue: std::sync::Arc<std::sync::Mutex<SteerInbox>>,
follow_up_queue: std::collections::VecDeque<String>,
doom_loop_last_call: Option<(String, String)>,
doom_loop_streak: u32,
model_change_log: Vec<crate::model_change::ModelChangeRecord>,
git_metadata: Option<crate::git_metadata::GitMetadataRecord>,
permissions_approval_cache: crate::permissions::ApprovalCache,
permissions_approval_handler:
Option<std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler>>,
mcp_prompts: std::collections::HashMap<String, Box<dyn crate::sdk::SdkPromptSource>>,
subagent_depth: usize,
subagent_concurrency_gauge: std::sync::Arc<std::sync::atomic::AtomicUsize>,
background_subagents: std::collections::HashMap<String, BackgroundSubagent>,
pending_child_approvals:
std::sync::Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>,
child_approval_handler_factory: Option<std::sync::Arc<ChildApprovalHandlerFactory>>,
subagent_store: Option<(std::sync::Arc<crate::store::SessionStore>, String)>,
claude_runtime_manifest: Option<crate::claude_runtime_state::ClaudeRuntimeManifest>,
background_jobs: std::collections::HashMap<String, BackgroundJob>,
background_concurrency_gauge: std::sync::Arc<std::sync::atomic::AtomicUsize>,
checkpoint_observer: Option<std::sync::Arc<crate::checkpoint::CheckpointObserver>>,
lsp_manager: Option<std::sync::Arc<crate::lsp::LspManager>>,
}
struct BackgroundJob {
child: tokio::process::Child,
command: String,
pid: Option<u32>,
output: std::sync::Arc<crate::background::CapturedOutput>,
started_at_ms: i64,
killed: bool,
_guard: crate::subagents::ConcurrencyGuard,
}
fn background_job_status(job: &mut BackgroundJob) -> crate::background::JobStatus {
if job.killed {
return crate::background::JobStatus::Killed;
}
match job.child.try_wait() {
Ok(Some(status)) => crate::background::JobStatus::Exited(status.code()),
Ok(None) | Err(_) => crate::background::JobStatus::Running,
}
}
#[cfg(unix)]
fn kill_job_process_group(job: &mut BackgroundJob) {
if let Some(pid) = job.pid {
unsafe {
libc::kill(-(pid as libc::pid_t), libc::SIGKILL);
}
}
let _ = job.child.start_kill();
let _ = job.child.try_wait();
}
#[cfg(not(unix))]
fn kill_job_process_group(job: &mut BackgroundJob) {
let _ = job.child.start_kill();
}
fn spawn_output_reader<R>(
reader: R,
output: std::sync::Arc<crate::background::CapturedOutput>,
cap: usize,
) -> tokio::task::JoinHandle<()>
where
R: tokio::io::AsyncRead + Unpin + Send + 'static,
{
tokio::spawn(async move {
use tokio::io::AsyncReadExt;
let mut reader = reader;
let mut buf = [0u8; 8192];
loop {
match reader.read(&mut buf).await {
Ok(0) => break,
Ok(n) => {
let chunk = String::from_utf8_lossy(&buf[..n]);
output.append(&chunk, cap);
}
Err(_) => break,
}
}
})
}
struct BackgroundSubagent {
handle: tokio::task::JoinHandle<(String, Result<String>, Vec<ChatMessage>)>,
task: String,
agent_type: Option<String>,
started_at_ms: i64,
}
impl Drop for Agent {
fn drop(&mut self) {
for (child_id, bg) in self.background_subagents.drain() {
if !bg.handle.is_finished() {
tracing::debug!(
child_id = %child_id,
"parent Agent dropped: aborting still-running background subagent \
to stop further provider spend"
);
}
bg.handle.abort();
}
for (job_id, mut job) in self.background_jobs.drain() {
if !job.killed {
tracing::debug!(
job_id = %job_id,
command = %job.command,
"parent Agent dropped: killing still-tracked background job's real \
OS process (and its whole process group — see \
`kill_job_process_group`)"
);
}
kill_job_process_group(&mut job);
}
if let Some(lsp) = &self.lsp_manager {
lsp.kill_all_sync();
}
}
}
fn run_api_key_cmd(cmd: &str) -> String {
match std::process::Command::new("sh").arg("-c").arg(cmd).output() {
Ok(out) if out.status.success() => String::from_utf8_lossy(&out.stdout).trim().to_string(),
Ok(out) => {
tracing::warn!(
"api_key_cmd exited with status {:?}; falling back to api_key_env",
out.status.code()
);
String::new()
}
Err(e) => {
tracing::warn!("api_key_cmd failed to run ({e}); falling back to api_key_env");
String::new()
}
}
}
fn capture_shell_env() -> std::collections::HashMap<String, String> {
let shell = std::env::var("SHELL").unwrap_or_else(|_| "sh".to_string());
let out = match std::process::Command::new(&shell)
.arg("-lc")
.arg("env")
.output()
{
Ok(o) if o.status.success() => o.stdout,
Ok(o) => {
tracing::warn!(
"shell_env_snapshot: `{shell} -lc env` exited with status {:?}; snapshot is empty",
o.status.code()
);
return std::collections::HashMap::new();
}
Err(e) => {
tracing::warn!(
"shell_env_snapshot: failed to run `{shell} -lc env` ({e}); snapshot is empty"
);
return std::collections::HashMap::new();
}
};
let text = String::from_utf8_lossy(&out);
let mut map = std::collections::HashMap::new();
for line in text.lines() {
if let Some((k, v)) = line.split_once('=') {
if !k.is_empty() {
map.insert(k.to_string(), v.to_string());
}
}
}
map
}
fn build_tool_context(
config: &Config,
) -> (
ToolContext,
Option<std::sync::Arc<crate::checkpoint::CheckpointObserver>>,
Option<std::sync::Arc<crate::lsp::LspManager>>,
) {
let shell_env = if config.shell_env_snapshot {
Some(std::sync::Arc::new(capture_shell_env()))
} else {
None
};
let checkpoint_observer = crate::checkpoint::observer_for_config(config);
let format_observer = crate::formatters::observer_for_config(config);
let lsp_manager = crate::lsp::manager_for_config(config);
let lsp_observer = lsp_manager
.clone()
.map(|m| std::sync::Arc::new(crate::lsp::LspDiagnosticsObserver::new(m)));
let mut observers: Vec<std::sync::Arc<dyn crate::tools::WriteObserver>> = Vec::new();
if let Some(cp) = &checkpoint_observer {
observers.push(cp.clone() as std::sync::Arc<dyn crate::tools::WriteObserver>);
}
if let Some(f) = &format_observer {
observers.push(f.clone() as std::sync::Arc<dyn crate::tools::WriteObserver>);
}
if let Some(l) = &lsp_observer {
observers.push(l.clone() as std::sync::Arc<dyn crate::tools::WriteObserver>);
}
let write_observer: Option<std::sync::Arc<dyn crate::tools::WriteObserver>> =
match observers.len() {
0 => None,
1 => observers.into_iter().next(),
_ => Some(std::sync::Arc::new(crate::tools::WriteObserverChain::new(
observers,
))),
};
let ctx = ToolContext {
cwd: config.cwd.clone(),
sandbox: config.sandbox,
multimodal_read: config.read_file_multimodal,
require_read_before_edit: config.edit_file_require_read_before_edit,
read_paths: std::sync::Arc::new(std::sync::Mutex::new(HashSet::new())),
notebook_aware: config.edit_file_notebook_aware,
shell_env,
nested_instructions: config.nested_instructions,
injected_instruction_dirs: std::sync::Arc::new(std::sync::Mutex::new(HashSet::new())),
network_policy: config.network_policy.clone(),
bash_timeout_secs: config
.tool_overrides
.get("bash")
.and_then(|o| o.timeout_secs),
write_observer,
sandbox_os_enabled: config.sandbox_os_enabled,
sandbox_escalation: config.sandbox_escalation,
sandbox_env_policy: config.sandbox_env_policy,
sandbox_approval_handler: None,
};
(ctx, checkpoint_observer, lsp_manager)
}
pub(crate) fn global_instructions_dir() -> std::path::PathBuf {
if let Ok(h) = std::env::var("SUPERCODE_HOME") {
if !h.is_empty() {
return std::path::PathBuf::from(h);
}
}
if let Ok(xdg) = std::env::var("XDG_CONFIG_HOME") {
if !xdg.is_empty() {
return std::path::PathBuf::from(xdg).join("supercode");
}
}
let home = std::env::var("HOME").unwrap_or_else(|_| ".".into());
std::path::PathBuf::from(home)
.join(".config")
.join("supercode")
}
fn import_path_is_safe(rel: &str) -> bool {
if rel.is_empty() || rel.contains('\0') {
return false;
}
let path = std::path::Path::new(rel);
if path.is_absolute() || rel.starts_with('~') {
return false;
}
!path
.components()
.any(|c| matches!(c, std::path::Component::ParentDir))
}
pub(crate) fn import_target_is_contained(
candidate: &std::path::Path,
root: &std::path::Path,
) -> bool {
let (Ok(real_root), Ok(real_candidate)) = (
std::fs::canonicalize(root),
std::fs::canonicalize(candidate),
) else {
return false;
};
real_candidate.starts_with(&real_root)
}
fn expand_instruction_imports(
text: &str,
dir: &std::path::Path,
root: &std::path::Path,
project_scoped: bool,
depth: u8,
) -> String {
if depth >= 4 {
return text.to_string();
}
let mut out = String::with_capacity(text.len());
for token in split_preserving_whitespace(text) {
if let Some(rel) = token.strip_prefix('@') {
if !rel.is_empty()
&& !rel.contains(char::is_whitespace)
&& (!project_scoped || import_path_is_safe(rel))
{
let candidate = dir.join(rel);
if !project_scoped || import_target_is_contained(&candidate, root) {
if let Ok(imported) = std::fs::read_to_string(&candidate) {
let imported = imported.trim();
if !imported.is_empty() {
let imported_dir = candidate.parent().unwrap_or(dir);
out.push_str(&expand_instruction_imports(
imported,
imported_dir,
root,
project_scoped,
depth + 1,
));
continue;
}
}
}
}
}
out.push_str(token);
}
out
}
fn split_preserving_whitespace(text: &str) -> Vec<&str> {
let mut out = Vec::new();
let mut start = 0;
let mut in_ws = None;
for (i, c) in text.char_indices() {
let ws = c.is_whitespace();
match in_ws {
None => in_ws = Some(ws),
Some(prev) if prev != ws => {
out.push(&text[start..i]);
start = i;
in_ws = Some(ws);
}
_ => {}
}
}
if start < text.len() {
out.push(&text[start..]);
}
out
}
fn append_instruction_file(
blob: &mut String,
path: &std::path::Path,
root: &std::path::Path,
label: &str,
imports_enabled: bool,
project_scoped: bool,
) {
let Ok(text) = std::fs::read_to_string(path) else {
return;
};
let text = text.trim();
if text.is_empty() {
return;
}
let dir = path.parent().unwrap_or(std::path::Path::new("."));
let content = if imports_enabled {
expand_instruction_imports(text, dir, root, project_scoped, 0)
} else {
text.to_string()
};
blob.push_str(&format!("\n\n# {label}\n{content}"));
}
fn assemble_project_instructions(config: &Config) -> String {
let mut blob = String::new();
let global_dir = global_instructions_dir();
for name in ["CLAUDE.md", "AGENTS.md"] {
append_instruction_file(
&mut blob,
&global_dir.join(name),
&global_dir,
name,
config.instruction_imports,
false,
);
}
for root in std::iter::once(&config.cwd).chain(config.additional_dirs.iter()) {
for name in ["CLAUDE.md", "AGENTS.md"] {
append_instruction_file(
&mut blob,
&root.join(name),
root,
name,
config.instruction_imports,
true,
);
}
}
if let Some(max) = config.project_doc_max_bytes {
if blob.len() > max {
let mut end = max;
while end > 0 && !blob.is_char_boundary(end) {
end -= 1;
}
blob.truncate(end);
blob.push_str(
"\n\n[supercode: instruction content truncated at core.project_doc_max_bytes]",
);
}
}
blob
}
fn env_context_block(config: &Config) -> String {
let mut lines = vec![
format!("cwd: {}", config.cwd.display()),
format!("platform: {}", std::env::consts::OS),
format!(
"date: {}",
crate::sidecar::now_rfc3339().get(..10).unwrap_or("")
),
];
if let Some(status) = env_context_git_status(&config.cwd) {
lines.push(status);
}
format!("\n\n# Environment\n{}", lines.join("\n"))
}
fn env_context_git_status(cwd: &std::path::Path) -> Option<String> {
let branch_out = std::process::Command::new("git")
.args(["rev-parse", "--abbrev-ref", "HEAD"])
.current_dir(cwd)
.output()
.ok()?;
if !branch_out.status.success() {
return None;
}
let branch = String::from_utf8_lossy(&branch_out.stdout)
.trim()
.to_string();
if branch.is_empty() {
return None;
}
let dirty = std::process::Command::new("git")
.args(["status", "--porcelain"])
.current_dir(cwd)
.output()
.ok()
.map(|o| !o.stdout.is_empty())
.unwrap_or(false);
Some(format!(
"git branch: {branch} ({})",
if dirty { "dirty" } else { "clean" }
))
}
impl Agent {
pub fn new(config: Config) -> Result<Self> {
let api_key = match &config.api_key {
Some(k) if !k.is_empty() => k.clone(),
_ => match config
.api_key_cmd
.as_deref()
.filter(|c| !c.is_empty())
.map(run_api_key_cmd)
{
Some(k) if !k.is_empty() => k,
_ => std::env::var(&config.api_key_env)
.ok()
.filter(|k| !k.is_empty())
.ok_or_else(|| Error::MissingApiKey(config.api_key_env.clone()))?,
},
};
let http_options = provider::HttpOptions::from_retry_config(
config.retry_enabled,
config.retry_max_retries,
config.retry_base_delay_ms,
);
let provider = OpenAiProvider::new_with_options(
config.base_url.clone(),
api_key,
config.extra_headers.clone(),
http_options,
);
let registry = ToolRegistry::from_config(&config);
Ok(Self::with_parts(config, Box::new(provider), registry))
}
pub fn with_provider(config: Config, provider: Box<dyn Provider>) -> Self {
let registry = ToolRegistry::from_config(&config);
Self::with_parts(config, provider, registry)
}
pub fn with_parts(
config: Config,
provider: Box<dyn Provider>,
mut registry: ToolRegistry,
) -> Self {
crate::plugins::register_into(&config, &mut registry);
let (ctx, checkpoint_observer, lsp_manager) = build_tool_context(&config);
let mut system = config.system_prompt.clone();
if config.load_project_context {
system.push_str(&assemble_project_instructions(&config));
}
if config.env_context {
system.push_str(&env_context_block(&config));
}
if config.context_injections {
for block in &config.context_injection_blocks {
system.push_str(&format!("\n\n# {}\n{}", block.name, block.content));
}
}
if config.module_registry && config.skills_enabled {
let has_read_pathway = config
.core_tools_enabled
.iter()
.any(|t| t == "read_file" || t == "bash");
if has_read_pathway {
let mut names: Vec<&str> = config.prompts.keys().map(String::as_str).collect();
names.sort_unstable();
if !names.is_empty() {
system.push_str("\n\n# Skills\nAvailable skill/prompt templates (today: named `[core.prompts]` templates, D-7 — invoke via `/name args`):\n");
for name in names {
system.push_str(&format!("- {name}\n"));
}
}
}
}
let history = vec![ChatMessage::system(system)];
let git_metadata = if config.session_git_metadata {
crate::git_metadata::capture(&config.cwd, now_ms())
} else {
None
};
let subagent_depth = config.subagent_depth;
Agent {
config,
provider: std::sync::Arc::from(provider),
registry,
history,
ctx,
total_output_tokens: 0,
activated_tools: HashSet::new(),
recorder: None,
reduction_policy: None,
reduction_log: ReductionLog::default(),
imported_prefix_len: None,
span_summarizer: None,
last_tool_schema_tier_signature: None,
context_limit: None,
requests_issued: false,
last_cache_activity_ms: None,
cache_established: false,
pending_cache_turn: (false, false, None),
session_titler: None,
usage_log: Vec::new(),
turn_index: 0,
steer_queue: std::sync::Arc::new(std::sync::Mutex::new(SteerInbox::default())),
follow_up_queue: std::collections::VecDeque::new(),
doom_loop_last_call: None,
doom_loop_streak: 0,
model_change_log: Vec::new(),
git_metadata,
permissions_approval_cache: crate::permissions::ApprovalCache::new(),
permissions_approval_handler: None,
mcp_prompts: std::collections::HashMap::new(),
subagent_depth,
subagent_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
background_subagents: std::collections::HashMap::new(),
pending_child_approvals: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
child_approval_handler_factory: None,
subagent_store: None,
claude_runtime_manifest: None,
background_jobs: std::collections::HashMap::new(),
background_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(
0,
)),
checkpoint_observer,
lsp_manager,
}
}
pub fn provider_arc(&self) -> std::sync::Arc<dyn Provider> {
self.provider.clone()
}
pub fn config(&self) -> &Config {
&self.config
}
pub fn checkpoint_observer(&self) -> Option<&crate::checkpoint::CheckpointObserver> {
self.checkpoint_observer.as_deref()
}
pub fn lsp_manager(&self) -> Option<&crate::lsp::LspManager> {
self.lsp_manager.as_deref()
}
pub async fn run_subagent(
&self,
system: impl Into<String>,
task: impl Into<String>,
) -> Result<String> {
let mut sub_config = Config::builder()
.model(self.config.model.clone())
.system_prompt(system)
.cwd(self.config.cwd.clone())
.sandbox(self.config.sandbox)
.max_iterations(self.config.max_iterations)
.build();
sub_config.base_url = self.config.base_url.clone();
let mut sub = Agent::with_provider_arc(sub_config, self.provider.clone());
sub.send(task).await
}
pub fn with_provider_arc(config: Config, provider: std::sync::Arc<dyn Provider>) -> Self {
let (ctx, checkpoint_observer, lsp_manager) = build_tool_context(&config);
let history = vec![ChatMessage::system(config.system_prompt.clone())];
let mut registry = ToolRegistry::from_config(&config);
crate::plugins::register_into(&config, &mut registry);
let git_metadata = if config.session_git_metadata {
crate::git_metadata::capture(&config.cwd, now_ms())
} else {
None
};
let subagent_depth = config.subagent_depth;
Agent {
config,
provider,
registry,
history,
ctx,
total_output_tokens: 0,
activated_tools: HashSet::new(),
recorder: None,
reduction_policy: None,
reduction_log: ReductionLog::default(),
imported_prefix_len: None,
span_summarizer: None,
last_tool_schema_tier_signature: None,
context_limit: None,
requests_issued: false,
last_cache_activity_ms: None,
cache_established: false,
pending_cache_turn: (false, false, None),
session_titler: None,
usage_log: Vec::new(),
turn_index: 0,
steer_queue: std::sync::Arc::new(std::sync::Mutex::new(SteerInbox::default())),
follow_up_queue: std::collections::VecDeque::new(),
doom_loop_last_call: None,
doom_loop_streak: 0,
model_change_log: Vec::new(),
git_metadata,
permissions_approval_cache: crate::permissions::ApprovalCache::new(),
permissions_approval_handler: None,
mcp_prompts: std::collections::HashMap::new(),
subagent_depth,
subagent_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
background_subagents: std::collections::HashMap::new(),
pending_child_approvals: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
child_approval_handler_factory: None,
subagent_store: None,
claude_runtime_manifest: None,
background_jobs: std::collections::HashMap::new(),
background_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(
0,
)),
checkpoint_observer,
lsp_manager,
}
}
pub fn run_in_background(
mut self,
prompt: impl Into<String>,
) -> tokio::task::JoinHandle<(Self, Result<String>)>
where
Self: Send + 'static,
{
let prompt = prompt.into();
tokio::spawn(async move {
let result = self.send(prompt).await;
(self, result)
})
}
pub fn resume(config: Config, session: Session) -> Result<Self> {
let mut agent = Agent::new(config)?;
agent.load_session(session);
Ok(agent)
}
pub fn resume_recorded(
config: Config,
session: Session,
sidecar_path: &std::path::Path,
) -> Result<Self> {
let mut agent = Agent::new(config)?;
let recorder = SidecarWriter::create(sidecar_path, &session)?;
agent.load_session(session);
agent.recorder = Some(recorder);
Ok(agent)
}
pub fn set_recorder(&mut self, w: SidecarWriter) {
self.recorder = Some(w);
}
pub fn set_reduction_policy(&mut self, policy: ReductionPolicy) {
self.reduction_policy = Some(policy);
}
pub fn reduction_policy(&self) -> Option<&ReductionPolicy> {
self.reduction_policy.as_ref()
}
pub fn set_schema_tier(&mut self, tier: crate::tools::SchemaTier) {
self.config.tool_schema_tier = tier;
}
pub fn set_tool_schema_tier(
&mut self,
name: impl Into<String>,
tier: crate::tools::SchemaTier,
) {
self.config
.tool_overrides
.entry(name.into())
.or_default()
.schema_tier = Some(tier);
}
fn schema_tier_signature(&self) -> u64 {
use std::hash::{Hash, Hasher};
let mut hasher = std::collections::hash_map::DefaultHasher::new();
self.config.tool_schema_tier.hash(&mut hasher);
let mut overrides: Vec<(&str, crate::tools::SchemaTier)> = self
.config
.tool_overrides
.iter()
.filter_map(|(name, o)| o.schema_tier.map(|t| (name.as_str(), t)))
.collect();
overrides.sort_by_key(|(name, _)| *name);
for (name, tier) in overrides {
name.hash(&mut hasher);
tier.hash(&mut hasher);
}
hasher.finish()
}
pub fn set_span_summarizer(
&mut self,
summarizer: impl reduce::summarize::SpanSummarizer + Send + Sync + 'static,
) {
self.span_summarizer = Some(std::sync::Arc::new(summarizer));
}
pub fn prepare_cleared_turns_summary(
&self,
msgs: &[ChatMessage],
policy: &ReductionPolicy,
prior: &ReductionLog,
) -> Option<reduce::PreparedClearSummary> {
let summarizer = self.span_summarizer.as_deref()?;
reduce::prepare_cleared_turns_summary(msgs, policy, prior, summarizer)
}
pub fn set_event_sink(&mut self, sink: crate::EventSink) {
self.config.event_sink = Some(sink);
}
pub fn set_permissions_approval_handler(
&mut self,
handler: impl crate::permissions::PermissionsApprovalHandler + 'static,
) {
let handler: std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler> =
std::sync::Arc::new(handler);
self.permissions_approval_handler = Some(handler.clone());
self.ctx.sandbox_approval_handler = Some(crate::sandbox::SandboxApprovalHandler(handler));
}
pub fn set_legacy_approval_handler(&mut self, handler: crate::config::ApprovalHandler) {
self.config.approval_handler = Some(handler);
}
pub fn set_subagent_store(
&mut self,
store: std::sync::Arc<crate::store::SessionStore>,
session_name: impl Into<String>,
) {
self.subagent_store = Some((store, session_name.into()));
}
pub fn set_claude_runtime_manifest(
&mut self,
manifest: crate::claude_runtime_state::ClaudeRuntimeManifest,
) {
self.config.claude_runtime_tools_enabled = true;
self.claude_runtime_manifest = Some(manifest);
}
pub fn restore_claude_project_agents(&mut self) -> Result<usize> {
let definitions = crate::claude_compat::load_project_agents(&self.config.cwd)?;
crate::claude_compat::enable_claude_subagent_compatibility(&mut self.config);
for imported in &definitions {
self.config.subagents_definitions.insert(
imported.definition.name.clone(),
imported.definition.clone(),
);
}
Ok(definitions.len())
}
pub fn claude_runtime_manifest(
&self,
) -> Option<&crate::claude_runtime_state::ClaudeRuntimeManifest> {
self.claude_runtime_manifest.as_ref()
}
pub fn claude_runtime_manifest_mut(
&mut self,
) -> Option<&mut crate::claude_runtime_state::ClaudeRuntimeManifest> {
self.claude_runtime_manifest.as_mut()
}
pub fn set_child_approval_handler_factory(
&mut self,
factory: impl Fn(
String,
std::sync::Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>,
) -> std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler>
+ Send
+ Sync
+ 'static,
) {
self.child_approval_handler_factory = Some(std::sync::Arc::new(factory));
}
pub fn pending_child_approvals(&self) -> Vec<crate::subagents::QueuedApproval> {
self.pending_child_approvals
.lock()
.map(|q| q.clone())
.unwrap_or_default()
}
pub fn set_session_titler(
&mut self,
titler: impl crate::session_title::SessionTitler + Send + Sync + 'static,
) {
self.session_titler = Some(std::sync::Arc::new(titler));
}
pub fn auto_title(&self) -> Option<String> {
let titler = self.session_titler.as_deref()?;
crate::session_title::auto_title(&self.history, titler)
}
pub fn usage_records(&self) -> &[crate::usage_log::UsageRecord] {
&self.usage_log
}
pub fn save_usage_log(&self, store: &crate::store::SessionStore, name: &str) -> Result<()> {
store.save_usage_log(name, &self.usage_log)
}
pub fn queue_steer(&self, message: impl Into<String>) {
self.steer_queue
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.queue_unchecked(message.into());
}
pub(crate) fn steer_queue_handle(&self) -> std::sync::Arc<std::sync::Mutex<SteerInbox>> {
self.steer_queue.clone()
}
pub fn queue_follow_up(&mut self, message: impl Into<String>) {
self.follow_up_queue.push_back(message.into());
}
pub fn queued_steer_count(&self) -> usize {
self.steer_queue
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.len()
+ self.follow_up_queue.len()
}
pub fn reduction_log(&self) -> &ReductionLog {
&self.reduction_log
}
pub fn set_context_limit(&mut self, limit: u64) {
self.context_limit = Some(limit);
}
pub fn context_limit(&self) -> Option<u64> {
self.context_limit
}
pub fn model(&self) -> &str {
&self.config.model
}
pub fn set_model(&mut self, model: impl Into<String>) {
self.config.model = model.into();
}
pub fn switch_model(&mut self, model: impl Into<String>) {
let to = model.into();
if !self.config.model_switch_allow_switch || self.config.model == to {
self.set_model(to);
return;
}
let from = self.config.model.clone();
let touched = reduce::rehydrate::filter_reasoning_artifacts(&mut self.history);
self.set_model(to.clone());
self.model_change_log
.push(crate::model_change::ModelChangeRecord::new(
self.turn_index,
from,
to,
true,
touched,
now_ms(),
));
}
pub fn model_change_records(&self) -> &[crate::model_change::ModelChangeRecord] {
&self.model_change_log
}
pub fn save_model_change_log(
&self,
store: &crate::store::SessionStore,
name: &str,
) -> Result<()> {
store.save_model_change_log(name, &self.model_change_log)
}
pub fn git_metadata(&self) -> Option<&crate::git_metadata::GitMetadataRecord> {
self.git_metadata.as_ref()
}
pub fn save_git_metadata(&self, store: &crate::store::SessionStore, name: &str) -> Result<()> {
match &self.git_metadata {
Some(record) => store.save_git_metadata(name, record),
None => Ok(()),
}
}
pub fn session_persist(&self) -> bool {
self.config.session_persist
}
pub fn session_name(&self) -> Option<&str> {
self.config.session_name.as_deref()
}
pub fn request_issued(&self) -> bool {
self.requests_issued
}
pub fn cache_established(&self) -> bool {
self.cache_established
}
pub fn imported_prefix_len(&self) -> Option<usize> {
self.imported_prefix_len
}
pub fn set_reduction_log(&mut self, log: ReductionLog) {
self.reduction_log = log;
}
pub fn load_session(&mut self, session: Session) {
let system = self.history.first().cloned();
self.history.clear();
if let Some(sys) = system {
self.history.push(sys);
}
self.history.extend(session.messages);
self.imported_prefix_len = Some(self.history.len());
self.cache_established = false;
self.last_cache_activity_ms = self
.history
.iter()
.rev()
.find_map(|m| m.metadata.get("timestamp"))
.and_then(|ts| crate::sidecar::rfc3339_to_ms(ts));
}
fn record(&mut self, msg: &ChatMessage) -> Result<()> {
if let Some(recorder) = self.recorder.as_mut() {
recorder.append(msg)?;
}
Ok(())
}
pub fn save_transcript(&self, path: impl AsRef<std::path::Path>) -> Result<()> {
let mut out = String::new();
for m in &self.history {
out.push_str(&serde_json::to_string(m).map_err(Error::Decode)?);
out.push('\n');
}
std::fs::write(path, out)?;
Ok(())
}
pub fn load_transcript(&mut self, path: impl AsRef<std::path::Path>) -> Result<()> {
let text = std::fs::read_to_string(path)?;
let mut history = Vec::new();
for line in text.lines().map(str::trim).filter(|l| !l.is_empty()) {
history.push(serde_json::from_str::<ChatMessage>(line).map_err(Error::Decode)?);
}
self.history = history;
Ok(())
}
pub fn checkpoint(&self) -> usize {
self.history.len()
}
pub fn rewind_to(&mut self, checkpoint: usize) {
self.history.truncate(checkpoint.min(self.history.len()));
}
pub async fn send_with_files(
&mut self,
text: impl Into<String>,
files: &[std::path::PathBuf],
) -> Result<String> {
let mut prompt = text.into();
for path in files {
let block = match std::fs::read(path) {
Ok(bytes) => match String::from_utf8(bytes.clone()) {
Ok(s) => format!("\n\n[file: {}]\n{}", path.display(), s),
Err(_) => format!(
"\n\n[file: {} — {} bytes, binary content omitted]",
path.display(),
bytes.len()
),
},
Err(e) => format!("\n\n[file: {} — could not read: {e}]", path.display()),
};
prompt.push_str(&block);
}
let expanded = self.expand_prompt_async(&prompt).await;
let msg = ChatMessage::user(expanded);
self.guard_candidate_message(&msg)?;
self.record(&msg)?;
self.history.push(msg);
self.run_loop().await
}
pub async fn send_with_images(
&mut self,
text: impl Into<String>,
image_urls: &[String],
) -> Result<String> {
let expanded = self.expand_prompt_async(&text.into()).await;
let msg = ChatMessage::user_with_images(expanded, image_urls);
self.guard_candidate_message(&msg)?;
self.record(&msg)?;
self.history.push(msg);
self.run_loop().await
}
pub fn expand_prompt(&self, input: &str) -> String {
let trimmed = input.trim_start();
let Some(rest) = trimmed.strip_prefix('/') else {
return input.to_string();
};
let (name, args) = match rest.split_once(char::is_whitespace) {
Some((n, a)) => (n, a.trim()),
None => (rest, ""),
};
match self.config.prompts.get(name) {
Some(template) => template.replace("{args}", args),
None => input.to_string(),
}
}
pub async fn expand_prompt_async(&self, input: &str) -> String {
let local = self.expand_prompt(input);
if local != input {
return local; }
let trimmed = input.trim_start();
let Some(rest) = trimmed.strip_prefix('/') else {
return input.to_string();
};
let (name, args) = match rest.split_once(char::is_whitespace) {
Some((n, a)) => (n, a.trim()),
None => (rest, ""),
};
let Some(source) = self.mcp_prompts.get(name) else {
return input.to_string();
};
let arg_map = match source.arg_names() {
[] => std::collections::BTreeMap::new(),
[single] => {
let mut m = std::collections::BTreeMap::new();
if !args.is_empty() {
m.insert(single.clone(), args.to_string());
}
m
}
_ => args
.split_whitespace()
.filter_map(|pair| pair.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
};
match source.render(arg_map).await {
Ok(rendered) => rendered,
Err(e) => format!("Error: mcp prompt `{name}` failed: {e}"),
}
}
pub fn register_mcp_prompt(
&mut self,
command_name: impl Into<String>,
source: impl crate::sdk::SdkPromptSource + 'static,
) {
self.mcp_prompts
.insert(command_name.into(), Box::new(source));
}
pub fn append_system_note(&mut self, text: &str) {
if let Some(system) = self.history.first_mut() {
if system.role == Role::System {
system
.content
.get_or_insert_with(String::new)
.push_str(text);
}
}
}
pub fn maybe_compact(&mut self) -> bool {
if !self.config.compaction_enabled {
return false;
}
let threshold = self.config.compact_after_messages;
let message_trigger = threshold.is_some_and(|t| self.history.len() > t);
let pressure_trigger = self.compaction_pressure_triggered();
if threshold.is_none() && self.config.compaction_reserve_tokens.is_none() {
return false;
}
if !message_trigger && !pressure_trigger {
return false;
}
if let Some(policy) = self.reduction_policy.as_mut() {
if let Some(t) = threshold {
policy.clear_turns_older_than = Some(t);
}
return message_trigger;
}
let keep_recent = if message_trigger {
(threshold.unwrap() / 2).max(2)
} else {
self.keep_recent_count_by_tokens()
};
if self.history.len() <= keep_recent {
return false;
}
let mut first = 1usize;
if matches!(self.config.cache_plan, CachePlan::ImportedPrefix) {
if let Some(protected) = self.imported_prefix_len {
first = first.max(protected);
}
}
let mut cut = self.history.len() - keep_recent;
if cut <= first {
return false;
}
while cut < self.history.len() && self.history[cut].role == Role::Tool {
cut += 1;
}
if cut >= self.history.len() {
return false;
}
let dropped = cut - first;
let summary_text = match &self.config.compaction_focus_instructions {
Some(focus) if !focus.is_empty() => format!(
"[earlier conversation compacted: {dropped} message(s) summarized to save context]\n\nFocus: {focus}"
),
_ => format!(
"[earlier conversation compacted: {dropped} message(s) summarized to save context]"
),
};
let summary = ChatMessage::system(summary_text);
let mut new_history = Vec::with_capacity(first + keep_recent + 2);
new_history.extend(self.history[..first].iter().cloned());
new_history.push(summary);
new_history.extend(self.history.split_off(cut));
self.history = new_history;
true
}
fn compaction_pressure_triggered(&self) -> bool {
let Some(reserve) = self.config.compaction_reserve_tokens else {
return false;
};
let limit = provider::model_context_limit(&self.config.model)
.unwrap_or(provider::UNKNOWN_MODEL_CONTEXT_FLOOR);
let used = crate::tokens::estimate_view_tokens(&self.history);
used.saturating_add(reserve) > limit
}
fn keep_recent_count_by_tokens(&self) -> usize {
let budget = self.config.compaction_keep_recent_tokens.unwrap_or(20_000);
let mut used = 0u64;
let mut count = 0usize;
for msg in self.history.iter().skip(1).rev() {
let t = crate::tokens::estimate_view_tokens(std::slice::from_ref(msg));
if used.saturating_add(t) > budget && count > 0 {
break;
}
used = used.saturating_add(t);
count += 1;
}
count.max(2)
}
pub fn register_tool(&mut self, tool: impl crate::tools::Tool + 'static) {
self.registry.register(tool);
if self.requests_issued {
self.cache_established = false;
}
}
pub fn history(&self) -> &[ChatMessage] {
&self.history
}
pub async fn send(&mut self, user_input: impl Into<String>) -> Result<String> {
let expanded = self.expand_prompt_async(&user_input.into()).await;
let msg = ChatMessage::user(expanded);
self.guard_candidate_message(&msg)?;
self.record(&msg)?;
self.history.push(msg);
self.run_loop().await
}
fn guard_candidate_message(&self, msg: &ChatMessage) -> Result<()> {
let Some(limit) = self.context_limit else {
return Ok(());
};
let messages = match &self.reduction_policy {
None => {
let mut messages = self.history.clone();
messages.push(msg.clone());
messages
}
Some(policy) => {
let has_system = self.history.first().is_some_and(|m| m.role == Role::System);
let mut reducible = self.history[usize::from(has_system)..].to_vec();
reducible.push(msg.clone());
let mut prepared = policy.clone();
reduce::prepare_read_freshness(&mut prepared, &reducible);
let (view, _) =
reduce::project_messages(&reducible, &prepared, &self.reduction_log);
let mut messages = Vec::with_capacity(view.len() + usize::from(has_system));
if has_system {
messages.push(self.history[0].clone());
}
messages.extend(view);
messages
}
};
let messages =
provider::apply_cache_plan(&messages, self.config.cache_plan, self.imported_prefix_len);
let tools = self.tool_schemas();
let (fits, projected_tokens) = crate::tokens::context_guard(&messages, &tools, limit);
if !fits {
return Err(Error::ContextLimitExceeded {
projected_tokens,
reserve_tokens: crate::tokens::CONTEXT_RESPONSE_RESERVE_TOKENS,
context_limit: limit,
model: self.config.model.clone(),
});
}
Ok(())
}
fn build_request_messages(&mut self) -> Vec<ChatMessage> {
let messages = match self.reduction_policy.clone() {
None => self.history.clone(),
Some(mut policy) => {
reduce::prepare_read_freshness(&mut policy, &self.history[1..]);
if matches!(self.config.cache_plan, CachePlan::ImportedPrefix) {
policy.protect_imported_prefix =
self.imported_prefix_len.map(|n| n.saturating_sub(1));
}
if policy.summarize_cleared_turns {
if let Some(summarizer) = self.span_summarizer.as_deref() {
policy.cleared_turns_summary = reduce::prepare_cleared_turns_summary(
&self.history[1..],
&policy,
&self.reduction_log,
summarizer,
);
}
}
let (view, log) =
reduce::project_messages(&self.history[1..], &policy, &self.reduction_log);
self.reduction_log = log;
let mut messages = Vec::with_capacity(view.len() + 1);
messages.push(self.history[0].clone());
messages.extend(view);
messages
}
};
let tier_sig = self.schema_tier_signature();
let busted =
provider::tier_change_is_cache_bust(self.last_tool_schema_tier_signature, tier_sig);
self.last_tool_schema_tier_signature = Some(tier_sig);
let effective_cache_plan = if busted {
CachePlan::Off
} else {
self.config.cache_plan
};
let will_annotate = matches!(effective_cache_plan, CachePlan::ImportedPrefix)
&& self.imported_prefix_len.is_some_and(|n| n > 0);
let idle_secs = self
.last_cache_activity_ms
.map(|last| (now_ms() - last).max(0) / 1000);
self.pending_cache_turn = (will_annotate, self.cache_established, idle_secs);
provider::apply_cache_plan(&messages, effective_cache_plan, self.imported_prefix_len)
}
fn drain_steer_queue(
queue: &mut std::collections::VecDeque<String>,
mode: SteeringMode,
) -> Option<String> {
if queue.is_empty() {
return None;
}
match mode {
SteeringMode::All => Some(queue.drain(..).collect::<Vec<_>>().join("\n\n")),
SteeringMode::OneAtATime => queue.pop_front(),
}
}
async fn run_loop(&mut self) -> Result<String> {
let _steer_turn = SteerTurnGuard::new(self.steer_queue.clone());
let mut output_tokens_used: u64 = 0;
if let Some(cp) = &self.checkpoint_observer {
let label = self
.history
.last()
.and_then(|m| m.content.as_deref())
.unwrap_or("")
.to_string();
cp.begin_turn(&label);
}
for _ in 0..self.config.max_iterations {
self.maybe_compact();
let steer_msg = {
let mut inbox = self
.steer_queue
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
inbox.drain(self.config.steering_mode)
};
if let Some(steer_msg) = steer_msg {
let msg = ChatMessage::user(steer_msg);
self.record(&msg)?;
self.history.push(msg);
}
let tools = self.tool_schemas();
let messages = self.build_request_messages();
if let Some(limit) = self.context_limit {
let (fits, projected_tokens) =
crate::tokens::context_guard(&messages, &tools, limit);
if !fits {
return Err(Error::ContextLimitExceeded {
projected_tokens,
reserve_tokens: crate::tokens::CONTEXT_RESPONSE_RESERVE_TOKENS,
context_limit: limit,
model: self.config.model.clone(),
});
}
}
let req = ChatRequest {
model: self.config.model.clone(),
messages,
tools,
temperature: self.config.temperature,
max_tokens: self.config.max_tokens,
effort: self.config.effort.clone(),
response_format: self.config.response_format.clone(),
extra_body: self.config.extra_body.clone(),
};
let (mut assistant, usage) = {
let sink = self.config.event_sink.as_ref();
let on_delta = move |s: &str| {
if let Some(sink) = sink {
sink(AgentEvent::TextDelta(s.to_string()));
}
};
self.requests_issued = true;
self.provider.complete(&req, &on_delta).await?
};
assistant
.metadata
.insert("model".to_string(), self.config.model.clone());
output_tokens_used += usage.completion_tokens;
self.total_output_tokens += usage.completion_tokens;
let (will_annotate, cache_established, idle_secs) = self.pending_cache_turn;
let warning_applies_to_this_model =
provider::is_anthropic_family_model(&self.config.model);
if self.config.cache_warnings && warning_applies_to_this_model {
if let Some(reason) =
provider::cache_cold_reason(will_annotate, cache_established, idle_secs, &usage)
{
self.emit(AgentEvent::CacheWarning {
message: reason.message(),
});
}
}
if will_annotate {
self.last_cache_activity_ms = Some(now_ms());
self.cache_established = true;
}
self.emit(AgentEvent::Usage(usage.clone()));
self.emit(AgentEvent::TurnCompleted);
self.usage_log
.push(crate::usage_log::UsageRecord::from_usage(
self.turn_index,
&self.config.model,
&usage,
now_ms(),
));
self.turn_index += 1;
self.record(&assistant)?;
self.history.push(assistant.clone());
let calls = assistant.tool_calls().to_vec();
if calls.is_empty() {
let steer_msg = self
.steer_queue
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.drain_or_close(self.config.steering_mode);
if let Some(steer_msg) = steer_msg {
let msg = ChatMessage::user(steer_msg);
self.record(&msg)?;
self.history.push(msg);
continue;
}
if let Some(follow_up_msg) =
Self::drain_steer_queue(&mut self.follow_up_queue, self.config.follow_up_mode)
{
let msg = ChatMessage::user(follow_up_msg);
self.record(&msg)?;
self.history.push(msg);
continue;
}
let final_content = assistant.content.clone().unwrap_or_default();
let veto_reason: Option<String> = self
.config
.stop_gate
.as_ref()
.and_then(|gate| gate(&final_content));
if let Some(reason) = veto_reason {
let msg = ChatMessage::user(reason);
self.record(&msg)?;
self.history.push(msg);
continue;
}
return Ok(assistant.content.unwrap_or_default());
}
if let Some(budget) = self.config.max_total_output_tokens {
if output_tokens_used >= budget {
for call in &calls {
let msg = ChatMessage::tool_result(
call.id.clone(),
call.function.name.clone(),
"[skipped: output token budget reached]".to_string(),
);
self.record(&msg)?;
self.history.push(msg);
}
return Ok(assistant.content.clone().unwrap_or_default());
}
}
if self.config.parallel_tool_calls && calls.len() > 1 {
for call in &calls {
self.emit(AgentEvent::tool_started(call));
}
let results = self.run_tools_concurrently(&calls).await;
for (call, (output, is_error)) in calls.iter().zip(results) {
self.emit(AgentEvent::ToolCallCompleted {
id: call.id.clone(),
name: call.function.name.clone(),
output: output.clone(),
is_error,
});
self.apply_tool_result(call, output, is_error)?;
}
} else {
for call in &calls {
self.emit(AgentEvent::tool_started(call));
let (output, is_error) = self.run_tool(call).await;
self.emit(AgentEvent::ToolCallCompleted {
id: call.id.clone(),
name: call.function.name.clone(),
output: output.clone(),
is_error,
});
self.apply_tool_result(call, output, is_error)?;
}
}
}
Err(Error::MaxIterations(self.config.max_iterations))
}
fn apply_tool_result(
&mut self,
call: &crate::message::ToolCall,
output: String,
is_error: bool,
) -> Result<()> {
if let Some(data_url) = output.strip_prefix(crate::tools::MULTIMODAL_IMAGE_MARKER) {
let notice = format!("[{}: image content attached below]", call.function.name);
let full_result = ChatMessage::tool_result_with_image(
call.id.clone(),
call.function.name.clone(),
notice.clone(),
data_url.to_string(),
);
let hist_result = ChatMessage::tool_result_with_image(
call.id.clone(),
call.function.name.clone(),
notice,
data_url.to_string(),
);
self.record(&full_result)?;
self.history.push(hist_result);
return Ok(());
}
let mut full_result =
ChatMessage::tool_result(call.id.clone(), call.function.name.clone(), output.clone());
let for_history = if self.recorder.is_some() && self.reduction_policy.is_some() {
output
} else {
self.cap_tool_output(output)
};
let mut hist_result =
ChatMessage::tool_result(call.id.clone(), call.function.name.clone(), for_history);
if is_error {
reduce::mark_tool_error(&mut full_result);
reduce::mark_tool_error(&mut hist_result);
}
self.record(&full_result)?;
self.history.push(hist_result);
Ok(())
}
fn cap_tool_output(&self, output: String) -> String {
let Some(max) = self.config.max_tool_output_bytes else {
return output;
};
if max == 0 || output.len() <= max {
return output;
}
let mut end = max;
while end > 0 && !output.is_char_boundary(end) {
end -= 1;
}
let total = output.len();
let mut s = output[..end].to_string();
let retention = if self.recorder.is_some() {
"full output in session sidecar"
} else {
"full output not retained"
};
s.push_str(&format!(
"{CAP_NOTICE_MARKER}{total} bytes total, showing first {end}; {retention}]"
));
s
}
fn run_tool<'a>(
&'a mut self,
call: &'a crate::message::ToolCall,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = (String, bool)> + Send + 'a>> {
Box::pin(async move {
let translated_builtin = if self.config.claude_runtime_tools_enabled {
match self.translate_claude_builtin_call(call) {
Ok(translated) => translated,
Err(error) => return (format!("Error: {error}"), true),
}
} else {
None
};
let call = translated_builtin.as_ref().unwrap_or(call);
if self.config.claude_runtime_tools_enabled
&& matches!(
call.function.name.as_str(),
CLAUDE_CRON_CREATE
| CLAUDE_CRON_DELETE
| CLAUDE_CRON_LIST
| CLAUDE_SCHEDULE_WAKEUP
)
{
return self.run_claude_runtime_tool(call);
}
if call.function.name == CLAUDE_AGENT && self.config.subagents_claude_agent_alias {
return match self.translate_claude_agent_call(call) {
Ok(translated) => self.run_spawn_subagent(&translated).await,
Err(error) => (format!("Error: {error}"), true),
};
}
if call.function.name == SPAWN_SUBAGENT {
return self.run_spawn_subagent(call).await;
}
if call.function.name == SUBAGENT_STATUS && self.config.subagents_enabled {
return self.run_subagent_status(call).await;
}
match self.prepare_tool_call(call) {
PreparedCall::Done(result) => result,
PreparedCall::Ready { name, args } => {
let tool = self.registry.get(&name).expect("prepared as Ready");
let (output, is_error) = match tool.execute(args, &self.ctx).await {
Ok(out) => (out, false),
Err(e) => (format!("Error: {e}"), true),
};
if let Some(hook) = &self.config.post_tool_hook {
hook(&name, &output, is_error);
}
(output, is_error)
}
}
})
}
fn prepare_tool_call(&mut self, call: &crate::message::ToolCall) -> PreparedCall {
let name = &call.function.name;
if name == TOOL_SEARCH {
return PreparedCall::Done(self.run_tool_search(call));
}
if name == EXPAND_REDUCTION {
return PreparedCall::Done(self.run_expand_reduction(call));
}
if name == SIDECAR_SEARCH {
return PreparedCall::Done(self.run_sidecar_search(call));
}
if self.config.tools_background_enabled {
if name == BACKGROUND_EXEC {
return PreparedCall::Done(self.run_background_exec(call));
}
if name == BACKGROUND_STATUS {
return PreparedCall::Done(self.run_background_status(call));
}
if name == BACKGROUND_LIST {
return PreparedCall::Done(self.run_background_list(call));
}
if name == BACKGROUND_KILL {
return PreparedCall::Done(self.run_background_kill(call));
}
}
if self.registry.get(name).is_none() {
let err = Error::UnknownTool(name.clone());
return PreparedCall::Done((format!("Error: {err}"), true));
}
if self.config.permissions_enabled {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: name.clone(),
message: e.to_string(),
};
return PreparedCall::Done((format!("Error: {err}"), true));
}
};
if let Some(reason) = self.permissions_gate_denial(name, &args) {
return PreparedCall::Done((format!("Error: {reason}"), true));
}
self.finish_prepare(name.clone(), args)
} else {
if self.config.needs_approval(name) {
let approved = self
.config
.approval_handler
.as_ref()
.map(|h| h(call))
.unwrap_or(false);
if !approved {
return PreparedCall::Done((
format!("Error: tool `{name}` was not approved for execution"),
true,
));
}
}
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: name.clone(),
message: e.to_string(),
};
return PreparedCall::Done((format!("Error: {err}"), true));
}
};
self.finish_prepare(name.clone(), args)
}
}
fn finish_prepare(&mut self, name: String, args: serde_json::Value) -> PreparedCall {
if let Some(reason) = self.check_doom_loop(&name, &args) {
return PreparedCall::Done((format!("Error: {reason}"), true));
}
if let Some(hook) = &self.config.pre_tool_hook {
if let Some(reason) = hook(&name, &args) {
return PreparedCall::Done((
format!("Error: blocked by pre-tool hook: {reason}"),
true,
));
}
}
PreparedCall::Ready { name, args }
}
const SUBJECT_BEARING_BUILTIN_TOOLS: &'static [&'static str] = &[
"bash",
"shell",
"read_file",
"write_file",
"edit_file",
"view_image",
"apply_patch",
];
fn permissions_gate_denial(&self, name: &str, args: &serde_json::Value) -> Option<String> {
self.permissions_gate_denial_impl(name, args, self.permissions_approval_handler.as_deref())
}
fn permissions_gate_denial_impl(
&self,
name: &str,
args: &serde_json::Value,
handler: Option<&dyn crate::permissions::PermissionsApprovalHandler>,
) -> Option<String> {
use crate::permissions::{self, Decision, PathKind};
let mut deny = self.config.tool_deny_patterns.clone();
deny.extend(permissions::protected_path_deny_rules(
&self.config.permissions_protected_paths,
));
let rules = permissions::RuleSet {
deny,
ask: self.config.permissions_ask_patterns.clone(),
allow: self.config.tool_allow_patterns.clone(),
};
let default = match self.config.approval {
crate::config::ApprovalPolicy::Never => Decision::Allow,
crate::config::ApprovalPolicy::OnRequest => {
if self.config.auto_approved_tools.contains(name) {
Decision::Allow
} else {
Decision::Ask
}
}
crate::config::ApprovalPolicy::Untrusted => Decision::Ask,
crate::config::ApprovalPolicy::ModelRequested => Decision::Allow,
};
let command = args.get("command").and_then(|v| v.as_str());
let path = args.get("path").and_then(|v| v.as_str());
let patch = (name == "apply_patch")
.then(|| args.get("patch").and_then(|v| v.as_str()))
.flatten();
let decision = if let Some(command) = command {
permissions::evaluate_command(&rules, name, command, default)
} else if let Some(patch) = patch {
let mut d = rules.evaluate(name, None).unwrap_or(default);
match crate::tools::patch_target_paths(patch) {
Ok(paths) => {
for p in &paths {
let pseudo = permissions::evaluate_path_safe(
&rules,
PathKind::Write,
&self.config.cwd,
p,
Decision::Allow,
);
let real_tool = permissions::evaluate_path_subject_safe(
&rules,
name,
&self.config.cwd,
p,
Decision::Allow,
);
d = d.stricter(pseudo).stricter(real_tool);
}
}
Err(_) => {
d = d.stricter(Decision::Ask);
}
}
d
} else if let Some(path) = path {
let kind = if matches!(name, "write_file" | "edit_file") {
PathKind::Write
} else {
PathKind::Read
};
let pseudo_decision =
permissions::evaluate_path_safe(&rules, kind, &self.config.cwd, path, default);
let real_tool_decision = permissions::evaluate_path_subject_safe(
&rules,
name,
&self.config.cwd,
path,
default,
);
pseudo_decision.stricter(real_tool_decision)
} else {
rules.evaluate(name, None).unwrap_or(default)
};
let subject = Self::SUBJECT_BEARING_BUILTIN_TOOLS
.contains(&name)
.then(|| command.or(path).or(patch))
.flatten();
let req = permissions::ApprovalRequest {
tool: name,
subject,
raw_args: args,
};
let approved = permissions::decision_to_approved(decision, || {
permissions::resolve_ask(&self.permissions_approval_cache, handler, &req)
});
if approved {
None
} else {
Some(format!(
"tool `{name}` was not approved for execution (permissions engine: {decision:?})"
))
}
}
fn background_permission_denial(&self, command: &str, job_id: &str) -> Option<String> {
let args = serde_json::json!({ "command": command });
if self.config.permissions_enabled {
if let Some(crate::subagents::BackgroundPromptsPolicy::Parent) =
self.config.subagents_background_prompts
{
let handler = crate::subagents::ParentQueueApprovalHandler {
child_agent_id: format!("bg:{job_id}"),
queue: self.pending_child_approvals.clone(),
};
self.permissions_gate_denial_impl("bash", &args, Some(&handler))
} else {
self.permissions_gate_denial_impl("bash", &args, None)
}
} else if self.config.needs_approval("bash") {
Some(
"tool `bash` requires approval, which a background job cannot request \
interactively (§2.2 C6: auto-policy denies)"
.to_string(),
)
} else {
None
}
}
async fn run_tools_concurrently(
&mut self,
calls: &[crate::message::ToolCall],
) -> Vec<(String, bool)> {
if calls.iter().any(|c| {
c.function.name == SPAWN_SUBAGENT
|| c.function.name == SUBAGENT_STATUS
|| (self.config.claude_runtime_tools_enabled
&& matches!(
c.function.name.as_str(),
CLAUDE_CRON_CREATE
| CLAUDE_CRON_DELETE
| CLAUDE_CRON_LIST
| CLAUDE_SCHEDULE_WAKEUP
))
}) {
let mut out = Vec::with_capacity(calls.len());
for call in calls {
out.push(self.run_tool(call).await);
}
return out;
}
let prepared: Vec<PreparedCall> = calls.iter().map(|c| self.prepare_tool_call(c)).collect();
let mut slots: Vec<Option<(String, bool)>> = prepared
.iter()
.map(|p| match p {
PreparedCall::Done(r) => Some(r.clone()),
PreparedCall::Ready { .. } => None,
})
.collect();
let ready_idxs: Vec<usize> = prepared
.iter()
.enumerate()
.filter(|(_, p)| matches!(p, PreparedCall::Ready { .. }))
.map(|(i, _)| i)
.collect();
if !ready_idxs.is_empty() {
let futs = ready_idxs.iter().map(|&i| {
let PreparedCall::Ready { name, args } = &prepared[i] else {
unreachable!("filtered to Ready above")
};
let tool = self.registry.get(name).expect("prepared as Ready");
let args = args.clone();
let ctx = self.ctx.clone();
async move {
match tool.execute(args, &ctx).await {
Ok(out) => (out, false),
Err(e) => (format!("Error: {e}"), true),
}
}
});
let results = futures::future::join_all(futs).await;
for (idx, result) in ready_idxs.iter().zip(results) {
slots[*idx] = Some(result);
}
}
let out: Vec<(String, bool)> = slots
.into_iter()
.map(|s| s.expect("every call resolved to Some above"))
.collect();
let ready_set: std::collections::HashSet<usize> = ready_idxs.into_iter().collect();
for (i, call) in calls.iter().enumerate() {
if !ready_set.contains(&i) {
continue;
}
let (output, is_error) = &out[i];
if let Some(hook) = &self.config.post_tool_hook {
hook(&call.function.name, output, *is_error);
}
}
out
}
fn check_doom_loop(&mut self, name: &str, args: &serde_json::Value) -> Option<String> {
let threshold = self.config.doom_loop_threshold?;
if threshold < 2 {
return None;
}
let key = (name.to_string(), args.to_string());
if self.doom_loop_last_call.as_ref() == Some(&key) {
self.doom_loop_streak += 1;
} else {
self.doom_loop_last_call = Some(key);
self.doom_loop_streak = 1;
}
if self.doom_loop_streak >= threshold {
Some(format!(
"doom-loop breaker: `{name}` called with identical arguments {} times in a row \
— try a different approach instead of repeating the same call",
self.doom_loop_streak
))
} else {
None
}
}
fn is_core_tool(&self, name: &str) -> bool {
match &self.config.tool_advertising {
ToolAdvertising::Full => true,
ToolAdvertising::Deferred { core } => core.iter().any(|c| c == name),
}
}
fn schema_for(&self, t: &dyn crate::tools::Tool) -> ToolSchema {
let raw = self.raw_schema_for(t);
let tier = self.config.schema_tier_for(t.name());
let (description, parameters) =
crate::tools::tiers::minify(&raw.description, &raw.parameters, tier);
ToolSchema {
name: raw.name,
description,
parameters,
}
}
fn raw_schema_for(&self, t: &dyn crate::tools::Tool) -> ToolSchema {
ToolSchema {
name: t.name().to_string(),
description: self
.config
.tool_description(t.name(), t.description())
.to_string(),
parameters: t.parameters(),
}
}
fn tool_search_schema() -> ToolSchema {
ToolSchema {
name: TOOL_SEARCH.to_string(),
description: "Search for additional tools not currently advertised (the deferred \
MCP surface and any other non-core tools). Matches keywords case-insensitively \
against each tool's name and description. Matched tools become callable starting \
with your NEXT message, not this one."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Keyword(s) to search for in tool names and descriptions."
},
"max_results": {
"type": "integer",
"description": "Maximum number of matching tools to return."
}
},
"required": ["query"],
"additionalProperties": false
}),
}
}
pub fn tool_schemas(&self) -> Vec<ToolSchema> {
let mut out = match &self.config.tool_advertising {
ToolAdvertising::Full => self
.registry
.iter()
.filter(|t| self.config.tool_enabled(t.name()))
.map(|t| self.schema_for(t))
.collect(),
ToolAdvertising::Deferred { .. } => {
let mut out: Vec<ToolSchema> = self
.registry
.iter()
.filter(|t| self.config.tool_enabled(t.name()))
.filter(|t| {
self.is_core_tool(t.name()) || self.activated_tools.contains(t.name())
})
.map(|t| self.schema_for(t))
.collect();
out.push(Self::tool_search_schema());
out
}
};
if self.reduction_policy.is_some() {
out.push(Self::expand_reduction_schema());
out.push(Self::sidecar_search_schema());
}
if self.config.subagents_enabled {
out.push(self.spawn_subagent_schema());
if self.config.subagents_claude_agent_alias {
out.push(self.claude_agent_schema());
}
if self.config.subagents_background {
out.push(Self::subagent_status_schema());
}
}
if self.config.claude_runtime_tools_enabled {
out.extend(self.claude_builtin_tool_schemas());
out.push(Self::claude_cron_create_schema());
out.push(Self::claude_cron_delete_schema());
out.push(Self::claude_cron_list_schema());
out.push(Self::claude_schedule_wakeup_schema());
}
if self.config.tools_background_enabled {
out.push(Self::background_exec_schema());
out.push(Self::background_status_schema());
out.push(Self::background_list_schema());
out.push(Self::background_kill_schema());
}
out
}
fn claude_builtin_tool_schemas(&self) -> Vec<ToolSchema> {
let mut schemas = Vec::new();
let mut push = |alias: &str, native: &str, description: &str, parameters| {
if self.registry.get(native).is_some() && self.config.tool_enabled(native) {
schemas.push(ToolSchema {
name: alias.to_string(),
description: description.to_string(),
parameters,
});
}
};
push(
CLAUDE_BASH,
"bash",
"Claude Code-compatible shell command execution.",
serde_json::json!({
"type": "object",
"properties": {
"command": {"type": "string"},
"timeout": {"type": "integer", "description": "Timeout in milliseconds."},
"description": {"type": "string"}
},
"required": ["command"],
"additionalProperties": true
}),
);
push(
CLAUDE_READ,
"read_file",
"Claude Code-compatible file reader.",
serde_json::json!({
"type": "object",
"properties": {
"file_path": {"type": "string"},
"offset": {"type": "integer"},
"limit": {"type": "integer"}
},
"required": ["file_path"],
"additionalProperties": false
}),
);
push(
CLAUDE_WRITE,
"write_file",
"Claude Code-compatible file writer.",
serde_json::json!({
"type": "object",
"properties": {"file_path": {"type": "string"}, "content": {"type": "string"}},
"required": ["file_path", "content"],
"additionalProperties": false
}),
);
push(
CLAUDE_EDIT,
"edit_file",
"Claude Code-compatible exact file edit.",
serde_json::json!({
"type": "object",
"properties": {
"file_path": {"type": "string"},
"old_string": {"type": "string"},
"new_string": {"type": "string"},
"replace_all": {"type": "boolean"}
},
"required": ["file_path", "old_string", "new_string"],
"additionalProperties": false
}),
);
push(
CLAUDE_GLOB,
"glob",
"Claude Code-compatible file glob.",
serde_json::json!({
"type": "object",
"properties": {"pattern": {"type": "string"}, "path": {"type": "string"}},
"required": ["pattern"],
"additionalProperties": false
}),
);
push(
CLAUDE_GREP,
"search",
"Claude Code-compatible content search.",
serde_json::json!({
"type": "object",
"properties": {"pattern": {"type": "string"}, "path": {"type": "string"}},
"required": ["pattern"],
"additionalProperties": true
}),
);
schemas
}
fn translate_claude_builtin_call(
&self,
call: &crate::message::ToolCall,
) -> Result<Option<crate::message::ToolCall>> {
let native = match call.function.name.as_str() {
CLAUDE_BASH => "bash",
CLAUDE_READ => "read_file",
CLAUDE_WRITE => "write_file",
CLAUDE_EDIT => "edit_file",
CLAUDE_GLOB => "glob",
CLAUDE_GREP => "search",
_ => return Ok(None),
};
let mut args = call.function.parsed_arguments()?;
let object = args
.as_object_mut()
.ok_or_else(|| Error::InvalidArguments {
tool: call.function.name.clone(),
message: "expected a JSON object".to_string(),
})?;
if let Some(path) = object.remove("file_path") {
object.entry("path".to_string()).or_insert(path);
}
if call.function.name == CLAUDE_BASH {
if let Some(timeout) = object.remove("timeout") {
object.entry("timeout_ms".to_string()).or_insert(timeout);
}
}
if call.function.name == CLAUDE_GLOB {
if let Some(path) = object
.remove("path")
.and_then(|value| value.as_str().map(str::to_owned))
{
if let Some(pattern) = object.get_mut("pattern") {
if let Some(value) = pattern.as_str() {
if !std::path::Path::new(value).is_absolute() {
*pattern = serde_json::Value::String(
std::path::Path::new(&path)
.join(value)
.to_string_lossy()
.into_owned(),
);
}
}
}
}
}
let mut translated = call.clone();
translated.function.name = native.to_string();
translated.function.arguments = serde_json::to_string(&args)?;
Ok(Some(translated))
}
fn claude_cron_create_schema() -> ToolSchema {
ToolSchema {
name: CLAUDE_CRON_CREATE.to_string(),
description: "Record a Claude-compatible cron job in the imported runtime manifest. \
The job inherits the manifest's ACTIVE or PAUSED posture; an embedding scheduler, \
not this agent loop, owns execution."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"cron": {"type": "string", "description": "Cron expression to preserve."},
"prompt": {"type": "string", "description": "Prompt associated with the job."},
"recurring": {"type": "boolean", "default": false},
"durable": {"type": "boolean", "default": false}
},
"required": ["cron", "prompt"],
"additionalProperties": false
}),
}
}
fn claude_cron_delete_schema() -> ToolSchema {
ToolSchema {
name: CLAUDE_CRON_DELETE.to_string(),
description: "Delete a Claude-compatible cron job from the imported manifest. \
This updates state only; an embedding scheduler owns execution."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {"id": {"type": "string"}},
"required": ["id"],
"additionalProperties": false
}),
}
}
fn claude_cron_list_schema() -> ToolSchema {
ToolSchema {
name: CLAUDE_CRON_LIST.to_string(),
description: "List imported Claude cron jobs and their explicit ACTIVE or PAUSED \
manifest posture. This agent loop itself does not run a scheduler."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {},
"additionalProperties": false
}),
}
}
fn claude_schedule_wakeup_schema() -> ToolSchema {
ToolSchema {
name: CLAUDE_SCHEDULE_WAKEUP.to_string(),
description: "Replace the one-shot wakeup stored in the imported Claude manifest. \
The wakeup inherits the manifest's ACTIVE or PAUSED posture; an embedding scheduler \
owns timer execution."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"delaySeconds": {"type": "integer", "minimum": 0},
"reason": {"type": "string"},
"prompt": {"type": "string"}
},
"required": ["delaySeconds"],
"additionalProperties": false
}),
}
}
fn background_exec_schema() -> ToolSchema {
ToolSchema {
name: BACKGROUND_EXEC.to_string(),
description: "Run a shell command in the BACKGROUND: spawns it as a detached \
process and returns a `job_id` IMMEDIATELY, before the command finishes — this \
call never returns the command's output. Poll `background_status` with the \
`job_id` to check progress and retrieve captured output; use `background_kill` \
to cancel it early. The command goes through the exact same sandbox/permission \
checks as a foreground `bash` call, and any check that would need an \
interactive approval is denied automatically (a background job cannot wait for \
one)."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "Shell command to run in the background via `sh -c`."
}
},
"required": ["command"],
"additionalProperties": false
}),
}
}
fn background_status_schema() -> ToolSchema {
ToolSchema {
name: BACKGROUND_STATUS.to_string(),
description: "Check on a background job spawned via background_exec: its \
running/exited/killed status, exit code (once known), and the command's \
captured stdout/stderr so far (bounded — very large output is truncated with a \
marker). Once the job has exited or been killed, this call also reaps it (it \
will no longer appear in background_list or accept further status polls)."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"job_id": {
"type": "string",
"description": "The id `background_exec` returned when this job was \
started."
}
},
"required": ["job_id"],
"additionalProperties": false
}),
}
}
fn background_list_schema() -> ToolSchema {
ToolSchema {
name: BACKGROUND_LIST.to_string(),
description: "List every background job currently tracked (running, or finished \
but not yet polled via background_status) — job id, command, status, pid, and \
start time for each. Does not retrieve output or reap anything."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {},
"additionalProperties": false
}),
}
}
fn background_kill_schema() -> ToolSchema {
ToolSchema {
name: BACKGROUND_KILL.to_string(),
description: "Kill a background job's real process immediately (a no-op, not an \
error, if it already exited on its own) and reap it."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"job_id": {
"type": "string",
"description": "The id `background_exec` returned when this job was \
started."
}
},
"required": ["job_id"],
"additionalProperties": false
}),
}
}
fn spawn_subagent_schema(&self) -> ToolSchema {
let mut names: Vec<&str> = self
.config
.subagents_definitions
.keys()
.map(String::as_str)
.collect();
names.sort_unstable();
let agent_type_desc = if names.is_empty() {
"Optional named subagent type to run (none configured — omit this and pass \
`system_prompt` instead)."
.to_string()
} else {
format!(
"Optional named subagent type to run: {}. Omit to run an ad-hoc subagent with \
your own `system_prompt` instead.",
names.join(", ")
)
};
let background_desc = if self.config.subagents_background {
"Run this subagent in the background instead of waiting for it — this call \
returns immediately with a `subagent_id`; poll `subagent_status` with that id for \
the result."
} else {
"Background subagents are disabled for this agent — this must be omitted or false."
};
ToolSchema {
name: SPAWN_SUBAGENT.to_string(),
description: "Spawn a subagent to work on a self-contained task and (by default) \
wait for its final answer, which is returned as this call's result. The \
subagent runs its own independent reasoning/tool loop; it does not see your \
conversation except for the `task` text you give it here."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"task": {
"type": "string",
"description": "The self-contained task/prompt for the subagent."
},
"agent_type": {
"type": "string",
"description": agent_type_desc
},
"system_prompt": {
"type": "string",
"description": "Inline system prompt for an ad-hoc subagent (ignored \
if `agent_type` is given — the named type's own prompt is used \
instead)."
},
"background": {
"type": "boolean",
"description": background_desc
}
},
"required": ["task"],
"additionalProperties": false
}),
}
}
fn claude_agent_schema(&self) -> ToolSchema {
let mut names: Vec<String> = self.config.subagents_definitions.keys().cloned().collect();
names.push("general-purpose".into());
names.sort_unstable();
names.dedup();
ToolSchema {
name: CLAUDE_AGENT.to_string(),
description: "Claude Code-compatible subagent dispatcher. Runs a named or ad-hoc \
child agent; children default to background execution in this compatibility mode."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"prompt": {"type": "string", "description": "Self-contained child task."},
"subagent_type": {
"type": "string",
"description": format!("Named agent type. Available: {}", names.join(", "))
},
"description": {
"type": "string",
"description": "Short human-facing task label; preserved as descriptive input."
},
"model": {
"type": "string",
"description": "Optional model alias or full provider slug for this child."
},
"run_in_background": {
"type": "boolean",
"description": "Whether to return immediately with a child id (default true)."
}
},
"required": ["prompt"],
"additionalProperties": false
}),
}
}
fn translate_claude_agent_call(
&self,
call: &crate::message::ToolCall,
) -> Result<crate::message::ToolCall> {
let args = call
.function
.parsed_arguments()
.map_err(|error| Error::InvalidArguments {
tool: CLAUDE_AGENT.to_string(),
message: error.to_string(),
})?;
let object = args.as_object().ok_or_else(|| Error::InvalidArguments {
tool: CLAUDE_AGENT.to_string(),
message: "arguments must be an object".to_string(),
})?;
let mut translated = serde_json::Map::new();
if let Some(value) = object.get("prompt") {
translated.insert("task".to_string(), value.clone());
}
if let Some(value) = object.get("subagent_type") {
if value.as_str() != Some("general-purpose") {
translated.insert("agent_type".to_string(), value.clone());
}
}
if let Some(value) = object.get("model") {
translated.insert("model".to_string(), value.clone());
}
translated.insert(
"background".to_string(),
object
.get("run_in_background")
.cloned()
.unwrap_or(serde_json::Value::Bool(true)),
);
Ok(crate::message::ToolCall {
id: call.id.clone(),
kind: call.kind.clone(),
function: crate::message::FunctionCall {
name: SPAWN_SUBAGENT.to_string(),
arguments: serde_json::Value::Object(translated).to_string(),
},
})
}
fn run_claude_runtime_tool(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(value) if value.is_object() => value,
Ok(_) => {
return (
format!("Error: {} arguments must be an object", call.function.name),
true,
)
}
Err(error) => return (format!("Error: {error}"), true),
};
let object = args.as_object().expect("checked object above");
let Some(manifest) = self.claude_runtime_manifest.as_mut() else {
return (
"Error: Claude runtime compatibility was enabled without an imported runtime \
manifest; refusing to invent scheduler state"
.to_string(),
true,
);
};
let active = matches!(
manifest.execution_state,
crate::claude_runtime_state::ClaudeRuntimeExecutionState::Active
);
let state = if active { "active" } else { "paused" };
let active_now_ms = active.then(now_ms);
let active_now_unix = active_now_ms.map(|ms| ms.div_euclid(1_000));
match call.function.name.as_str() {
CLAUDE_CRON_LIST => {
let jobs: Vec<serde_json::Value> = manifest
.active_crons
.iter()
.map(|job| {
serde_json::json!({
"id": job.id,
"cron": job.schedule,
"prompt": job.prompt,
"recurring": job.recurring,
"durable": job.durable_requested,
"state": state
})
})
.collect();
let notice = if active {
"The manifest is active; an attached scheduler may claim due jobs."
} else {
"Imported jobs are preserved but no scheduler is running."
};
(
serde_json::json!({
"execution_state": state,
"execution_notice": notice,
"jobs": jobs
})
.to_string(),
false,
)
}
CLAUDE_CRON_CREATE => {
let Some(schedule) = object.get("cron").and_then(serde_json::Value::as_str) else {
return ("Error: CronCreate requires string `cron`".to_string(), true);
};
let Some(prompt) = object.get("prompt").and_then(serde_json::Value::as_str) else {
return (
"Error: CronCreate requires string `prompt`".to_string(),
true,
);
};
let recurring = object
.get("recurring")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let durable_requested = object
.get("durable")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let mut sequence = 1_u64;
let id = loop {
let candidate = format!("sc{sequence:06}");
if !manifest.active_crons.iter().any(|job| job.id == candidate) {
break candidate;
}
sequence += 1;
};
let kind = if recurring { "recurring " } else { "" };
let created_at = active_now_ms.map(crate::sidecar::ms_to_rfc3339);
let result = if active {
format!(
"Scheduled {kind}job {id} ({schedule}) in ACTIVE state. The job is \
eligible for execution by the attached scheduler."
)
} else {
format!(
"Scheduled {kind}job {id} ({schedule}) in PAUSED state. The job is preserved \
in the continuation manifest but no scheduler is running and it will not execute."
)
};
manifest
.active_crons
.push(crate::claude_runtime_state::ClaudeCronJob {
id: id.clone(),
tool_use_id: call.id.clone(),
schedule: schedule.to_string(),
recurring,
durable_requested,
prompt: prompt.to_string(),
created_at,
expires_after_seconds: None,
creation_result: result.clone(),
});
manifest
.active_crons
.sort_by(|left, right| left.id.cmp(&right.id));
if let Some(now_unix) = active_now_unix {
let scheduler_before = manifest.scheduler.clone();
if let Err(error) = manifest.reconcile_scheduler(now_unix) {
manifest.active_crons.retain(|job| job.id != id);
manifest.scheduler = scheduler_before;
return (format!("Error: {error}"), true);
}
}
(result, false)
}
CLAUDE_CRON_DELETE => {
let Some(id) = object.get("id").and_then(serde_json::Value::as_str) else {
return ("Error: CronDelete requires string `id`".to_string(), true);
};
let Some(index) = manifest.active_crons.iter().position(|job| job.id == id) else {
return (
format!("Error: unknown {state} Claude cron job `{id}`"),
true,
);
};
let removed = manifest.active_crons.remove(index);
if let Some(now_unix) = active_now_unix {
let scheduler_before = manifest.scheduler.clone();
if let Err(error) = manifest.reconcile_scheduler(now_unix) {
manifest.active_crons.insert(index, removed);
manifest.scheduler = scheduler_before;
return (format!("Error: {error}"), true);
}
}
let result = if active {
format!("Cancelled job {id}. The job was removed from ACTIVE scheduler state.")
} else {
format!("Cancelled job {id}. The job was PAUSED; no execution occurred.")
};
(result, false)
}
CLAUDE_SCHEDULE_WAKEUP => {
let Some(delay_seconds) = object
.get("delaySeconds")
.and_then(serde_json::Value::as_u64)
else {
return (
"Error: ScheduleWakeup requires integer `delaySeconds`".to_string(),
true,
);
};
let reason = object
.get("reason")
.and_then(serde_json::Value::as_str)
.map(str::to_string);
let prompt = object
.get("prompt")
.and_then(serde_json::Value::as_str)
.map(str::to_string);
let created_at = active_now_ms.map(crate::sidecar::ms_to_rfc3339);
let scheduled_for = active_now_ms
.map(|now| {
let delay_ms = i64::try_from(delay_seconds)
.unwrap_or(i64::MAX)
.saturating_mul(1_000);
crate::sidecar::ms_to_rfc3339(now.saturating_add(delay_ms))
})
.unwrap_or_else(|| "PAUSED".to_string());
let result = if active {
format!(
"Next wakeup scheduled for {scheduled_for} (in {delay_seconds}s). Runtime \
state is ACTIVE; the attached scheduler may execute it."
)
} else {
format!(
"Next wakeup scheduled for PAUSED (in {delay_seconds}s). The request replaced \
the prior wakeup in the manifest, but no timer is running and it will not execute."
)
};
let previous_wakeups = if active {
Some(manifest.pending_wakeups.clone())
} else {
None
};
manifest.pending_wakeups.clear();
manifest
.pending_wakeups
.push(crate::claude_runtime_state::ClaudeWakeup {
tool_use_id: call.id.clone(),
delay_seconds,
reason,
prompt,
created_at,
scheduled_for: Some(scheduled_for),
creation_result: result.clone(),
});
if let Some(now_unix) = active_now_unix {
let scheduler_before = manifest.scheduler.clone();
if let Err(error) = manifest.reconcile_scheduler(now_unix) {
manifest.pending_wakeups = previous_wakeups.unwrap_or_default();
manifest.scheduler = scheduler_before;
return (format!("Error: {error}"), true);
}
}
(result, false)
}
_ => unreachable!("runtime tool dispatch is name-gated"),
}
}
fn subagent_status_schema() -> ToolSchema {
ToolSchema {
name: SUBAGENT_STATUS.to_string(),
description: "Check on (and, once finished, retrieve the result of) a background \
subagent spawned via spawn_subagent with background=true. Pass the \
`subagent_id` that spawn returned."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"subagent_id": {
"type": "string",
"description": "The id `spawn_subagent` returned when this subagent \
was spawned."
}
},
"required": ["subagent_id"],
"additionalProperties": false
}),
}
}
fn build_child_config(
&self,
definition: Option<&crate::subagents::NamedAgentDefinition>,
inline_system_prompt: Option<String>,
model_override: Option<String>,
) -> Config {
let system_prompt = definition
.map(|d| d.system_prompt.clone())
.filter(|s| !s.is_empty())
.or(inline_system_prompt)
.unwrap_or_else(|| self.config.system_prompt.clone());
let model = model_override.unwrap_or_else(|| self.config.model.clone());
let mut child = Config::builder()
.model(model)
.system_prompt(system_prompt)
.cwd(self.config.cwd.clone())
.sandbox(self.config.sandbox)
.approval(self.config.approval)
.max_iterations(self.config.max_iterations)
.build();
child.base_url = self.config.base_url.clone();
child.api_key = self.config.api_key.clone();
child.api_key_env = self.config.api_key_env.clone();
child.api_key_cmd = self.config.api_key_cmd.clone();
child.max_total_output_tokens = self.config.max_total_output_tokens;
child.max_tool_output_bytes = self.config.max_tool_output_bytes;
child.max_tokens = self.config.max_tokens;
child.doom_loop_threshold = self.config.doom_loop_threshold;
child.edit_file_require_read_before_edit = self.config.edit_file_require_read_before_edit;
child.tool_overrides = self.config.tool_overrides.clone();
child.auto_approved_tools = self.config.auto_approved_tools.clone();
child.tool_deny_patterns = self.config.tool_deny_patterns.clone();
child.tool_allow_patterns = self.config.tool_allow_patterns.clone();
child.permissions_enabled = self.config.permissions_enabled;
child.permissions_ask_patterns = self.config.permissions_ask_patterns.clone();
child.permissions_protected_paths = self.config.permissions_protected_paths.clone();
child.network_policy = self.config.network_policy.clone();
child.sandbox_os_enabled = self.config.sandbox_os_enabled;
child.sandbox_escalation = self.config.sandbox_escalation;
child.sandbox_env_policy = self.config.sandbox_env_policy;
if let Some(def) = definition {
if let Some(allowed) = &def.tools {
for name in &self.config.core_tools_enabled {
if !allowed.iter().any(|t| t == name) {
child
.tool_overrides
.entry(name.clone())
.or_default()
.enabled = Some(false);
}
}
}
}
child.core_tools_enabled = self.config.core_tools_enabled.clone();
child.module_registry = self.config.module_registry;
child.module_activation = self.config.module_activation.clone();
child.subagents_enabled = self.config.subagents_enabled;
child.subagents_max_depth = self.config.subagents_max_depth;
child.subagents_max_concurrent = self.config.subagents_max_concurrent;
child.subagents_background = self.config.subagents_background;
child.subagents_background_prompts = self.config.subagents_background_prompts;
child.subagents_claude_agent_alias = self.config.subagents_claude_agent_alias;
child.subagents_definitions = self.config.subagents_definitions.clone();
child.subagent_depth = self.subagent_depth + 1;
child
}
async fn run_spawn_subagent(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
if !self.config.subagents_enabled {
let err = Error::UnknownTool(SPAWN_SUBAGENT.to_string());
return (format!("Error: {err}"), true);
}
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: SPAWN_SUBAGENT.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let task = args
.get("task")
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_string();
if task.is_empty() {
let err = Error::InvalidArguments {
tool: SPAWN_SUBAGENT.to_string(),
message: "`task` is required and must be non-empty".to_string(),
};
return (format!("Error: {err}"), true);
}
let agent_type = args
.get("agent_type")
.and_then(serde_json::Value::as_str)
.map(String::from);
let inline_system_prompt = args
.get("system_prompt")
.and_then(serde_json::Value::as_str)
.map(String::from);
let background = args
.get("background")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let requested_model = args
.get("model")
.and_then(serde_json::Value::as_str)
.map(|model| crate::model_catalog::resolve_alias(model, &[]));
let definition = match &agent_type {
Some(name) => match self.config.subagents_definitions.get(name) {
Some(d) => Some(d.clone()),
None => {
let err = Error::SubagentDefinitionNotFound(name.clone());
return (format!("Error: {err}"), true);
}
},
None => None,
};
if background {
if !self.config.subagents_background {
let err = Error::tool(
SPAWN_SUBAGENT,
"background=true requires capabilities.subagents.background = true",
);
return (format!("Error: {err}"), true);
}
if self.config.subagents_background_prompts.is_none() {
let err = Error::SubagentBackgroundPolicyMissing;
return (format!("Error: {err}"), true);
}
}
if let Err(e) =
crate::subagents::check_depth(self.subagent_depth, self.config.subagents_max_depth)
{
return (format!("Error: {e}"), true);
}
let Some(guard) = crate::subagents::try_acquire(
&self.subagent_concurrency_gauge,
self.config.subagents_max_concurrent,
) else {
let err = Error::SubagentConcurrencyExceeded {
max_concurrent: self.config.subagents_max_concurrent,
};
return (format!("Error: {err}"), true);
};
let child_id = next_subagent_id();
let child_config = self.build_child_config(
definition.as_ref(),
inline_system_prompt,
requested_model.or_else(|| definition.as_ref().and_then(|d| d.model.clone())),
);
let child_model = child_config.model.clone();
let mut child = Agent::with_provider_arc(child_config, self.provider.clone());
child.subagent_depth = self.subagent_depth + 1;
child.subagent_concurrency_gauge = self.subagent_concurrency_gauge.clone();
if background {
if let Some(crate::subagents::BackgroundPromptsPolicy::Parent) =
self.config.subagents_background_prompts
{
let handler: std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler> =
match &self.child_approval_handler_factory {
Some(factory) => {
factory(child_id.clone(), self.pending_child_approvals.clone())
}
None => std::sync::Arc::new(crate::subagents::ParentQueueApprovalHandler {
child_agent_id: child_id.clone(),
queue: self.pending_child_approvals.clone(),
}),
};
child.ctx.sandbox_approval_handler =
Some(crate::sandbox::SandboxApprovalHandler(handler.clone()));
child.permissions_approval_handler = Some(handler);
}
}
let lineage = crate::subagents::SubagentLineage {
child_agent_id: child_id.clone(),
parent_session_id: self.subagent_store.as_ref().map(|(_, name)| name.clone()),
parent_tool_use_id: call.id.clone(),
depth: self.subagent_depth + 1,
agent_type: agent_type.clone(),
task: task.clone(),
background,
spawned_at_ms: now_ms(),
model: child_model,
};
if let Some((store, parent_name)) = &self.subagent_store {
let _ = store.save_subagent_lineage(parent_name, &child_id, &lineage);
}
if background {
let spawned_task_text = task.clone();
self.background_subagents.insert(
child_id.clone(),
BackgroundSubagent {
handle: tokio::spawn(async move {
let _guard = guard;
let result = child.send(spawned_task_text).await;
let transcript = child.history()[1..].to_vec();
(child_id, result, transcript)
}),
task,
agent_type,
started_at_ms: lineage.spawned_at_ms,
},
);
let out = serde_json::json!({
"subagent_id": lineage.child_agent_id,
"status": "spawned",
"background": true,
});
return (out.to_string(), false);
}
let result = child.send(task).await;
let transcript = child.history()[1..].to_vec();
self.persist_subagent_transcript(&child_id, &lineage, &transcript);
drop(guard);
match result {
Ok(text) => (text, false),
Err(e) => (format!("Error: subagent `{child_id}` failed: {e}"), true),
}
}
async fn run_subagent_status(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: SUBAGENT_STATUS.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let Some(id) = args.get("subagent_id").and_then(serde_json::Value::as_str) else {
let err = Error::InvalidArguments {
tool: SUBAGENT_STATUS.to_string(),
message: "`subagent_id` is required".to_string(),
};
return (format!("Error: {err}"), true);
};
let Some(entry) = self.background_subagents.get(id) else {
let err = Error::SubagentNotFound(id.to_string());
return (format!("Error: {err}"), true);
};
if !entry.handle.is_finished() {
let out = serde_json::json!({
"subagent_id": id,
"status": "pending",
"task": entry.task,
"agent_type": entry.agent_type,
"started_at_ms": entry.started_at_ms,
});
return (out.to_string(), false);
}
let entry = self
.background_subagents
.remove(id)
.expect("checked Some above");
let (child_id, result, transcript) = match entry.handle.await {
Ok(v) => v,
Err(join_err) => {
let err = Error::tool(
SUBAGENT_STATUS,
format!("subagent `{id}` task panicked: {join_err}"),
);
return (format!("Error: {err}"), true);
}
};
if let Some((store, parent_name)) = self.subagent_store.clone() {
if let Ok(Some(lineage)) = store.load_subagent_lineage(&parent_name, &child_id) {
self.persist_subagent_transcript(&child_id, &lineage, &transcript);
}
}
match result {
Ok(text) => {
let out = serde_json::json!({
"subagent_id": child_id,
"status": "done",
"result": text,
});
(out.to_string(), false)
}
Err(e) => {
let out = serde_json::json!({
"subagent_id": child_id,
"status": "error",
"message": e.to_string(),
});
(out.to_string(), true)
}
}
}
fn run_background_exec(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: BACKGROUND_EXEC.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let command = args
.get("command")
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_string();
if command.is_empty() {
let err = Error::InvalidArguments {
tool: BACKGROUND_EXEC.to_string(),
message: "`command` is required and must be non-empty".to_string(),
};
return (format!("Error: {err}"), true);
}
let job_id = crate::background::next_job_id(now_ms());
if let Some(reason) = self.background_permission_denial(&command, &job_id) {
return (format!("Error: {reason}"), true);
}
if let Some(hook) = &self.config.pre_tool_hook {
if let Some(reason) = hook(BACKGROUND_EXEC, &args) {
return (format!("Error: blocked by pre-tool hook: {reason}"), true);
}
}
let Some(guard) = crate::subagents::try_acquire(
&self.background_concurrency_gauge,
self.config.tools_background_max_concurrent,
) else {
let err = Error::BackgroundJobConcurrencyExceeded {
max_concurrent: self.config.tools_background_max_concurrent,
};
return (format!("Error: {err}"), true);
};
let mut cmd = match crate::tools::build_sandboxed_sh(&command, &self.ctx) {
Ok(cmd) => cmd,
Err(e) => return (format!("Error: {e}"), true),
};
cmd.current_dir(&self.ctx.cwd)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.kill_on_drop(true);
#[cfg(unix)]
cmd.process_group(0);
let mut child = match cmd.spawn() {
Ok(c) => c,
Err(e) => {
drop(guard);
let err = Error::tool(
BACKGROUND_EXEC,
format!("failed to spawn background command: {e}"),
);
return (format!("Error: {err}"), true);
}
};
let pid = child.id();
let output = std::sync::Arc::new(crate::background::CapturedOutput::new());
let cap = self.config.tools_background_max_output_bytes;
if let Some(stdout) = child.stdout.take() {
let _stdout_reader = spawn_output_reader(stdout, output.clone(), cap);
}
if let Some(stderr) = child.stderr.take() {
let _stderr_reader = spawn_output_reader(stderr, output.clone(), cap);
}
let started_at_ms = now_ms();
self.background_jobs.insert(
job_id.clone(),
BackgroundJob {
child,
command: command.clone(),
pid,
output,
started_at_ms,
killed: false,
_guard: guard,
},
);
let out = serde_json::json!({
"job_id": job_id,
"status": "running",
"pid": pid,
"command": command,
});
(out.to_string(), false)
}
fn run_background_status(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: BACKGROUND_STATUS.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let Some(job_id) = args.get("job_id").and_then(serde_json::Value::as_str) else {
let err = Error::InvalidArguments {
tool: BACKGROUND_STATUS.to_string(),
message: "`job_id` is required".to_string(),
};
return (format!("Error: {err}"), true);
};
let job_id = job_id.to_string();
let (command, pid, started_at_ms, status, output_so_far, truncated, delta) = {
let Some(job) = self.background_jobs.get_mut(&job_id) else {
let err = Error::BackgroundJobNotFound(job_id);
return (format!("Error: {err}"), true);
};
let status = background_job_status(job);
let (output_so_far, truncated) = job.output.snapshot();
let delta = job.output.drain_new();
(
job.command.clone(),
job.pid,
job.started_at_ms,
status,
output_so_far,
truncated,
delta,
)
};
if !delta.is_empty() {
self.emit(AgentEvent::BackgroundOutput {
job_id: job_id.clone(),
chunk: delta,
truncated,
});
}
let exit_code = match status {
crate::background::JobStatus::Exited(code) => code,
_ => None,
};
let out = serde_json::json!({
"job_id": job_id,
"command": command,
"status": status.as_str(),
"exit_code": exit_code,
"pid": pid,
"started_at_ms": started_at_ms,
"output": output_so_far,
"output_truncated": truncated,
});
if !matches!(status, crate::background::JobStatus::Running) {
self.background_jobs.remove(&job_id);
}
(out.to_string(), false)
}
fn run_background_list(&mut self, _call: &crate::message::ToolCall) -> (String, bool) {
let mut jobs = Vec::new();
for (job_id, job) in self.background_jobs.iter_mut() {
let status = background_job_status(job);
jobs.push(serde_json::json!({
"job_id": job_id,
"command": job.command,
"status": status.as_str(),
"pid": job.pid,
"started_at_ms": job.started_at_ms,
}));
}
let out = serde_json::json!({ "jobs": jobs });
(out.to_string(), false)
}
fn run_background_kill(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: BACKGROUND_KILL.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let Some(job_id) = args.get("job_id").and_then(serde_json::Value::as_str) else {
let err = Error::InvalidArguments {
tool: BACKGROUND_KILL.to_string(),
message: "`job_id` is required".to_string(),
};
return (format!("Error: {err}"), true);
};
let job_id = job_id.to_string();
let Some(mut job) = self.background_jobs.remove(&job_id) else {
let err = Error::BackgroundJobNotFound(job_id);
return (format!("Error: {err}"), true);
};
kill_job_process_group(&mut job);
job.killed = true;
let out = serde_json::json!({
"job_id": job_id,
"status": "killed",
"pid": job.pid,
});
(out.to_string(), false)
}
fn persist_subagent_transcript(
&self,
child_id: &str,
lineage: &crate::subagents::SubagentLineage,
transcript: &[ChatMessage],
) {
let Some((store, parent_name)) = &self.subagent_store else {
return;
};
let mut session = match Session::from_claude_code_str("") {
Ok(s) => s,
Err(_) => return,
};
session.meta.source = crate::session::SessionSource::Native;
session.meta.agent_id = Some(lineage.child_agent_id.clone());
session.meta.parent_tool_use_id = Some(lineage.parent_tool_use_id.clone());
session.meta.lineage = lineage.to_lineage_map();
let sidecar_jsonl = session.to_native_jsonl_v2(transcript);
let _ = store.save_subagent_transcript(parent_name, child_id, &sidecar_jsonl);
let _ = store.save_subagent_lineage(parent_name, child_id, lineage);
}
fn run_tool_search(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: TOOL_SEARCH.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let query = args
.get("query")
.and_then(serde_json::Value::as_str)
.unwrap_or("")
.to_lowercase();
let max_results = args
.get("max_results")
.and_then(serde_json::Value::as_u64)
.map(|n| n as usize);
let mut matches: Vec<ToolSchema> = self
.registry
.iter()
.filter(|t| self.config.tool_enabled(t.name()))
.filter(|t| !self.is_core_tool(t.name()))
.filter(|t| !self.activated_tools.contains(t.name()))
.filter(|t| {
query.is_empty()
|| t.name().to_lowercase().contains(&query)
|| self
.config
.tool_description(t.name(), t.description())
.to_lowercase()
.contains(&query)
})
.map(|t| self.raw_schema_for(t))
.collect();
if let Some(max) = max_results {
matches.truncate(max);
}
for m in &matches {
self.activated_tools.insert(m.name.clone());
}
let result = serde_json::to_string(&matches).unwrap_or_else(|_| "[]".to_string());
(result, false)
}
fn expand_reduction_schema() -> ToolSchema {
ToolSchema {
name: EXPAND_REDUCTION.to_string(),
description: "Fetch back the original content hidden behind a reduction stub in \
your current view — a truncated tool output, cleared old turns, or an elided \
file read that was hidden to save context. Each stub line names a reduction id \
like r0042-9f3c: pass ONLY that id here, and never quote or repeat a stub line \
itself in your replies. The original is durably kept in the session sidecar. \
Pass `byte_range` to fetch a slice of a large one at a time instead of all of \
it at once; ranged results are prefixed with a `bytes start..end of total` \
header so you can plan the next slice."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"reduction_id": {
"type": "string",
"description": "The reduction id named in the stub line, e.g. \
\"r0042-9f3c\". Pass the id alone."
},
"byte_range": {
"type": "array",
"items": {"type": "integer"},
"minItems": 2,
"maxItems": 2,
"description": "Optional [start, end) byte offsets within the original \
content to fetch instead of all of it. Exactly two non-negative \
integers with start <= end."
}
},
"required": ["reduction_id"],
"additionalProperties": false
}),
}
}
fn sidecar_search_schema() -> ToolSchema {
ToolSchema {
name: SIDECAR_SEARCH.to_string(),
description: "Search content currently hidden from your view by reduction stubs \
(large tool outputs, cleared old turns, elided file reads) for a substring or \
regex. Only hidden content is searched, never what you can already see. \
Returns match snippets with each match's reduction_id for use with \
expand_reduction; refer to results by their reduction id rather than quoting \
stub lines. Results are capped — if `truncated` is true, narrow the query."
.to_string(),
parameters: serde_json::json!({
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Non-empty substring or regex to search for \
(case-insensitive)."
}
},
"required": ["query"],
"additionalProperties": false
}),
}
}
fn recorded_messages(&self) -> std::result::Result<Option<Vec<ChatMessage>>, String> {
let Some(recorder) = &self.recorder else {
return Ok(None);
};
let raw = std::fs::read_to_string(recorder.path())
.map_err(|e| format!("failed to read the session sidecar: {e}"))?;
let session = Session::from_sidecar_str(&raw)
.map_err(|e| format!("failed to parse the session sidecar: {e}"))?;
Ok(Some(session.messages))
}
fn run_expand_reduction(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: EXPAND_REDUCTION.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let Some(id) = args.get("reduction_id").and_then(serde_json::Value::as_str) else {
return (
"Error: expand_reduction requires a `reduction_id` string argument".to_string(),
true,
);
};
let recorded = match self.recorded_messages() {
Ok(r) => r,
Err(e) => return (format!("Error: expand_reduction: {e}"), true),
};
let recorded = recorded.as_deref();
let byte_range = match args.get("byte_range") {
None | Some(serde_json::Value::Null) => None,
Some(v) => {
let parsed = v
.as_array()
.filter(|a| a.len() == 2)
.and_then(|a| Some((a[0].as_u64()? as usize, a[1].as_u64()? as usize)));
match parsed {
Some(range) => Some(range),
None => {
let total = reduce::rehydrate::reduction_total_bytes(
&self.reduction_log,
&self.history[1..],
recorded,
id,
)
.map(|n| format!("; the original is {n} bytes"))
.unwrap_or_default();
return (
format!(
"Error: expand_reduction: malformed byte_range {v} — expected \
[start, end): exactly two non-negative integers with \
start <= end{total}"
),
true,
);
}
}
}
};
match reduce::rehydrate::expand_reduction(
&self.reduction_log,
&self.history[1..],
recorded,
id,
byte_range,
) {
Ok(outcome) => match outcome.range {
Some((start, end)) => (
format!(
"[{id}: bytes {start}..{end} of {total}]\n{content}",
total = outcome.total_bytes,
content = outcome.content
),
false,
),
None => (outcome.content, false),
},
Err(e) => (format!("Error: {e}"), true),
}
}
fn run_sidecar_search(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
let args = match call.function.parsed_arguments() {
Ok(v) => v,
Err(e) => {
let err = Error::InvalidArguments {
tool: SIDECAR_SEARCH.to_string(),
message: e.to_string(),
};
return (format!("Error: {err}"), true);
}
};
let query = args
.get("query")
.and_then(serde_json::Value::as_str)
.unwrap_or("");
if query.trim().is_empty() {
return (
"Error: sidecar_search requires a non-empty `query` string argument".to_string(),
true,
);
}
let recorded = match self.recorded_messages() {
Ok(r) => r,
Err(e) => return (format!("Error: sidecar_search: {e}"), true),
};
match reduce::rehydrate::sidecar_search(
&self.reduction_log,
&self.history[1..],
recorded.as_deref(),
query,
) {
Ok(result) => (
serde_json::to_string(&result).unwrap_or_else(|_| "{}".to_string()),
false,
),
Err(e) => (format!("Error: {e}"), true),
}
}
fn emit(&self, event: AgentEvent) {
if let Some(sink) = &self.config.event_sink {
sink(event);
}
}
pub fn turn_count(&self) -> usize {
self.history
.iter()
.filter(|m| m.role != Role::System)
.count()
}
pub fn total_output_tokens(&self) -> u64 {
self.total_output_tokens
}
}
#[cfg(test)]
mod api_key_cmd_tests {
use super::*;
#[test]
fn default_none_falls_through_to_missing_api_key_error() {
let config = Config::builder()
.api_key_env("SUPERCODE_TEST_UNSET_VAR_API_KEY_CMD")
.build();
assert!(config.api_key.is_none());
assert!(config.api_key_cmd.is_none());
let err = Agent::new(config).err().expect("no key source configured");
assert!(matches!(err, Error::MissingApiKey(_)));
}
#[test]
fn api_key_cmd_alone_resolves_successfully() {
let config = Config::builder()
.api_key_cmd("echo sk-test-from-helper")
.api_key_env("SUPERCODE_TEST_UNSET_VAR_API_KEY_CMD_2")
.build();
assert!(Agent::new(config).is_ok());
}
#[test]
fn api_key_cmd_failure_falls_through_to_env() {
std::env::set_var(
"SUPERCODE_TEST_API_KEY_CMD_FALLBACK",
"sk-from-env-fallback",
);
let config = Config::builder()
.api_key_cmd("exit 1")
.api_key_env("SUPERCODE_TEST_API_KEY_CMD_FALLBACK")
.build();
assert!(Agent::new(config).is_ok());
std::env::remove_var("SUPERCODE_TEST_API_KEY_CMD_FALLBACK");
}
#[test]
fn api_key_cmd_failure_with_no_fallback_still_errors() {
let config = Config::builder()
.api_key_cmd("exit 1")
.api_key_env("SUPERCODE_TEST_UNSET_VAR_API_KEY_CMD_3")
.build();
let err = Agent::new(config)
.err()
.expect("helper failed, no env fallback");
assert!(matches!(err, Error::MissingApiKey(_)));
}
#[test]
fn run_api_key_cmd_trims_output() {
assert_eq!(run_api_key_cmd("echo ' sk-abc123 '"), "sk-abc123");
}
#[test]
fn run_api_key_cmd_spawn_failure_returns_empty() {
assert_eq!(
run_api_key_cmd("/no/such/binary/at/all --flag"),
String::new()
);
}
}