use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
#[cfg(unix)]
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64_STD};
use serde::Deserialize;
use serde_json::json;
use lingshu_types::{ToolError, ToolSchema};
#[cfg(unix)]
use crate::describe_execution_filesystem;
#[cfg(unix)]
use crate::execution_tmp::{BACKEND_TMP_ROOT, wrap_command_with_tmp_env};
use crate::execution_tmp::{ensure_default_shared_tmp_dir, temp_env_pairs};
#[cfg(unix)]
use crate::registry::ToolRegistry;
use crate::registry::{ToolContext, ToolHandler};
#[cfg(unix)]
use crate::tools::backend_pool::get_or_create_backend;
#[cfg(unix)]
use crate::tools::backends;
use crate::tools::backends::redact_output;
const DEFAULT_TIMEOUT_SECS: u64 = 300;
const MAX_STDOUT_BYTES: usize = 50_000;
const MAX_STDERR_BYTES: usize = 10_000;
#[cfg(unix)]
const MAX_TOOL_CALLS: usize = 50;
#[cfg(unix)]
const SANDBOX_ALLOWED_TOOLS: &[&str] = &[
"web_search",
"web_extract",
"web_crawl",
"read_file",
"pdf_to_markdown",
"write_file",
"search_files",
"patch",
"terminal",
"vision_analyze",
"transcribe_audio",
"session_search",
];
#[cfg(unix)]
const TERMINAL_BLOCKED_PARAMS: &[&str] = &["background", "check_interval", "pty", "watch_patterns"];
pub struct ExecuteCodeToolReal;
#[derive(Deserialize)]
struct ExecuteCodeArgs {
#[serde(default = "default_language")]
language: String,
code: String,
#[serde(default)]
timeout: Option<u64>,
}
fn default_language() -> String {
"python".to_string()
}
fn resolve_runtime(lang: &str) -> Option<(&'static str, &'static str)> {
match lang.to_lowercase().as_str() {
"python" | "python3" | "py" => Some(("python3", ".py")),
"javascript" | "js" | "node" => Some(("node", ".js")),
"typescript" | "ts" => Some(("npx tsx", ".ts")),
"bash" | "sh" | "shell" => Some(("bash", ".sh")),
"ruby" | "rb" => Some(("ruby", ".rb")),
"perl" | "pl" => Some(("perl", ".pl")),
"rust" | "rs" => Some(("rustc_run", ".rs")), _ => None,
}
}
#[cfg(unix)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum RpcTransport {
Uds,
File,
}
#[cfg(unix)]
fn generate_tools_module(available_tools: &[&str], transport: RpcTransport) -> String {
let mut stubs = String::new();
let available_set: std::collections::HashSet<&str> = available_tools.iter().copied().collect();
for &tool in SANDBOX_ALLOWED_TOOLS {
if !available_set.contains(tool) {
stubs.push_str(&format!(
r#"
def {tool}(*args, **kwargs):
raise RuntimeError("Tool '{tool}' is not available in this execute_code session. Available: " + ", ".join(AVAILABLE_TOOLS))
"#
));
continue;
}
let stub = match tool {
"web_search" => {
r#"
def web_search(query: str, max_results: int = 5, backend: str = None):
"""Search the web. Returns {"query", "backend", "results":[{"url","title","description"}]}."""
args = {"query": query, "max_results": max_results}
if backend is not None:
args["backend"] = backend
return _call("web_search", args)
"#
}
"web_extract" => {
r#"
def web_extract(url: str, max_chars: int = 8000, backend: str = None, render_js_fallback: bool = True):
"""Extract one URL. Returns {"backend","result":{"url","title","content","extractor","content_type","content_format"}}."""
args = {"url": url, "max_chars": max_chars, "render_js_fallback": render_js_fallback}
if backend is not None:
args["backend"] = backend
return _call("web_extract", args)
"#
}
"web_crawl" => {
r#"
def web_crawl(url: str, instructions: str = None, max_pages: int = 8, max_depth: int = 2, max_chars_per_page: int = 4000, same_path_only: bool = False, backend: str = None, render_js_fallback: bool = True):
"""Recursively crawl a website starting from a URL.
Returns {"success", "backend", "pages_visited", "results":[{"url","title","depth","content","extractor","content_type","content_format"}]}.
Use instructions to focus on specific content (e.g. 'find API docs').
Prefer this over looping web_extract when you need multiple linked pages.
"""
args = {"url": url, "max_pages": max_pages, "max_depth": max_depth,
"max_chars_per_page": max_chars_per_page, "same_path_only": same_path_only,
"render_js_fallback": render_js_fallback}
if instructions is not None:
args["instructions"] = instructions
if backend is not None:
args["backend"] = backend
return _call("web_crawl", args)
"#
}
"read_file" => {
r#"
def read_file(path: str, offset: int = 1, limit: int = 500):
"""Read a file (1-indexed lines). Returns dict with "content" and "total_lines"."""
return _call("read_file", {"path": path, "offset": offset, "limit": limit})
"#
}
"pdf_to_markdown" => {
r#"
def pdf_to_markdown(path: str, output_path: str = None, max_chars: int = 20000):
"""Convert a local PDF file to Markdown using EdgeParse.
This is fast structural PDF parsing, not OCR.
Returns {"success", "path", "output_path", "extractor", "parsing_mode", "content_format", "truncated", "total_chars", "markdown"}.
"""
args = {"path": path, "max_chars": max_chars}
if output_path is not None:
args["output_path"] = output_path
return _call("pdf_to_markdown", args)
"#
}
"write_file" => {
r#"
def write_file(path: str, content: str):
"""Write content to a file (always overwrites). Returns dict with status."""
return _call("write_file", {"path": path, "content": content})
"#
}
"search_files" => {
r#"
def search_files(pattern: str, target: str = "content", path: str = ".", file_glob: str = None, limit: int = 50, offset: int = 0, output_mode: str = "content", context: int = 0):
"""Search file contents (target="content") or find files by name (target="files"). Returns dict with "matches"."""
return _call("search_files", {"pattern": pattern, "target": target, "path": path, "file_glob": file_glob, "limit": limit, "offset": offset, "output_mode": output_mode, "context": context})
"#
}
"patch" => {
r#"
def patch(path: str = None, old_string: str = None, new_string: str = None, replace_all: bool = False, mode: str = "replace", patch: str = None):
"""Targeted find-and-replace (mode="replace") or V4A multi-file patches (mode="patch"). Returns dict with status."""
return _call("patch", {"path": path, "old_string": old_string, "new_string": new_string, "replace_all": replace_all, "mode": mode, "patch": patch})
"#
}
"terminal" => {
r#"
def terminal(command: str, timeout: int = None, workdir: str = None):
"""Run a shell command (foreground only). Returns dict with "output" and "exit_code"."""
return _call("terminal", {"command": command, "timeout": timeout, "workdir": workdir})
"#
}
"vision_analyze" => {
r#"
def vision_analyze(image_source: str, question: str = "Describe this image in detail.", detail: str = "high"):
"""Analyze a local image file or https:// image URL using the vision LLM.
image_source: absolute file path (png/jpg/webp/…) or https:// URL.
Returns dict with "content" containing the model's description.
Use this for any local image path or remote image URL.
Do NOT use this for browser screenshots — use browser_vision instead.
"""
return _call("vision_analyze", {"image_source": image_source, "question": question, "detail": detail})
"#
}
"transcribe_audio" => {
r#"
def transcribe_audio(file_path: str, provider: str = None, model: str = None, language: str = "en"):
"""Transcribe speech from an audio file to text.
file_path: path to audio file (mp3, mp4, m4a, wav, webm, ogg, etc.).
provider: 'local' (default, free), 'groq', or 'openai'.
Returns dict with "text" containing the transcription.
"""
args = {"file_path": file_path, "language": language}
if provider is not None:
args["provider"] = provider
if model is not None:
args["model"] = model
return _call("transcribe_audio", args)
"#
}
"session_search" => {
r#"
def session_search(query: str, limit: int = 10):
"""Search past conversation sessions using full-text search.
Returns dict with "results" list of {session_id, snippet, timestamp}.
Use this to retrieve context from earlier conversations.
"""
return _call("session_search", {"query": query, "limit": limit})
"#
}
_ => continue,
};
stubs.push_str(stub);
}
let header = match transport {
RpcTransport::Uds => {
r#""""Auto-generated Lingshu tools RPC stubs."""
import json, os, socket, shlex, time
_sock = None
# ---------------------------------------------------------------------------
# Convenience helpers (avoid common scripting pitfalls)
# ---------------------------------------------------------------------------
def json_parse(text: str):
"""Parse JSON tolerant of control characters (strict=False).
Use this instead of json.loads() when parsing output from terminal()
or web_extract() that may contain raw tabs/newlines in strings."""
return json.loads(text, strict=False)
def shell_quote(s: str) -> str:
"""Shell-escape a string for safe interpolation into commands.
Use this when inserting dynamic content into terminal() commands:
terminal(f"echo {shell_quote(user_input)}")
"""
return shlex.quote(s)
def retry(fn, max_attempts=3, delay=2):
"""Retry a function up to max_attempts times with exponential backoff.
Use for transient failures (network errors, API rate limits):
result = retry(lambda: terminal("gh issue list ..."))
"""
last_err = None
for attempt in range(max_attempts):
try:
return fn()
except Exception as e:
last_err = e
if attempt < max_attempts - 1:
time.sleep(delay * (2 ** attempt))
raise last_err
def _connect():
global _sock
if _sock is None:
_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
_sock.connect(os.environ["EDGECRAB_RPC_SOCKET"])
_sock.settimeout(300)
return _sock
def _call(tool_name, args):
"""Send a tool call to the parent process and return the parsed result."""
conn = _connect()
request = json.dumps({"tool": tool_name, "args": args}) + "\n"
conn.sendall(request.encode())
buf = b""
while True:
chunk = conn.recv(65536)
if not chunk:
raise RuntimeError("Agent process disconnected")
buf += chunk
if buf.endswith(b"\n"):
break
raw = buf.decode().strip()
result = json.loads(raw)
if isinstance(result, str):
try:
return json.loads(result)
except (json.JSONDecodeError, TypeError):
return result
return result
"#
}
RpcTransport::File => {
r#""""Auto-generated Lingshu tools RPC stubs (file transport)."""
import json, os, shlex, time
_seq = 0
_RPC_DIR = os.environ.get("EDGECRAB_RPC_DIR", "/tmp/lingshu_rpc")
# ---------------------------------------------------------------------------
# Convenience helpers (avoid common scripting pitfalls)
# ---------------------------------------------------------------------------
def json_parse(text: str):
"""Parse JSON tolerant of control characters (strict=False).
Use this instead of json.loads() when parsing output from terminal()
or web_extract() that may contain raw tabs/newlines in strings."""
return json.loads(text, strict=False)
def shell_quote(s: str) -> str:
"""Shell-escape a string for safe interpolation into commands.
Use this when inserting dynamic content into terminal() commands:
terminal(f"echo {shell_quote(user_input)}")
"""
return shlex.quote(s)
def retry(fn, max_attempts=3, delay=2):
"""Retry a function up to max_attempts times with exponential backoff.
Use for transient failures (network errors, API rate limits):
result = retry(lambda: terminal("gh issue list ..."))
"""
last_err = None
for attempt in range(max_attempts):
try:
return fn()
except Exception as e:
last_err = e
if attempt < max_attempts - 1:
time.sleep(delay * (2 ** attempt))
raise last_err
def _call(tool_name, args):
"""Send a tool call to the parent process and wait for a response file."""
global _seq
_seq += 1
seq_str = f"{_seq:06d}"
req_file = os.path.join(_RPC_DIR, f"req_{seq_str}")
res_file = os.path.join(_RPC_DIR, f"res_{seq_str}")
tmp = req_file + ".tmp"
with open(tmp, "w") as f:
json.dump({"tool": tool_name, "args": args, "seq": _seq}, f)
os.rename(tmp, req_file)
deadline = time.monotonic() + 300
poll_interval = 0.05
while not os.path.exists(res_file):
if time.monotonic() > deadline:
raise RuntimeError(f"RPC timeout: no response for {tool_name} after 300s")
time.sleep(poll_interval)
poll_interval = min(poll_interval * 1.2, 0.25)
with open(res_file) as f:
raw = f.read()
try:
os.unlink(res_file)
except OSError:
pass
result = json.loads(raw)
if isinstance(result, str):
try:
return json.loads(result)
except (json.JSONDecodeError, TypeError):
return result
return result
"#
}
};
format!("{header}\nAVAILABLE_TOOLS = {available_tools:?}\n\n{stubs}\n")
}
#[cfg(unix)]
async fn rpc_server_loop(
listener: tokio::net::UnixListener,
registry: Arc<ToolRegistry>,
ctx: ToolContext,
tool_call_counter: Arc<std::sync::atomic::AtomicUsize>,
allowed_tools: Arc<Vec<String>>,
) {
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
let accept_result = tokio::time::timeout(Duration::from_secs(5), listener.accept()).await;
let stream = match accept_result {
Ok(Ok((stream, _))) => stream,
_ => return,
};
let (reader, mut writer) = stream.into_split();
let mut lines = BufReader::new(reader).lines();
while let Ok(Some(line)) = lines.next_line().await {
let line = line.trim().to_string();
if line.is_empty() {
continue;
}
let request: serde_json::Value = match serde_json::from_str(&line) {
Ok(v) => v,
Err(e) => {
let resp = json!({"error": format!("Invalid RPC request: {e}")});
let _ = writer.write_all(format!("{}\n", resp).as_bytes()).await;
continue;
}
};
let tool_name = request
.get("tool")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let mut tool_args = request.get("args").cloned().unwrap_or(json!({}));
if !allowed_tools.contains(&tool_name) {
let available = allowed_tools.join(", ");
let resp = json!({
"error": format!(
"Tool '{}' is not available in execute_code. Available: {}",
tool_name, available
)
});
let _ = writer.write_all(format!("{}\n", resp).as_bytes()).await;
continue;
}
let count = tool_call_counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if count >= MAX_TOOL_CALLS {
let resp = json!({
"error": format!(
"Tool call limit reached ({}). No more tool calls allowed.",
MAX_TOOL_CALLS
)
});
let _ = writer.write_all(format!("{}\n", resp).as_bytes()).await;
continue;
}
if tool_name == "terminal"
&& let Some(obj) = tool_args.as_object_mut()
{
for param in TERMINAL_BLOCKED_PARAMS {
obj.remove(*param);
}
}
let result = match registry.dispatch(&tool_name, tool_args, &ctx).await {
Ok(output) => output,
Err(e) => json!({"error": e.to_string()}).to_string(),
};
let _ = writer
.write_all(format!("{}\n", rpc_response_payload(result)).as_bytes())
.await;
}
}
#[cfg(unix)]
fn rpc_response_payload(result: String) -> String {
if serde_json::from_str::<serde_json::Value>(&result).is_ok() {
result
} else {
json!(result).to_string()
}
}
#[cfg(unix)]
fn maybe_rewrite_terminal_args(
tool_name: &str,
tool_args: &mut serde_json::Value,
default_remote_workdir: Option<&str>,
) {
if tool_name != "terminal" {
return;
}
let Some(obj) = tool_args.as_object_mut() else {
return;
};
for param in TERMINAL_BLOCKED_PARAMS {
obj.remove(*param);
}
if let Some(workdir) = default_remote_workdir {
let has_workdir = obj
.get("workdir")
.and_then(|value| value.as_str())
.is_some_and(|value| !value.is_empty());
if !has_workdir && let Some(command) = obj.get("command").and_then(|value| value.as_str()) {
obj.insert(
"command".into(),
json!(format!(
"cd {} && {}",
backends::shell_quote(workdir),
command
)),
);
}
}
}
#[cfg(unix)]
struct RemoteRpcLoop {
backend: Arc<dyn backends::ExecutionBackend>,
registry: Arc<ToolRegistry>,
ctx: ToolContext,
rpc_dir: String,
tool_call_counter: Arc<std::sync::atomic::AtomicUsize>,
allowed_tools: Arc<Vec<String>>,
stop: tokio_util::sync::CancellationToken,
terminal_default_workdir: Option<String>,
}
#[cfg(unix)]
async fn remote_rpc_poll_loop(loop_ctx: RemoteRpcLoop) {
while !loop_ctx.stop.is_cancelled() {
let pending = match loop_ctx
.backend
.execute_oneshot(
&format!(
"ls -1 {} 2>/dev/null | grep '^req_' || true",
backends::shell_quote(&loop_ctx.rpc_dir)
),
".",
Duration::from_secs(10),
loop_ctx.stop.clone(),
)
.await
{
Ok(output) if output.exit_code == 0 => output
.stdout
.lines()
.map(str::trim)
.filter(|line| !line.is_empty() && !line.ends_with(".tmp"))
.map(|line| format!("{}/{}", loop_ctx.rpc_dir, line))
.collect::<Vec<_>>(),
Ok(_) => Vec::new(),
Err(_) => Vec::new(),
};
if pending.is_empty() {
tokio::select! {
_ = loop_ctx.stop.cancelled() => break,
_ = tokio::time::sleep(Duration::from_millis(100)) => {}
}
continue;
}
for req_file in pending {
if loop_ctx.stop.is_cancelled() {
break;
}
let read = match loop_ctx
.backend
.execute_oneshot(
&format!("cat {}", backends::shell_quote(&req_file)),
".",
Duration::from_secs(10),
loop_ctx.stop.clone(),
)
.await
{
Ok(output) if output.exit_code == 0 => output.stdout,
_ => {
let _ = loop_ctx
.backend
.execute_oneshot(
&format!("rm -f {}", backends::shell_quote(&req_file)),
".",
Duration::from_secs(5),
tokio_util::sync::CancellationToken::new(),
)
.await;
continue;
}
};
let request: serde_json::Value = match serde_json::from_str(read.trim()) {
Ok(value) => value,
Err(_) => {
let _ = loop_ctx
.backend
.execute_oneshot(
&format!("rm -f {}", backends::shell_quote(&req_file)),
".",
Duration::from_secs(5),
tokio_util::sync::CancellationToken::new(),
)
.await;
continue;
}
};
let tool_name = request
.get("tool")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let seq = request
.get("seq")
.and_then(|v| v.as_u64())
.unwrap_or_default();
let mut tool_args = request.get("args").cloned().unwrap_or(json!({}));
maybe_rewrite_terminal_args(
&tool_name,
&mut tool_args,
loop_ctx.terminal_default_workdir.as_deref(),
);
let result = if !loop_ctx.allowed_tools.contains(&tool_name) {
let available = loop_ctx.allowed_tools.join(", ");
json!({
"error": format!(
"Tool '{}' is not available in execute_code. Available: {}",
tool_name, available
)
})
.to_string()
} else if loop_ctx
.tool_call_counter
.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
>= MAX_TOOL_CALLS
{
json!({
"error": format!(
"Tool call limit reached ({}). No more tool calls allowed.",
MAX_TOOL_CALLS
)
})
.to_string()
} else {
match loop_ctx
.registry
.dispatch(&tool_name, tool_args, &loop_ctx.ctx)
.await
{
Ok(output) => output,
Err(e) => json!({"error": e.to_string()}).to_string(),
}
};
let res_file = format!("{}/res_{seq:06}", loop_ctx.rpc_dir);
let _ = write_remote_file(
&*loop_ctx.backend,
&res_file,
&rpc_response_payload(result),
loop_ctx.stop.clone(),
)
.await;
let _ = loop_ctx
.backend
.execute_oneshot(
&format!("rm -f {}", backends::shell_quote(&req_file)),
".",
Duration::from_secs(5),
tokio_util::sync::CancellationToken::new(),
)
.await;
}
}
}
const SAFE_ENV_PREFIXES: &[&str] = &[
"PATH",
"HOME",
"USER",
"LANG",
"LC_",
"TERM",
"TMPDIR",
"TMP",
"TEMP",
"SHELL",
"LOGNAME",
"XDG_",
"PYTHONPATH",
"VIRTUAL_ENV",
"CONDA",
];
const SECRET_SUBSTRINGS: &[&str] = &[
"KEY",
"TOKEN",
"SECRET",
"PASSWORD",
"CREDENTIAL",
"PASSWD",
"AUTH",
];
fn build_child_env(sock_path: &str, cwd: &std::path::Path) -> HashMap<String, String> {
let mut env = HashMap::new();
let tmp_root = ensure_default_shared_tmp_dir()
.unwrap_or_else(|_| crate::execution_tmp::default_shared_tmp_dir());
for (k, v) in std::env::vars() {
let upper = k.to_uppercase();
if SECRET_SUBSTRINGS.iter().any(|s| upper.contains(s)) {
continue;
}
if SAFE_ENV_PREFIXES.iter().any(|p| upper.starts_with(p)) {
env.insert(k, v);
}
}
env.insert("EDGECRAB_RPC_SOCKET".into(), sock_path.into());
env.insert("PYTHONDONTWRITEBYTECODE".into(), "1".into());
for (k, v) in temp_env_pairs(&tmp_root.to_string_lossy()) {
env.insert(k, v);
}
if let Ok(tz) = std::env::var("EDGECRAB_TIMEZONE")
&& !tz.is_empty()
{
env.insert("TZ".into(), tz);
}
let cwd_str = cwd.to_string_lossy().to_string();
if let Some(existing) = env.get("PYTHONPATH") {
env.insert("PYTHONPATH".into(), format!("{cwd_str}:{existing}"));
} else {
env.insert("PYTHONPATH".into(), cwd_str);
}
env
}
#[cfg(unix)]
async fn write_remote_file(
backend: &dyn backends::ExecutionBackend,
path: &str,
content: &str,
cancel: tokio_util::sync::CancellationToken,
) -> Result<(), ToolError> {
let encoded = BASE64_STD.encode(content.as_bytes());
let scratch_path = format!("{path}.b64");
let parent = std::path::Path::new(path)
.parent()
.map(|p| p.to_string_lossy().into_owned())
.filter(|p| !p.is_empty())
.unwrap_or_else(|| ".".into());
backend
.execute_oneshot(
&format!(
"mkdir -p {} && : > {}",
backends::shell_quote(&parent),
backends::shell_quote(&scratch_path)
),
".",
Duration::from_secs(10),
cancel.clone(),
)
.await?;
const CHUNK_BYTES: usize = 24_576;
for chunk in encoded.as_bytes().chunks(CHUNK_BYTES) {
let chunk = std::str::from_utf8(chunk).map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("Internal base64 encoding error: {e}"),
})?;
backend
.execute_oneshot(
&format!(
"printf '%s' {} >> {}",
backends::shell_quote(chunk),
backends::shell_quote(&scratch_path)
),
".",
Duration::from_secs(30),
cancel.clone(),
)
.await?;
}
let decode = backend
.execute_oneshot(
&format!(
"(base64 -d < {} > {} || base64 -D -i {} -o {}) && rm -f {}",
backends::shell_quote(&scratch_path),
backends::shell_quote(path),
backends::shell_quote(&scratch_path),
backends::shell_quote(path),
backends::shell_quote(&scratch_path)
),
".",
Duration::from_secs(30),
cancel,
)
.await?;
if decode.exit_code != 0 {
return Err(ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!(
"Failed to materialize remote file {}: {}",
path,
decode.format(4_000, 2_000)
),
});
}
Ok(())
}
#[cfg(unix)]
async fn prepare_remote_sandbox(
backend: &dyn backends::ExecutionBackend,
sandbox_id: &str,
cancel: tokio_util::sync::CancellationToken,
) -> Result<String, ToolError> {
let command = format!(
"tmp_root=\"${{TMPDIR:-/tmp}}\"; dir=\"$tmp_root/lingshu_exec_{}\"; mkdir -p \"$dir/rpc\" && printf '%s' \"$dir\"",
sandbox_id
);
let output = backend
.execute_oneshot(&command, ".", Duration::from_secs(15), cancel)
.await?;
if output.exit_code != 0 {
return Err(ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!(
"Failed to prepare remote sandbox directory: {}",
output.format(4_000, 2_000)
),
});
}
let path = output.stdout.trim().to_string();
if path.is_empty() {
return Err(ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: "Remote backend returned an empty sandbox directory path.".into(),
});
}
Ok(path)
}
#[cfg(unix)]
fn remote_command_for_runtime(
runtime: &str,
script_name: &str,
rpc_dir: Option<&str>,
tmp_root: &str,
) -> String {
let env_prefix = if let Some(rpc_dir) = rpc_dir {
let mut out = format!(
"EDGECRAB_RPC_DIR={} PYTHONDONTWRITEBYTECODE=1",
backends::shell_quote(rpc_dir)
);
if let Ok(tz) = std::env::var("EDGECRAB_TIMEZONE")
&& !tz.trim().is_empty()
{
out.push(' ');
out.push_str("TZ=");
out.push_str(&backends::shell_quote(tz.trim()));
}
out
} else {
String::new()
};
let command = match runtime {
"rustc_run" => format!(
"rustc {} -o script_bin && ./script_bin",
backends::shell_quote(script_name)
),
value if value.contains(' ') => {
let prefix = if env_prefix.is_empty() {
String::new()
} else {
format!("{env_prefix} ")
};
format!("{prefix}{value} {}", backends::shell_quote(script_name))
}
value => {
let prefix = if env_prefix.is_empty() {
String::new()
} else {
format!("{env_prefix} ")
};
format!("{prefix}{value} {}", backends::shell_quote(script_name))
}
};
wrap_command_with_tmp_env(&command, tmp_root)
}
#[cfg(unix)]
fn process_outcome_from_exit(exit_code: i32) -> ProcessOutcome {
match exit_code {
124 => ProcessOutcome::TimedOut,
130 => ProcessOutcome::Cancelled,
code => ProcessOutcome::Completed(code),
}
}
fn render_execute_code_result(
stdout: String,
stderr: String,
outcome: ProcessOutcome,
tool_calls_made: usize,
duration: f64,
timeout_secs: u64,
) -> String {
match outcome {
ProcessOutcome::Completed(exit) => {
let status = if exit == 0 { "success" } else { "error" };
let mut result = json!({
"status": status,
"output": stdout,
"tool_calls_made": tool_calls_made,
"duration_seconds": (duration * 100.0).round() / 100.0,
});
if exit != 0 {
if !stderr.is_empty() {
result["error"] = json!(stderr);
result["output"] = json!(format!("{stdout}\n--- stderr ---\n{stderr}"));
} else {
result["error"] = json!(format!("Script exited with code {exit}"));
}
}
result.to_string()
}
ProcessOutcome::TimedOut => json!({
"status": "timeout",
"error": format!("Script timed out after {timeout_secs}s and was killed."),
"output": if stderr.is_empty() { stdout } else { format!("{stdout}\n--- stderr ---\n{stderr}") },
"tool_calls_made": tool_calls_made,
"duration_seconds": (duration * 100.0).round() / 100.0,
})
.to_string(),
ProcessOutcome::Cancelled => json!({
"status": "interrupted",
"error": "Execution interrupted by cancellation.",
"output": format!("{stdout}\n[execution interrupted — user sent a new message]"),
"tool_calls_made": tool_calls_made,
"duration_seconds": (duration * 100.0).round() / 100.0,
})
.to_string(),
}
}
#[cfg(unix)]
fn resolve_sandbox_tools(ctx: &ToolContext) -> Vec<&'static str> {
let fs = describe_execution_filesystem(&ctx.config, &ctx.cwd);
let allow_terminal = fs.execute_code_terminal_is_safe();
let Some(registry) = ctx.tool_registry.as_ref() else {
return Vec::new();
};
let mut resolved = Vec::new();
for &tool_name in SANDBOX_ALLOWED_TOOLS {
if tool_name == "terminal" && !allow_terminal {
continue;
}
let Some(toolset) = registry.toolset_for_tool(tool_name) else {
continue;
};
if ctx.config.is_tool_enabled(tool_name, &toolset) {
resolved.push(tool_name);
}
}
resolved
}
#[derive(Debug)]
struct HeadTailCapture {
head_limit: usize,
tail_limit: usize,
head: Vec<u8>,
tail: VecDeque<u8>,
total: usize,
}
impl HeadTailCapture {
fn new(head_limit: usize, tail_limit: usize) -> Self {
Self {
head_limit,
tail_limit,
head: Vec::new(),
tail: VecDeque::new(),
total: 0,
}
}
fn push(&mut self, chunk: &[u8]) {
self.total = self.total.saturating_add(chunk.len());
let mut remaining = chunk;
if self.head.len() < self.head_limit {
let keep = (self.head_limit - self.head.len()).min(remaining.len());
self.head.extend_from_slice(&remaining[..keep]);
remaining = &remaining[keep..];
}
if self.tail_limit == 0 || remaining.is_empty() {
return;
}
for &byte in remaining {
self.tail.push_back(byte);
while self.tail.len() > self.tail_limit {
self.tail.pop_front();
}
}
}
fn into_text(self) -> String {
if self.total <= self.head.len() + self.tail.len() {
let mut bytes = self.head;
bytes.extend(self.tail);
return String::from_utf8_lossy(&bytes).into_owned();
}
let head_text = String::from_utf8_lossy(&self.head).into_owned();
let tail_bytes: Vec<u8> = self.tail.into_iter().collect();
let tail_text = String::from_utf8_lossy(&tail_bytes).into_owned();
let omitted = self
.total
.saturating_sub(self.head.len())
.saturating_sub(tail_bytes.len());
format!(
"{head_text}\n\n... [OUTPUT TRUNCATED - {omitted} chars omitted out of {} total] ...\n\n{tail_text}",
self.total
)
}
}
#[derive(Debug)]
struct HeadOnlyCapture {
limit: usize,
bytes: Vec<u8>,
total: usize,
}
impl HeadOnlyCapture {
fn new(limit: usize) -> Self {
Self {
limit,
bytes: Vec::new(),
total: 0,
}
}
fn push(&mut self, chunk: &[u8]) {
self.total = self.total.saturating_add(chunk.len());
if self.bytes.len() >= self.limit {
return;
}
let keep = (self.limit - self.bytes.len()).min(chunk.len());
self.bytes.extend_from_slice(&chunk[..keep]);
}
fn into_text(self) -> String {
let text = String::from_utf8_lossy(&self.bytes).into_owned();
if self.total <= self.limit {
text
} else {
format!("{text}... (stderr truncated)")
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProcessOutcome {
Completed(i32),
TimedOut,
Cancelled,
}
#[derive(Debug)]
struct ProcessRunResult {
stdout: String,
stderr: String,
outcome: ProcessOutcome,
}
async fn capture_stdout(
mut stdout: tokio::process::ChildStdout,
progress: Option<std::sync::Arc<crate::tool_progress_tail::ToolProgressTail>>,
) -> Result<String, std::io::Error> {
use std::sync::Arc;
use tokio::io::AsyncReadExt;
let mut capture = HeadTailCapture::new(MAX_STDOUT_BYTES * 2 / 5, MAX_STDOUT_BYTES * 3 / 5);
let mut buf = [0u8; 4096];
let mut writer = progress
.as_ref()
.map(|reporter| crate::tool_progress_tail::TailByteWriter::stdout(Arc::clone(reporter)));
loop {
let read = stdout.read(&mut buf).await?;
if read == 0 {
break;
}
capture.push(&buf[..read]);
if let Some(writer) = writer.as_mut() {
writer.push(&buf[..read]);
}
}
if let Some(writer) = writer.as_mut() {
writer.finish();
}
if let Some(reporter) = progress {
reporter.flush();
}
Ok(capture.into_text())
}
async fn capture_stderr(
mut stderr: tokio::process::ChildStderr,
progress: Option<std::sync::Arc<crate::tool_progress_tail::ToolProgressTail>>,
) -> Result<String, std::io::Error> {
use std::sync::Arc;
use tokio::io::AsyncReadExt;
let mut capture = HeadOnlyCapture::new(MAX_STDERR_BYTES);
let mut buf = [0u8; 4096];
let mut writer = progress
.as_ref()
.map(|reporter| crate::tool_progress_tail::TailByteWriter::stderr(Arc::clone(reporter)));
loop {
let read = stderr.read(&mut buf).await?;
if read == 0 {
break;
}
capture.push(&buf[..read]);
if let Some(writer) = writer.as_mut() {
writer.push(&buf[..read]);
}
}
if let Some(writer) = writer.as_mut() {
writer.finish();
}
Ok(capture.into_text())
}
#[cfg(unix)]
async fn kill_process_group(child: &mut tokio::process::Child, escalate: bool) {
if let Some(pid) = child.id() {
unsafe {
libc::killpg(pid as i32, libc::SIGTERM);
}
if escalate {
tokio::time::sleep(Duration::from_secs(2)).await;
unsafe {
libc::killpg(pid as i32, libc::SIGKILL);
}
}
}
let _ = child.start_kill();
}
#[cfg(not(unix))]
async fn kill_process_group(child: &mut tokio::process::Child, _escalate: bool) {
let _ = child.start_kill();
}
#[cfg_attr(not(test), allow(dead_code))]
fn head_tail_truncate(text: &str, max_bytes: usize) -> String {
if text.len() <= max_bytes {
return text.to_string();
}
let head_bytes = (max_bytes * 2) / 5; let tail_bytes = max_bytes - head_bytes;
let head_end = (0..=head_bytes)
.rev()
.find(|&idx| text.is_char_boundary(idx))
.unwrap_or(0);
let tail_start_raw = text.len().saturating_sub(tail_bytes);
let tail_start = (tail_start_raw..=text.len())
.find(|&idx| text.is_char_boundary(idx))
.unwrap_or(text.len());
if tail_start <= head_end {
return text[..head_end].to_string();
}
let head = &text[..head_end];
let tail = &text[tail_start..];
let omitted = text.len() - head_end - (text.len() - tail_start);
format!(
"{head}\n\n... [OUTPUT TRUNCATED - {omitted} chars omitted out of {} total] ...\n\n{tail}",
text.len()
)
}
fn strip_ansi(text: &str) -> String {
match strip_ansi_escapes::strip_str(text) {
s if s.is_empty() && !text.is_empty() => text.to_string(),
s => s.to_string(),
}
}
async fn run_command_capture(
mut cmd: tokio::process::Command,
timeout_secs: u64,
cancel: tokio_util::sync::CancellationToken,
progress: Option<std::sync::Arc<crate::tool_progress_tail::ToolProgressTail>>,
) -> Result<ProcessRunResult, ToolError> {
use std::process::Stdio;
cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = cmd.spawn().map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("Failed to spawn process: {e}"),
})?;
let stdout = child
.stdout
.take()
.ok_or_else(|| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: "Failed to capture stdout".into(),
})?;
let stderr = child
.stderr
.take()
.ok_or_else(|| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: "Failed to capture stderr".into(),
})?;
let progress_stdout = progress.clone();
let progress_stderr = progress.clone();
let stdout_task = tokio::spawn(capture_stdout(stdout, progress_stdout));
let stderr_task = tokio::spawn(capture_stderr(stderr, progress_stderr));
let outcome = tokio::select! {
wait_result = child.wait() => {
let status = wait_result.map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("Process wait failed: {e}"),
})?;
ProcessOutcome::Completed(status.code().unwrap_or(-1))
}
_ = tokio::time::sleep(Duration::from_secs(timeout_secs)) => {
kill_process_group(&mut child, true).await;
let _ = child.wait().await;
ProcessOutcome::TimedOut
}
_ = cancel.cancelled() => {
kill_process_group(&mut child, false).await;
let _ = child.wait().await;
ProcessOutcome::Cancelled
}
};
let stdout = stdout_task
.await
.map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("stdout capture join failed: {e}"),
})?
.map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("stdout capture failed: {e}"),
})?;
let stderr = stderr_task
.await
.map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("stderr capture join failed: {e}"),
})?
.map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("stderr capture failed: {e}"),
})?;
Ok(ProcessRunResult {
stdout,
stderr,
outcome,
})
}
fn configure_process_group(_cmd: &mut tokio::process::Command) {
#[cfg(unix)]
unsafe {
_cmd.pre_exec(|| {
libc::setpgid(0, 0);
Ok(())
});
}
}
fn build_description() -> String {
"Run a Python script that can call Lingshu tools programmatically. \
Use this when you need 3+ tool calls with processing logic between them, \
need to filter/reduce large tool outputs before they enter your context, \
need conditional branching (if X then Y else Z), or need to loop \
(fetch N pages, process N files, retry on failure).\n\n\
Use normal tool calls instead when: single tool call with no processing, \
you need to see the full result and apply complex reasoning, \
or the task requires interactive user input.\n\n\
Available via `from lingshu_tools import ...`:\n\n\
web_search(query: str, max_results: int = 5, backend: str = None) -> dict\n\
Returns {\"query\", \"backend\", \"results\": [{\"url\", \"title\", \"description\"}, ...]}\n\
web_extract(url: str, max_chars: int = 8000, backend: str = None, render_js_fallback: bool = True) -> dict\n\
Returns {\"backend\", \"result\": {\"url\", \"title\", \"content\", \"extractor\", \"content_type\", \"content_format\"}}\n\
web_crawl(url: str, instructions: str = None, max_pages: int = 8, max_depth: int = 2, backend: str = None, render_js_fallback: bool = True) -> dict\n\
Crawl multiple linked pages from a start URL. Returns {\"success\", \"backend\", \"pages_visited\", \"results\": [{\"url\", \"title\", \"depth\", \"content\", \"extractor\", \"content_type\", \"content_format\"}, ...]}\n\
read_file(path: str, offset: int = 1, limit: int = 500) -> dict\n\
Lines are 1-indexed. Returns {\"content\": \"...\", \"total_lines\": N}\n\
pdf_to_markdown(path: str, output_path: str = None, max_chars: int = 20000) -> dict\n\
Convert a local PDF into Markdown using EdgeParse. Fast structural parsing only, not OCR.\n\
write_file(path: str, content: str) -> dict\n\
Always overwrites the entire file.\n\
search_files(pattern: str, target=\"content\", path=\".\", file_glob=None, limit=50) -> dict\n\
target: \"content\" (search inside files) or \"files\" (find files by name)\n\
patch(path: str, old_string: str, new_string: str, replace_all: bool = False) -> dict\n\
Replaces old_string with new_string in the file.\n\
terminal(command: str, timeout=None, workdir=None) -> dict\n\
Foreground only (no background/pty). Returns {\"output\": \"...\", \"exit_code\": N}\n\
vision_analyze(image_source: str, question: str = \"Describe this image.\", detail: str = \"high\") -> dict\n\
Analyze a local image file path or https:// URL. Returns {\"content\": \"...\"}\n\
transcribe_audio(file_path: str, provider: str = None, language: str = \"en\") -> dict\n\
Transcribe an audio file (mp3/wav/m4a/etc.) to text. Returns {\"text\": \"...\"}\n\
session_search(query: str, limit: int = 10) -> dict\n\
Full-text search over past sessions. Returns {\"results\": [...]}\n\n\
Limits: 5-minute timeout, 50KB stdout cap, max 50 tool calls per script. \
terminal() is foreground-only (no background or pty).\n\n\
Filesystem rule: direct script file I/O never enforces Lingshu's file-tool path jail. \
In local mode it sees host paths directly; in non-local terminal backends it runs inside that backend and shares the backend filesystem with `terminal()`. \
Temp-aware code should prefer `os.environ['EDGECRAB_TMPDIR']` (also exported as `TMPDIR`, `TMP`, and `TEMP`) instead of hard-coding `/tmp`. \
Prefer read_file/write_file/patch/search_files when you need workspace-root and path-restriction enforcement on host files.\n\n\
Print your final result to stdout. Use Python stdlib (json, re, math, csv, \
datetime, collections, etc.) for processing between tool calls.\n\n\
Also available (no import needed — built into lingshu_tools):\n\
json_parse(text: str) — json.loads with strict=False; use for terminal() output with control chars\n\
shell_quote(s: str) — shlex.quote(); use when interpolating dynamic strings into shell commands\n\
retry(fn, max_attempts=3, delay=2) — retry with exponential backoff for transient failures\n\n\
Also supports non-Python languages (javascript, typescript, bash, ruby, perl, rust) \
but without RPC tool access — those run as simple subprocess scripts.\n\n\
Note: the exact callable subset is also constrained by the current \
session's enabled toolsets; unavailable helper functions raise a clear \
runtime error inside the sandbox."
.to_string()
}
#[cfg(unix)]
async fn execute_remote(
ctx: &ToolContext,
runtime: &str,
ext: &str,
code: &str,
timeout_secs: u64,
is_python: bool,
sandbox_tools: &[&'static str],
) -> Result<String, ToolError> {
let backend = get_or_create_backend(ctx).await?;
if !backend.supports_remote_execute_code() {
return Err(ToolError::Unavailable {
tool: "execute_code".into(),
reason: format!(
"Remote execute_code is not supported by the {} backend.",
backend.kind()
),
});
}
let exec_start = std::time::Instant::now();
let sandbox_dir = prepare_remote_sandbox(
&*backend,
&uuid::Uuid::new_v4().simple().to_string()[..12],
ctx.cancel.clone(),
)
.await?;
let script_name = format!("script{ext}");
let script_path = format!("{sandbox_dir}/{script_name}");
write_remote_file(&*backend, &script_path, code, ctx.cancel.clone()).await?;
let tool_call_counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let mut rpc_task = None;
let rpc_stop = tokio_util::sync::CancellationToken::new();
if is_python {
let tools_path = format!("{sandbox_dir}/lingshu_tools.py");
let tools_src = generate_tools_module(sandbox_tools, RpcTransport::File);
write_remote_file(&*backend, &tools_path, &tools_src, ctx.cancel.clone()).await?;
if let Some(registry) = ctx.tool_registry.clone() {
let rpc_ctx = ToolContext {
task_id: ctx.task_id.clone(),
cwd: ctx.cwd.clone(),
session_id: ctx.session_id.clone(),
user_task: ctx.user_task.clone(),
cancel: ctx.cancel.clone(),
config: ctx.config.clone(),
state_db: ctx.state_db.clone(),
platform: ctx.platform,
process_table: ctx.process_table.clone(),
provider: ctx.provider.clone(),
tool_registry: ctx.tool_registry.clone(),
delegate_depth: ctx.delegate_depth,
delegate_agent_id: ctx.delegate_agent_id.clone(),
delegate_parent_id: ctx.delegate_parent_id.clone(),
sub_agent_runner: ctx.sub_agent_runner.clone(),
delegation_event_tx: ctx.delegation_event_tx.clone(),
clarify_tx: None,
approval_tx: None,
on_skills_changed: ctx.on_skills_changed.clone(),
gateway_sender: ctx.gateway_sender.clone(),
origin_chat: ctx.origin_chat.clone(),
session_key: ctx.session_key.clone(),
todo_store: ctx.todo_store.clone(),
current_tool_call_id: None,
current_tool_name: None,
injected_messages: ctx.injected_messages.clone(),
tool_progress_tx: None,
watch_notification_tx: None,
mutation_turn: None,
lsp_gate: None,
kanban_task_id: None,
};
let rpc_dir = format!("{sandbox_dir}/rpc");
let allowed = Arc::new(sandbox_tools.iter().map(|tool| tool.to_string()).collect());
let poll_backend = backend.clone();
let poll_stop = rpc_stop.clone();
let poll_counter = tool_call_counter.clone();
let poll_workdir = sandbox_dir.clone();
rpc_task = Some(tokio::spawn(async move {
remote_rpc_poll_loop(RemoteRpcLoop {
backend: poll_backend,
registry,
ctx: rpc_ctx,
rpc_dir,
tool_call_counter: poll_counter,
allowed_tools: allowed,
stop: poll_stop,
terminal_default_workdir: Some(poll_workdir),
})
.await;
}));
}
}
let rpc_dir = format!("{sandbox_dir}/rpc");
let remote_command = format!(
"cd {} && {}",
backends::shell_quote(&sandbox_dir),
remote_command_for_runtime(
runtime,
&script_name,
is_python.then_some(rpc_dir.as_str()),
BACKEND_TMP_ROOT,
)
);
let (reporter, options) =
crate::tool_progress_tail::ToolProgressTail::reporter_and_options(ctx);
let output = backend
.execute(
&remote_command,
".",
Duration::from_secs(timeout_secs),
ctx.cancel.clone(),
options,
)
.await?;
if let Some(reporter) = reporter.as_ref() {
reporter.flush();
}
rpc_stop.cancel();
if let Some(task) = rpc_task {
let _ = tokio::time::timeout(Duration::from_secs(5), task).await;
}
let _ = backend
.execute_oneshot(
&format!("rm -rf {}", backends::shell_quote(&sandbox_dir)),
".",
Duration::from_secs(15),
tokio_util::sync::CancellationToken::new(),
)
.await;
let stdout = strip_ansi(&redact_output(&head_tail_truncate(
&output.stdout,
MAX_STDOUT_BYTES,
)));
let stderr = strip_ansi(&redact_output(crate::safe_truncate(
&output.stderr,
MAX_STDERR_BYTES,
)));
let tool_calls_made = tool_call_counter.load(std::sync::atomic::Ordering::Relaxed);
let duration = exec_start.elapsed().as_secs_f64();
Ok(render_execute_code_result(
stdout,
stderr,
process_outcome_from_exit(output.exit_code),
tool_calls_made,
duration,
timeout_secs,
))
}
#[async_trait]
impl ToolHandler for ExecuteCodeToolReal {
fn name(&self) -> &'static str {
"execute_code"
}
fn toolset(&self) -> &'static str {
"code_execution"
}
fn emoji(&self) -> &'static str {
"🐍"
}
fn is_available(&self) -> bool {
cfg!(unix)
}
fn schema(&self) -> ToolSchema {
ToolSchema {
name: "execute_code".into(),
description: build_description(),
parameters: json!({
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "Python code to execute. Must be non-empty — only call when you already have a concrete code payload to run. Import tools with `from lingshu_tools import web_search, terminal, ...` and print your final result to stdout."
},
"language": {
"type": "string",
"description": "Programming language (default: python). For non-Python: javascript, typescript, bash, ruby, perl, rust — these run without RPC tool access."
},
"timeout": {
"type": "integer",
"description": "Timeout in seconds (default: 300, max: 600)"
}
},
"required": ["code"]
}),
strict: None,
}
}
async fn execute(
&self,
args: serde_json::Value,
ctx: &ToolContext,
) -> Result<String, ToolError> {
let args: ExecuteCodeArgs =
serde_json::from_value(args).map_err(|e| {
let raw = e.to_string();
let message = if raw.contains("missing field `code`") {
"Invalid execute_code args: missing field `code`. Only call execute_code when you already have a concrete code payload to run; do not use it as a planning placeholder.".to_string()
} else {
format!("Invalid execute_code args: {raw}")
};
ToolError::InvalidArgs {
tool: "execute_code".into(),
message,
}
})?;
if args.code.trim().is_empty() {
return Err(ToolError::InvalidArgs {
tool: "execute_code".into(),
message: "No code provided.".into(),
});
}
if ctx.cancel.is_cancelled() {
return Err(ToolError::Other("Cancelled".into()));
}
let (runtime, ext) = resolve_runtime(&args.language).ok_or_else(|| {
ToolError::InvalidArgs {
tool: "execute_code".into(),
message: format!(
"Unsupported language: '{}'. Supported: python, javascript, typescript, bash, ruby, perl, rust",
args.language
),
}
})?;
let is_python = matches!(
args.language.to_lowercase().as_str(),
"python" | "python3" | "py" | ""
);
let timeout_secs = args.timeout.unwrap_or(DEFAULT_TIMEOUT_SECS).min(600);
let (progress_reporter, _) =
crate::tool_progress_tail::ToolProgressTail::reporter_and_options(ctx);
#[cfg(unix)]
let sandbox_tools = resolve_sandbox_tools(ctx);
let exec_start = std::time::Instant::now();
#[cfg(unix)]
if ctx.config.terminal_backend != backends::BackendKind::Local {
return execute_remote(
ctx,
runtime,
ext,
&args.code,
timeout_secs,
is_python,
&sandbox_tools,
)
.await;
}
let temp_dir = tempfile::tempdir().map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("Failed to create temp dir: {e}"),
})?;
let script_path = temp_dir.path().join(format!("script{ext}"));
std::fs::write(&script_path, &args.code).map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("Failed to write script: {e}"),
})?;
#[cfg(unix)]
let rpc_state = if is_python {
let sock_path = format!("/tmp/lingshu_rpc_{}.sock", uuid::Uuid::new_v4());
let tools_src = generate_tools_module(&sandbox_tools, RpcTransport::Uds);
let tools_path = temp_dir.path().join("lingshu_tools.py");
std::fs::write(&tools_path, &tools_src).map_err(|e| ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("Failed to write tools module: {e}"),
})?;
let listener = tokio::net::UnixListener::bind(&sock_path).map_err(|e| {
ToolError::ExecutionFailed {
tool: "execute_code".into(),
message: format!("Failed to bind RPC socket: {e}"),
}
})?;
let tool_call_counter = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let rpc_handle = if let Some(ref reg) = ctx.tool_registry {
let reg = reg.clone();
let rpc_ctx = ToolContext {
task_id: ctx.task_id.clone(),
cwd: ctx.cwd.clone(),
session_id: ctx.session_id.clone(),
user_task: ctx.user_task.clone(),
cancel: ctx.cancel.clone(),
config: ctx.config.clone(),
state_db: ctx.state_db.clone(),
platform: ctx.platform,
process_table: ctx.process_table.clone(),
provider: ctx.provider.clone(),
tool_registry: ctx.tool_registry.clone(),
delegate_depth: ctx.delegate_depth,
delegate_agent_id: ctx.delegate_agent_id.clone(),
delegate_parent_id: ctx.delegate_parent_id.clone(),
sub_agent_runner: ctx.sub_agent_runner.clone(),
delegation_event_tx: ctx.delegation_event_tx.clone(),
clarify_tx: None, approval_tx: None, on_skills_changed: ctx.on_skills_changed.clone(),
gateway_sender: ctx.gateway_sender.clone(),
origin_chat: ctx.origin_chat.clone(),
session_key: ctx.session_key.clone(),
todo_store: ctx.todo_store.clone(),
current_tool_call_id: None,
current_tool_name: None,
injected_messages: ctx.injected_messages.clone(),
tool_progress_tx: None,
watch_notification_tx: None,
mutation_turn: None,
lsp_gate: None,
kanban_task_id: None,
};
let counter = tool_call_counter.clone();
let allowed: Arc<Vec<String>> =
Arc::new(sandbox_tools.iter().map(|s| s.to_string()).collect());
Some(tokio::spawn(async move {
rpc_server_loop(listener, reg, rpc_ctx, counter, allowed).await;
}))
} else {
drop(listener);
None
};
Some((sock_path, tool_call_counter, rpc_handle))
} else {
None
};
#[cfg(not(unix))]
let rpc_state: Option<(
String,
Arc<std::sync::atomic::AtomicUsize>,
Option<tokio::task::JoinHandle<()>>,
)> = None;
let sock = if is_python {
#[cfg(unix)]
{
rpc_state.as_ref().map(|(s, _, _)| s.as_str()).unwrap_or("")
}
#[cfg(not(unix))]
{
""
}
} else {
""
};
let child_env = build_child_env(sock, &ctx.cwd);
let run_result = if runtime == "rustc_run" {
let bin_path = temp_dir.path().join("script");
let mut compile_cmd = tokio::process::Command::new("rustc");
compile_cmd
.arg(&script_path)
.arg("-o")
.arg(&bin_path)
.current_dir(&ctx.cwd)
.env_clear()
.envs(&child_env);
configure_process_group(&mut compile_cmd);
let compile = run_command_capture(
compile_cmd,
timeout_secs,
ctx.cancel.clone(),
progress_reporter.clone(),
)
.await?;
match compile.outcome {
ProcessOutcome::Completed(0) => {}
ProcessOutcome::Completed(_) => {
let stderr = strip_ansi(&redact_output(&compile.stderr));
return Ok(json!({
"status": "error",
"error": format!("Compilation error:\n{stderr}"),
"tool_calls_made": 0,
"duration_seconds": (exec_start.elapsed().as_secs_f64() * 100.0).round() / 100.0,
})
.to_string());
}
ProcessOutcome::TimedOut => {
return Ok(json!({
"status": "timeout",
"error": format!("Rust compilation timed out after {timeout_secs}s and was killed."),
"output": strip_ansi(&redact_output(&compile.stdout)),
"tool_calls_made": 0,
"duration_seconds": (exec_start.elapsed().as_secs_f64() * 100.0).round() / 100.0,
})
.to_string());
}
ProcessOutcome::Cancelled => {
return Ok(json!({
"status": "interrupted",
"error": "Rust compilation interrupted by cancellation.",
"output": format!("{}\n[execution interrupted — user sent a new message]", strip_ansi(&redact_output(&compile.stdout))),
"tool_calls_made": 0,
"duration_seconds": (exec_start.elapsed().as_secs_f64() * 100.0).round() / 100.0,
})
.to_string());
}
}
let mut run_cmd = tokio::process::Command::new(bin_path);
run_cmd.current_dir(&ctx.cwd).env_clear().envs(&child_env);
configure_process_group(&mut run_cmd);
run_command_capture(
run_cmd,
timeout_secs,
ctx.cancel.clone(),
progress_reporter.clone(),
)
.await?
} else if runtime.contains(' ') {
let parts: Vec<&str> = runtime.split_whitespace().collect();
let mut cmd = tokio::process::Command::new(parts[0]);
for arg in &parts[1..] {
cmd.arg(arg);
}
cmd.arg(&script_path)
.current_dir(&ctx.cwd)
.env_clear()
.envs(&child_env);
configure_process_group(&mut cmd);
run_command_capture(
cmd,
timeout_secs,
ctx.cancel.clone(),
progress_reporter.clone(),
)
.await?
} else {
let mut cmd = tokio::process::Command::new(runtime);
cmd.arg(&script_path)
.current_dir(&ctx.cwd)
.env_clear()
.envs(&child_env);
configure_process_group(&mut cmd);
run_command_capture(
cmd,
timeout_secs,
ctx.cancel.clone(),
progress_reporter.clone(),
)
.await?
};
let duration = exec_start.elapsed().as_secs_f64();
let tool_calls_made = rpc_state
.as_ref()
.map(|(_, c, _)| c.load(std::sync::atomic::Ordering::Relaxed))
.unwrap_or(0);
#[cfg(unix)]
if let Some((sock_path, _, rpc_handle)) = &rpc_state {
let _ = std::fs::remove_file(sock_path);
if let Some(h) = rpc_handle {
h.abort();
}
}
let stdout = strip_ansi(&redact_output(&run_result.stdout));
let stderr = strip_ansi(&redact_output(&run_result.stderr));
Ok(render_execute_code_result(
stdout,
stderr,
run_result.outcome,
tool_calls_made,
duration,
timeout_secs,
))
}
}
inventory::submit!(&ExecuteCodeToolReal as &dyn ToolHandler);
#[cfg(test)]
mod tests {
use super::*;
#[cfg(unix)]
use crate::tools::backend_pool::cleanup_backend_for_task;
#[cfg(unix)]
use crate::tools::backends::singularity::SINGULARITY_TEST_ENV_LOCK;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::sync::Arc;
#[test]
fn resolve_python() {
let (rt, ext) = resolve_runtime("python").expect("should resolve");
assert_eq!(rt, "python3");
assert_eq!(ext, ".py");
}
#[test]
fn resolve_javascript() {
let (rt, ext) = resolve_runtime("js").expect("should resolve");
assert_eq!(rt, "node");
assert_eq!(ext, ".js");
}
#[test]
fn resolve_bash() {
let (rt, ext) = resolve_runtime("bash").expect("should resolve");
assert_eq!(rt, "bash");
assert_eq!(ext, ".sh");
}
#[test]
fn resolve_rust() {
let (rt, ext) = resolve_runtime("rust").expect("should resolve");
assert_eq!(rt, "rustc_run");
assert_eq!(ext, ".rs");
}
#[test]
fn resolve_unknown_returns_none() {
assert!(resolve_runtime("cobol").is_none());
}
#[test]
fn head_tail_truncate_short() {
let text = "hello world";
assert_eq!(head_tail_truncate(text, 100), "hello world");
}
#[test]
fn head_tail_truncate_long() {
let text = "A".repeat(1000);
let result = head_tail_truncate(&text, 100);
assert!(result.contains("OUTPUT TRUNCATED"));
assert!(result.len() < 200); }
#[test]
fn head_tail_truncate_preserves_utf8_boundaries() {
let text = "🙂".repeat(400);
let result = head_tail_truncate(&text, 101);
assert!(!result.is_empty());
}
#[test]
fn strip_ansi_removes_colors() {
let text = "\x1b[31mred\x1b[0m normal";
let result = strip_ansi(text);
assert_eq!(result, "red normal");
}
#[cfg(unix)]
#[test]
fn generate_tools_module_contains_stubs() {
let module = generate_tools_module(&["web_search", "terminal"], RpcTransport::Uds);
assert!(module.contains("def web_search("));
assert!(module.contains("max_results: int = 5"));
assert!(module.contains("backend: str = None"));
assert!(module.contains("def terminal("));
assert!(module.contains("def json_parse("));
assert!(module.contains("def shell_quote("));
assert!(module.contains("def retry("));
assert!(module.contains("def read_file("));
assert!(module.contains("not available in this execute_code session"));
}
#[cfg(unix)]
#[test]
fn generate_tools_module_uses_current_web_extract_signature() {
let module = generate_tools_module(&["web_extract"], RpcTransport::Uds);
assert!(module.contains(
"def web_extract(url: str, max_chars: int = 8000, backend: str = None, render_js_fallback: bool = True):"
));
assert!(module.contains(r#"args = {"url": url, "max_chars": max_chars, "render_js_fallback": render_js_fallback}"#));
}
#[cfg(unix)]
#[test]
fn generate_tools_module_all_tools() {
let module = generate_tools_module(SANDBOX_ALLOWED_TOOLS, RpcTransport::Uds);
for tool in SANDBOX_ALLOWED_TOOLS {
assert!(
module.contains(&format!("def {tool}(")),
"missing stub for {tool}"
);
}
}
#[cfg(unix)]
#[test]
fn generate_tools_module_file_transport_uses_rpc_dir() {
let module = generate_tools_module(&["terminal"], RpcTransport::File);
assert!(module.contains("EDGECRAB_RPC_DIR"));
assert!(module.contains("req_"));
assert!(module.contains("res_"));
}
#[test]
fn build_child_env_strips_secrets() {
let env = build_child_env("/tmp/test.sock", std::path::Path::new("/tmp"));
assert_eq!(
env.get("EDGECRAB_RPC_SOCKET").map(|s| s.as_str()),
Some("/tmp/test.sock")
);
assert_eq!(
env.get("PYTHONDONTWRITEBYTECODE").map(|s| s.as_str()),
Some("1")
);
assert_eq!(
env.get("EDGECRAB_TMPDIR"),
env.get("TMPDIR"),
"child env must align EDGECRAB_TMPDIR and TMPDIR"
);
}
#[cfg(unix)]
#[test]
fn remote_command_for_runtime_exports_backend_tmpdir() {
let command =
remote_command_for_runtime("python3", "script.py", Some("/tmp/rpc"), BACKEND_TMP_ROOT);
assert!(command.contains("EDGECRAB_TMPDIR='/tmp/lingshu-tmp'"));
assert!(command.contains("TMPDIR='/tmp/lingshu-tmp'"));
assert!(command.contains("python3 'script.py'"));
}
#[cfg(unix)]
#[test]
fn resolve_sandbox_tools_respects_disabled_toolsets() {
let mut ctx = ToolContext::test_context();
ctx.tool_registry = Some(Arc::new(ToolRegistry::new()));
ctx.config.disabled_toolsets = vec!["web".into(), "terminal".into()];
let tools = resolve_sandbox_tools(&ctx);
assert!(!tools.contains(&"web_search"));
assert!(!tools.contains(&"terminal"));
assert!(tools.contains(&"read_file"));
}
#[cfg(unix)]
#[test]
fn resolve_sandbox_tools_keeps_terminal_for_remote_backends() {
let mut ctx = ToolContext::test_context();
ctx.tool_registry = Some(Arc::new(ToolRegistry::new()));
ctx.config.terminal_backend = crate::tools::backends::BackendKind::Modal;
let tools = resolve_sandbox_tools(&ctx);
assert!(tools.contains(&"terminal"));
assert!(tools.contains(&"read_file"));
}
#[test]
fn build_description_mentions_tools() {
let desc = build_description();
assert!(desc.contains("web_search"));
assert!(desc.contains("web_crawl"));
assert!(desc.contains(
"web_extract(url: str, max_chars: int = 8000, backend: str = None, render_js_fallback: bool = True)"
));
assert!(desc.contains("transcribe_audio"));
assert!(desc.contains("session_search"));
assert!(desc.contains("vision_analyze"));
assert!(desc.contains("terminal"));
assert!(desc.contains("lingshu_tools"));
assert!(
desc.contains("direct script file I/O never enforces Lingshu's file-tool path jail")
);
assert!(desc.contains(
"runs inside that backend and shares the backend filesystem with `terminal()`"
));
}
#[tokio::test]
async fn execute_bash_echo() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool
.execute(
json!({ "language": "bash", "code": "echo hello world" }),
&ctx,
)
.await
.expect("should succeed");
assert!(result.contains("hello world"));
assert!(result.contains("success") || result.contains("Exit code: 0"));
}
#[tokio::test]
async fn execute_python_print() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool
.execute(
json!({ "language": "python", "code": "print(2 + 2)" }),
&ctx,
)
.await;
match result {
Ok(output) => assert!(output.contains("4") || output.contains("status")),
Err(ToolError::ExecutionFailed { .. }) => {} Err(e) => panic!("unexpected error: {e:?}"),
}
}
#[tokio::test]
async fn execute_python_default_language() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool
.execute(json!({ "code": "print('default python')" }), &ctx)
.await;
match result {
Ok(output) => assert!(output.contains("default python") || output.contains("status")),
Err(ToolError::ExecutionFailed { .. }) => {}
Err(e) => panic!("unexpected error: {e:?}"),
}
}
#[tokio::test]
async fn execute_empty_code_rejected() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool.execute(json!({ "code": " " }), &ctx).await;
assert!(matches!(result, Err(ToolError::InvalidArgs { .. })));
}
#[tokio::test]
async fn execute_invalid_language() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool
.execute(json!({ "language": "cobol", "code": "DISPLAY 'HI'" }), &ctx)
.await;
assert!(matches!(result, Err(ToolError::InvalidArgs { .. })));
}
#[tokio::test]
async fn execute_bash_failure() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool
.execute(json!({ "language": "bash", "code": "exit 42" }), &ctx)
.await
.expect("should return JSON even on failure");
assert!(result.contains("error") || result.contains("42"));
}
#[tokio::test]
async fn execute_bash_uses_sanitized_env() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
unsafe {
std::env::set_var("OPENAI_API_KEY", "sk-test-secret-12345");
}
let result = tool
.execute(
json!({ "language": "bash", "code": "printf '%s' \"${OPENAI_API_KEY:-NOT_SET}\"" }),
&ctx,
)
.await
.expect("should return json");
unsafe {
std::env::remove_var("OPENAI_API_KEY");
}
assert!(result.contains("NOT_SET"), "got: {result}");
assert!(!result.contains("sk-test-secret-12345"), "got: {result}");
}
#[tokio::test]
async fn execute_timeout_kills_process_and_reports_timeout() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool
.execute(
json!({ "language": "bash", "code": "sleep 2", "timeout": 1 }),
&ctx,
)
.await
.expect("should return timeout json");
let parsed: serde_json::Value =
serde_json::from_str(&result).expect("output should be json");
assert_eq!(parsed["status"], "timeout");
}
#[tokio::test]
async fn execute_returns_json() {
let tool = ExecuteCodeToolReal;
let ctx = ToolContext::test_context();
let result = tool
.execute(json!({ "language": "bash", "code": "echo ok" }), &ctx)
.await
.expect("should succeed");
let parsed: serde_json::Value =
serde_json::from_str(&result).expect("output should be JSON");
assert!(parsed.get("status").is_some());
assert!(parsed.get("tool_calls_made").is_some());
assert!(parsed.get("duration_seconds").is_some());
}
#[test]
fn tool_metadata() {
let tool = ExecuteCodeToolReal;
assert_eq!(tool.name(), "execute_code");
assert_eq!(tool.toolset(), "code_execution");
assert_eq!(tool.emoji(), "🐍");
}
#[cfg(unix)]
fn write_fake_singularity_runtime(dir: &tempfile::TempDir, body: &str) -> std::path::PathBuf {
let path = dir.path().join("fake-apptainer");
std::fs::write(&path, body).expect("write fake runtime");
let mut perms = std::fs::metadata(&path).expect("metadata").permissions();
perms.set_mode(0o755);
std::fs::set_permissions(&path, perms).expect("chmod");
path
}
#[cfg(unix)]
#[tokio::test]
#[allow(clippy::await_holding_lock)]
async fn execute_code_remote_terminal_shares_backend_workdir() {
let _guard = SINGULARITY_TEST_ENV_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let dir = tempfile::tempdir().expect("tmpdir");
let fake = write_fake_singularity_runtime(
&dir,
r#"#!/bin/sh
set -eu
case "${1:-}" in
version)
echo "apptainer 1.0.0"
;;
instance)
exit 0
;;
exec)
workdir="."
last=""
while [ "$#" -gt 0 ]; do
case "$1" in
--pwd)
workdir="$2"
shift 2
;;
*)
last="$1"
shift
;;
esac
done
mkdir -p "$workdir"
cd "$workdir"
exec sh -c "$last"
;;
esac
"#,
);
unsafe {
std::env::set_var("EDGECRAB_SINGULARITY_BIN", &fake);
}
let mut ctx = ToolContext::test_context();
ctx.cwd = dir.path().to_path_buf();
ctx.task_id = format!("exec-remote-{}", uuid::Uuid::new_v4().simple());
ctx.config.terminal_backend = backends::BackendKind::Singularity;
ctx.tool_registry = Some(Arc::new(ToolRegistry::new()));
let tool = ExecuteCodeToolReal;
let result = tool
.execute(
json!({
"code": "from lingshu_tools import terminal\nwith open('note.txt', 'w') as fh:\n fh.write('backend-note')\nprint(terminal('cat note.txt'))\n"
}),
&ctx,
)
.await
.expect("remote execute_code");
assert!(result.contains("backend-note"), "got: {result}");
let _ = cleanup_backend_for_task(&ctx.task_id).await;
unsafe {
std::env::remove_var("EDGECRAB_SINGULARITY_BIN");
}
}
}