diff --git a/README.md b/README.md index 7cd4a31..bff7b32 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,7 @@ Mac 客户端（v1 即将发布）：拖入 Applications → 首启完成 onboar | [docs/DEVELOPMENT_PLAN.md](docs/DEVELOPMENT_PLAN.md) | 整体开发方案 v0.5（1500+ 行 / §3 模块 / §6 里程碑） | | [docs/VISUAL_DESIGN.html](docs/VISUAL_DESIGN.html) | 视觉设计 v0.4（11 屏 mockup） | | [docs/security-model.md](docs/security-model.md) | 威胁模型 + 防御层 + 攻击向量测试 + 已知缺口 | +| [docs/design/session-format-v1.md](docs/design/session-format-v1.md) | 统一 session JSONL、旧格式迁移与 writer ownership | | [docs/design/sandbox-plan-worktree.md](docs/design/sandbox-plan-worktree.md) | sandbox × plan mode × worktree 关系矩阵 | | [docs/design/plugin-security.md](docs/design/plugin-security.md) | plugin 信任 ladder + sandbox 子进程 | | [docs/design/effort-levels.md](docs/design/effort-levels.md) | 5 档 effort 到 DeepSeek API 参数映射 | diff --git a/apps/cli/src/headless.ts b/apps/cli/src/headless.ts index c24213b..f76f997 100644 --- a/apps/cli/src/headless.ts +++ b/apps/cli/src/headless.ts @@ -21,6 +21,7 @@ import { EFFORT_PARAMS, HookDispatcher, ReadTool, + RuntimeHost, SessionManager, ToolRegistry, WebFetchTool, @@ -39,7 +40,6 @@ import { loadSkills, makeSkillTool, resolveCredentials, - runAgent, wirePlugins, collectPluginContributions, type AgentEvent, @@ -271,9 +271,18 @@ export async function runHeadless(opts: HeadlessOpts): Promise { } let exitCode = 0; try { - const result = await runAgent({ + const runtime = new RuntimeHost({ provider, tools, + cwd, + mode, + permissions: settings.permissions, + hooks, + pluginDirs: pluginContrib.dirs, + autoMode: settings.autoMode, + sandboxConfig: settings.sandbox, + }); + const result = await runtime.run({ systemPrompt, userMessage, history: [], @@ -281,15 +290,9 @@ export async function runHeadless(opts: HeadlessOpts): Promise { maxTokens, temperature, maxTurns, - cwd, + signal: ctrl.signal, session: { manager: sessions, id: session.id }, - mode, - permissions: settings.permissions, - hooks, - pluginDirs: pluginContrib.dirs, autoCompact: { contextWindow: contextWindowFor(model), threshold: 0.8 }, - autoMode: settings.autoMode, - sandboxConfig: settings.sandbox, // In headless mode there's no human to ask: auto-deny anything that // would normally need approval. Users wanting auto-yes should pass // --mode dontAsk or --mode bypassPermissions (gated by trust). @@ -441,6 +444,7 @@ function formatEventText(out: Writable, e: AgentEvent): void { return; case 'usage': case 'thinking_delta': + case 'model_step_complete': case 'turn_complete': return; } diff --git a/apps/cli/src/repl.ts b/apps/cli/src/repl.ts index 150854d..fb88bee 100644 --- a/apps/cli/src/repl.ts +++ b/apps/cli/src/repl.ts @@ -8,6 +8,7 @@ import { EFFORT_PARAMS, HookDispatcher, ReadTool, + RuntimeHost, SessionManager, TaskManager, ToolRegistry, @@ -37,7 +38,6 @@ import { contextWindowFor, makeSkillTool, resolveCredentials, - runAgent, settingsPaths, wirePlugins, collectPluginContributions, @@ -429,6 +429,17 @@ export async function startRepl(opts: ReplOpts): Promise { } let history: StoredMessage[] = resolved.seededHistory; + const runtime = new RuntimeHost({ + provider, + tools, + cwd, + mode, + permissions: settings.permissions, + hooks, + pluginDirs: pluginContrib.dirs, + autoMode: settings.autoMode, + sandboxConfig: settings.sandbox, + }); const ctx: SessionContext = { cwd, model, @@ -471,25 +482,20 @@ export async function startRepl(opts: ReplOpts): Promise { // reading ctx.model/ctx.mode live so /model and /mode switches are honored. const tasks = new TaskManager((spec) => { const ac = new AbortController(); - const done = runAgent({ - provider, - tools, - systemPrompt, - userMessage: spec.prompt, - model: ctx.model, - maxTokens, - temperature, - cwd: ctx.cwd, - signal: ac.signal, - mode: ctx.mode as Mode, - permissions: settings.permissions, - hooks, - pluginDirs: pluginContrib.dirs, - sandboxConfig: settings.sandbox, - autoMode: settings.autoMode, - subAgentDepth: 1, - systemReminders: false, - }).then((r) => assistantText(r.history)); + const done = runtime + .run({ + systemPrompt, + userMessage: spec.prompt, + model: ctx.model, + maxTokens, + temperature, + cwd: ctx.cwd, + signal: ac.signal, + modeOverride: ctx.mode as Mode, + subAgentDepth: 1, + systemReminders: false, + }) + .then((r) => assistantText(r.history)); return { done, abort: () => ac.abort() }; }); ctx.tasks = tasks; @@ -649,9 +655,7 @@ export async function startRepl(opts: ReplOpts): Promise { } // Otherwise: send to agent (with mode/permission/hooks gating from M3b) - const result = await runAgent({ - provider, - tools, + const result = await runtime.run({ systemPrompt, userMessage: userInput, history, @@ -663,13 +667,8 @@ export async function startRepl(opts: ReplOpts): Promise { // ctx.sessionId (not the launch `session.id`) so a live `/resume ` // switch redirects new messages to the resumed session. session: { manager: sessions, id: ctx.sessionId }, - mode: ctx.mode as Mode, - permissions: settings.permissions, - hooks, - pluginDirs: pluginContrib.dirs, + modeOverride: ctx.mode as Mode, autoCompact: { contextWindow: contextWindowFor(ctx.model), threshold: 0.8 }, - autoMode: settings.autoMode, - sandboxConfig: settings.sandbox, // Session-scoped manager: the agent's TaskCreate calls land here too, so // background tasks persist across turns and show up in /tasks. taskManager: tasks, @@ -762,6 +761,7 @@ function formatEvent(out: Writable, e: AgentEvent): void { else out.write(` ✓ ${truncate(e.result.content, 200)}\n`); return; case 'usage': + case 'model_step_complete': return; case 'error': out.write(`\n ✕ ${e.error}\n`); diff --git a/apps/desktop/src-tauri/Cargo.lock b/apps/desktop/src-tauri/Cargo.lock index d25f9f5..ce392f9 100644 --- a/apps/desktop/src-tauri/Cargo.lock +++ b/apps/desktop/src-tauri/Cargo.lock @@ -675,6 +675,7 @@ name = "deepcode_desktop" version = "0.1.6" dependencies = [ "dirs 5.0.1", + "libc", "serde", "serde_json", "sha2", diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index 703e41d..7a0e4f0 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -27,6 +27,7 @@ sha2 = "0.10" thiserror = "1" tokio = { version = "1", features = ["fs", "rt-multi-thread", "macros", "sync", "time", "process"] } dirs = "5" +libc = "0.2" [profile.release] panic = "abort" diff --git a/apps/desktop/src-tauri/src/commands.rs b/apps/desktop/src-tauri/src/commands.rs index a839406..eeb7bc0 100644 --- a/apps/desktop/src-tauri/src/commands.rs +++ b/apps/desktop/src-tauri/src/commands.rs @@ -4,6 +4,8 @@ use crate::credentials::{self, Credentials}; use crate::settings; use serde::Serialize; +use std::collections::HashMap; +use std::io::Write; use std::path::PathBuf; #[derive(Serialize)] @@ -137,9 +139,10 @@ pub fn session_create(cwd: String) -> Result { let id = format!("{}-{}", date, rand_id); let dir = home.join(".deepcode").join("sessions"); std::fs::create_dir_all(&dir).map_err(|e| format!("mkdir {}: {}", dir.display(), e))?; - let path = dir.join(format!("{}.jsonl", id)); + let path = dir.join(format!("{}.v1.jsonl", id)); let header = serde_json::json!({ "type": "session_meta", + "schema_version": 1, "id": id, "cwd": cwd, "created_at": secs, @@ -153,15 +156,18 @@ pub fn session_create(cwd: String) -> Result { /// Append a single JSON line to a session's JSONL file. #[tauri::command] pub fn session_append(id: String, message: serde_json::Value) -> Result<(), String> { + safe_session_id(&id)?; let Some(home) = dirs::home_dir() else { return Err("no home directory".into()); }; - let path = home - .join(".deepcode") - .join("sessions") - .join(format!("{}.jsonl", id)); - let line = format!("{}\n", message); - use std::io::Write; + let dir = home.join(".deepcode").join("sessions"); + std::fs::create_dir_all(&dir).map_err(|e| format!("mkdir {}: {}", dir.display(), e))?; + let _lock = SessionWriterLock::acquire(&dir, &id)?; + let path = ensure_canonical_session(&dir, &id)?; + let mut normalized = message; + normalized["type"] = serde_json::Value::String("message".to_string()); + normalized["schema_version"] = serde_json::Value::Number(1.into()); + let line = format!("{}\n", normalized); let mut f = std::fs::OpenOptions::new() .create(true) .append(true) @@ -177,26 +183,120 @@ pub fn session_append(id: String, message: serde_json::Value) -> Result<(), Stri /// timestamp }`. Returns an empty vec if the file doesn't exist. #[tauri::command] pub fn session_read(id: String) -> Result, String> { + safe_session_id(&id)?; let Some(home) = dirs::home_dir() else { return Err("no home directory".into()); }; - let path = home - .join(".deepcode") - .join("sessions") - .join(format!("{}.jsonl", id)); + let dir = home.join(".deepcode").join("sessions"); + let path = readable_session_path(&dir, &id); let text = match std::fs::read_to_string(&path) { Ok(t) => t, Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(vec![]), Err(e) => return Err(format!("read {}: {}", path.display(), e)), }; + parse_session_messages(&text) +} + +struct SessionWriterLock { + path: PathBuf, +} + +impl SessionWriterLock { + fn acquire(dir: &std::path::Path, id: &str) -> Result { + let path = dir.join(format!("{id}.writer.lock")); + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&path) + .map_err(|e| { + if e.kind() == std::io::ErrorKind::AlreadyExists { + format!("session {id} already has an active writer") + } else { + format!("open {}: {}", path.display(), e) + } + })?; + writeln!(file, "pid={}", std::process::id()).map_err(|e| e.to_string())?; + Ok(Self { path }) + } +} + +impl Drop for SessionWriterLock { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.path); + } +} + +fn readable_session_path(dir: &std::path::Path, id: &str) -> PathBuf { + let canonical = dir.join(format!("{id}.v1.jsonl")); + if canonical.exists() { + canonical + } else { + dir.join(format!("{id}.jsonl")) + } +} + +fn ensure_canonical_session(dir: &std::path::Path, id: &str) -> Result { + let canonical = dir.join(format!("{id}.v1.jsonl")); + if canonical.exists() { + return Ok(canonical); + } + let legacy = dir.join(format!("{id}.jsonl")); + let text = match std::fs::read_to_string(&legacy) { + Ok(text) => text, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => String::new(), + Err(error) => return Err(format!("read {}: {}", legacy.display(), error)), + }; + let messages = parse_session_messages(&text)?; + let sidecar = dir.join(format!("{id}.meta.json")); + let legacy_meta = std::fs::read_to_string(sidecar) + .ok() + .and_then(|raw| serde_json::from_str::(&raw).ok()); + let mut header = text + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .find(|value| value.get("type").and_then(|v| v.as_str()) == Some("session_meta")) + .or(legacy_meta) + .unwrap_or_else(|| serde_json::json!({ "type": "session_meta", "id": id, "cwd": "" })); + header["type"] = serde_json::Value::String("session_meta".to_string()); + header["schema_version"] = serde_json::Value::Number(1.into()); + header["id"] = serde_json::Value::String(id.to_string()); + if let Some(created_at) = header.get("createdAt").cloned() { + header["created_at"] = created_at; + } + if let Some(updated_at) = header.get("updatedAt").cloned() { + header["updated_at"] = updated_at; + } + let mut lines = vec![header.to_string()]; + for mut message in messages { + message["type"] = serde_json::Value::String("message".to_string()); + message["schema_version"] = serde_json::Value::Number(1.into()); + lines.push(message.to_string()); + } + let temp = dir.join(format!("{id}.v1.{}.tmp", std::process::id())); + std::fs::write(&temp, lines.join("\n") + "\n") + .map_err(|e| format!("write {}: {}", temp.display(), e))?; + std::fs::rename(&temp, &canonical) + .map_err(|e| format!("rename {}: {}", temp.display(), e))?; + Ok(canonical) +} + +fn parse_session_messages(text: &str) -> Result, String> { + let lines: Vec<&str> = text.split('\n').collect(); + let last_content = lines.iter().rposition(|line| !line.trim().is_empty()); let mut out = Vec::new(); - for line in text.lines() { + for (index, line) in lines.iter().enumerate() { let line = line.trim(); if line.is_empty() { continue; } - let Ok(v) = serde_json::from_str::(line) else { - continue; // tolerate a partial trailing line + let v = match serde_json::from_str::(line) { + Ok(value) => value, + Err(_) if Some(index) == last_content && !text.ends_with('\n') => { + continue; // recover an interrupted final append only + } + Err(error) => { + return Err(format!("corrupt session at line {}: {}", index + 1, error)); + } }; // Desktop sessions tag messages with type:"message"; CLI/headless sessions // write bare {role, content} lines with no type. Accept both, skip meta. @@ -206,6 +306,12 @@ pub fn session_read(id: String) -> Result, String> { Some("user") | Some("assistant") ); if t == Some("message") || (t.is_none() && is_role_msg) { + if !v.get("content").is_some_and(|content| content.is_array()) { + return Err(format!( + "corrupt session at line {}: message content must be an array", + index + 1 + )); + } out.push(v); } } @@ -296,13 +402,13 @@ fn derive_session_title(path: &std::path::Path) -> Option { /// Set (or clear, with "") a session's manual title on its session_meta header. #[tauri::command] pub fn session_set_title(id: String, title: String) -> Result<(), String> { + safe_session_id(&id)?; let Some(home) = dirs::home_dir() else { return Err("no home directory".into()); }; - let path = home - .join(".deepcode") - .join("sessions") - .join(format!("{id}.jsonl")); + let dir = home.join(".deepcode").join("sessions"); + let _lock = SessionWriterLock::acquire(&dir, &id)?; + let path = ensure_canonical_session(&dir, &id)?; let text = std::fs::read_to_string(&path).map_err(|e| format!("read {}: {}", path.display(), e))?; let trimmed = title.trim(); let mut lines: Vec = text.lines().map(|l| l.to_string()).collect(); @@ -320,7 +426,9 @@ pub fn session_set_title(id: String, title: String) -> Result<(), String> { } if !updated { // No meta header (older session) — prepend one carrying the title. - let meta = serde_json::json!({ "type": "session_meta", "id": id, "title": trimmed }); + let meta = serde_json::json!({ + "type": "session_meta", "schema_version": 1, "id": id, "title": trimmed + }); lines.insert(0, meta.to_string()); } std::fs::write(&path, lines.join("\n") + "\n") @@ -359,7 +467,7 @@ pub fn list_sessions() -> Result, String> { Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(vec![]), Err(e) => return Err(format!("read_dir {}: {}", dir.display(), e)), }; - let mut out = Vec::new(); + let mut selected: HashMap = HashMap::new(); for entry in read.flatten() { let path = entry.path(); if !path.is_file() { @@ -368,11 +476,20 @@ pub fn list_sessions() -> Result, String> { let Some(name) = path.file_name().and_then(|s| s.to_str()) else { continue; }; - if !name.ends_with(".jsonl") { + let (id, canonical) = if let Some(id) = name.strip_suffix(".v1.jsonl") { + (id.to_string(), true) + } else if let Some(id) = name.strip_suffix(".jsonl") { + (id.to_string(), false) + } else { continue; + }; + if canonical || !selected.contains_key(&id) { + selected.insert(id, path); } - let id = name.trim_end_matches(".jsonl").to_string(); - let meta = entry.metadata().map_err(|e| e.to_string())?; + } + let mut out = Vec::new(); + for (id, path) in selected { + let meta = std::fs::metadata(&path).map_err(|e| e.to_string())?; let updated_at_secs = meta .modified() .ok() @@ -407,11 +524,17 @@ pub fn session_delete(id: String) -> Result<(), String> { let Some(home) = dirs::home_dir() else { return Err("no home directory".into()); }; - let path = home - .join(".deepcode") - .join("sessions") - .join(format!("{id}.jsonl")); - std::fs::remove_file(&path).map_err(|e| format!("delete {}: {}", path.display(), e)) + let dir = home.join(".deepcode").join("sessions"); + let mut removed = false; + for name in [format!("{id}.v1.jsonl"), format!("{id}.jsonl"), format!("{id}.meta.json")] { + let path = dir.join(name); + match std::fs::remove_file(&path) { + Ok(()) => removed = true, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(format!("delete {}: {}", path.display(), error)), + } + } + if removed { Ok(()) } else { Err(format!("session not found: {id}")) } } /// Archive a session by moving its JSONL into sessions/archived/ — excluded from @@ -426,9 +549,17 @@ pub fn session_archive(id: String) -> Result<(), String> { let archived = dir.join("archived"); std::fs::create_dir_all(&archived) .map_err(|e| format!("mkdir {}: {}", archived.display(), e))?; - let from = dir.join(format!("{id}.jsonl")); - let to = archived.join(format!("{id}.jsonl")); - std::fs::rename(&from, &to).map_err(|e| format!("archive {}: {}", from.display(), e)) + let mut moved = false; + for name in [format!("{id}.v1.jsonl"), format!("{id}.jsonl"), format!("{id}.meta.json")] { + let from = dir.join(&name); + let to = archived.join(&name); + match std::fs::rename(&from, &to) { + Ok(()) => moved = true, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(format!("archive {}: {}", from.display(), error)), + } + } + if moved { Ok(()) } else { Err(format!("session not found: {id}")) } } /// Path to the `deepcode` CLI so the GUI can drop users into it for advanced @@ -773,6 +904,68 @@ mod contract_tests { assert!(name.is_none() && desc.is_none()); } + #[test] + fn session_parser_accepts_both_legacy_formats_and_truncated_tail() { + let text = concat!( + "{\"type\":\"session_meta\",\"id\":\"x\"}\n", + "{\"type\":\"message\",\"role\":\"user\",\"content\":[]}\n", + "{\"role\":\"assistant\",\"content\":[]}\n", + "{\"role\":\"assistant\"" + ); + let messages = parse_session_messages(text).unwrap(); + assert_eq!(messages.len(), 2); + } + + #[test] + fn session_parser_rejects_middle_corruption() { + let text = concat!( + "{\"role\":\"user\",\"content\":[]}\n", + "{not-json}\n", + "{\"role\":\"assistant\",\"content\":[]}\n" + ); + let error = parse_session_messages(text).unwrap_err(); + assert!(error.contains("line 2"), "got {error}"); + } + + #[test] + fn canonical_session_normalizes_without_touching_legacy() { + let root = std::env::temp_dir().join(format!( + "dc-session-v1-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&root).unwrap(); + let legacy = root.join("legacy.jsonl"); + let original = "{\"role\":\"user\",\"content\":[]}\n"; + std::fs::write(&legacy, original).unwrap(); + std::fs::write( + root.join("legacy.meta.json"), + "{\"id\":\"legacy\",\"cwd\":\"/core\",\"createdAt\":\"2025-01-01T00:00:00Z\",\"updatedAt\":\"2025-01-02T00:00:00Z\"}", + ) + .unwrap(); + + let _lock = SessionWriterLock::acquire(&root, "legacy").unwrap(); + let canonical = ensure_canonical_session(&root, "legacy").unwrap(); + assert_eq!(std::fs::read_to_string(&legacy).unwrap(), original); + let normalized = std::fs::read_to_string(canonical).unwrap(); + let records: Vec = normalized + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!(records.len(), 2); + assert_eq!(records[0]["schema_version"], 1); + assert_eq!(records[0]["cwd"], "/core"); + assert_eq!(records[0]["created_at"], "2025-01-01T00:00:00Z"); + assert_eq!(records[1]["type"], "message"); + assert_eq!(records[1]["schema_version"], 1); + assert!(SessionWriterLock::acquire(&root, "legacy").is_err()); + drop(_lock); + let _ = std::fs::remove_dir_all(root); + } + #[test] fn skill_info_serializes_camel_case() { let v = serde_json::to_value(SkillInfo { diff --git a/apps/desktop/src-tauri/src/lib.rs b/apps/desktop/src-tauri/src/lib.rs index 866d119..ed42f40 100644 --- a/apps/desktop/src-tauri/src/lib.rs +++ b/apps/desktop/src-tauri/src/lib.rs @@ -24,7 +24,9 @@ use commands::{ }; use snapshots::session_snapshots; use tauri::Manager; -use tools::{tool_bash, tool_edit, tool_glob, tool_grep, tool_read, tool_write}; +use tools::{ + tool_bash, tool_bash_cancel, tool_edit, tool_glob, tool_grep, tool_read, tool_write, BashState, +}; use voice::{voice_cancel, voice_start, voice_status, voice_stop, VoiceState}; #[cfg_attr(mobile, tauri::mobile_entry_point)] @@ -37,6 +39,7 @@ pub fn run() { .plugin(tauri_plugin_updater::Builder::new().build()) .plugin(tauri_plugin_process::init()) .manage(VoiceState::default()) + .manage(BashState::default()) .invoke_handler(tauri::generate_handler![ get_app_info, read_credentials, @@ -62,6 +65,7 @@ pub fn run() { tool_write, tool_edit, tool_bash, + tool_bash_cancel, tool_glob, tool_grep, session_snapshots, diff --git a/apps/desktop/src-tauri/src/tools.rs b/apps/desktop/src-tauri/src/tools.rs index 3f43c07..f2eee17 100644 --- a/apps/desktop/src-tauri/src/tools.rs +++ b/apps/desktop/src-tauri/src/tools.rs @@ -5,10 +5,12 @@ use crate::snapshots; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; use std::path::Path; use std::process::Stdio; use tokio::io::AsyncReadExt; use tokio::process::Command; +use tokio::sync::{oneshot, Mutex}; // ────────────────────────────────────────────────────────────────────────── // Snapshot capture @@ -240,10 +242,42 @@ pub struct BashOk { pub stderr: String, pub exit_code: i32, pub timed_out: bool, + pub cancelled: bool, } +#[derive(Default)] +pub struct BashState { + // `Some(sender)` is running; `None` records an abort that raced ahead of + // command registration so the process never escapes cancellation. + active: Mutex>>>, +} + +#[cfg(unix)] +fn kill_process_group(pid: u32) { + // The shell is placed in its own process group below, so a negative PID + // terminates the shell and every descendant it spawned. + unsafe { + libc::kill(-(pid as i32), libc::SIGKILL); + } +} + +#[cfg(not(unix))] +fn kill_process_group(_pid: u32) {} + #[tauri::command] -pub async fn tool_bash(input: BashInput) -> Result { +pub async fn tool_bash( + input: BashInput, + command_id: String, + state: tauri::State<'_, BashState>, +) -> Result { + run_bash(input, command_id, &state).await +} + +async fn run_bash( + input: BashInput, + command_id: String, + state: &BashState, +) -> Result { let timeout = std::time::Duration::from_millis(input.timeout_ms.unwrap_or(120_000)); let mut cmd = Command::new("/bin/sh"); cmd.arg("-c").arg(&input.command); @@ -251,8 +285,24 @@ pub async fn tool_bash(input: BashInput) -> Result { cmd.current_dir(cwd); } cmd.stdout(Stdio::piped()).stderr(Stdio::piped()); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + cmd.as_std_mut().process_group(0); + } let mut child = cmd.spawn().map_err(|e| format!("spawn: {e}"))?; + let pid = child.id().ok_or("spawned process has no pid")?; + let (cancel_tx, mut cancel_rx) = oneshot::channel(); + { + let mut active = state.active.lock().await; + if matches!(active.get(&command_id), Some(None)) { + active.remove(&command_id); + drop(cancel_tx); + } else { + active.insert(command_id.clone(), Some(cancel_tx)); + } + } let mut stdout_pipe = child.stdout.take().ok_or("no stdout pipe")?; let mut stderr_pipe = child.stderr.take().ok_or("no stderr pipe")?; @@ -268,31 +318,77 @@ pub async fn tool_bash(input: BashInput) -> Result { s }); - let mut timed_out = false; - let exit_status = match tokio::time::timeout(timeout, child.wait()).await { - Ok(s) => s.map_err(|e| format!("wait: {e}"))?, - Err(_) => { - timed_out = true; + enum Finish { + Exited(std::io::Result), + TimedOut, + Cancelled, + } + let finish = tokio::select! { + status = child.wait() => Finish::Exited(status), + _ = tokio::time::sleep(timeout) => Finish::TimedOut, + _ = &mut cancel_rx => Finish::Cancelled, + }; + state.active.lock().await.remove(&command_id); + + let (exit_code, timed_out, cancelled) = match finish { + Finish::Exited(status) => ( + status + .map_err(|e| format!("wait: {e}"))? + .code() + .unwrap_or(-1), + false, + false, + ), + Finish::TimedOut => { + kill_process_group(pid); + let _ = child.start_kill(); + let _ = child.wait().await; + (124, true, false) + } + Finish::Cancelled => { + kill_process_group(pid); let _ = child.start_kill(); let _ = child.wait().await; - return Ok(BashOk { - stdout: String::new(), - stderr: format!("timeout after {}ms", timeout.as_millis()), - exit_code: 124, - timed_out, - }); + (130, false, true) } }; let stdout = stdout_task.await.unwrap_or_default(); - let stderr = stderr_task.await.unwrap_or_default(); + let mut stderr = stderr_task.await.unwrap_or_default(); + if timed_out { + stderr.push_str(&format!("\ntimeout after {}ms", timeout.as_millis())); + } + if cancelled { + stderr.push_str("\naborted by user"); + } Ok(BashOk { stdout, stderr, - exit_code: exit_status.code().unwrap_or(-1), + exit_code, timed_out, + cancelled, }) } +#[tauri::command] +pub async fn tool_bash_cancel( + command_id: String, + state: tauri::State<'_, BashState>, +) -> Result { + Ok(cancel_bash(command_id, &state).await) +} + +async fn cancel_bash(command_id: String, state: &BashState) -> bool { + let mut active = state.active.lock().await; + match active.remove(&command_id) { + Some(Some(cancel)) => cancel.send(()).is_ok(), + Some(None) => true, + None => { + active.insert(command_id, None); + true + } + } +} + // ────────────────────────────────────────────────────────────────────────── // Glob (filesystem pattern match) // ────────────────────────────────────────────────────────────────────────── @@ -454,17 +550,59 @@ mod casing_tests { stderr: String::new(), exit_code: 0, timed_out: false, + cancelled: false, }) .unwrap(); let k = keys(&v); // The exit-code badge bug: renderer compares r.exitCode !== 0. assert!(k.contains(&"exitCode".to_string()), "got {k:?}"); assert!(k.contains(&"timedOut".to_string()), "got {k:?}"); + assert!(k.contains(&"cancelled".to_string()), "got {k:?}"); assert!( !k.contains(&"exit_code".to_string()), "snake_case leaked: {k:?}" ); } + + #[cfg(unix)] + #[tokio::test] + async fn bash_cancel_kills_descendants() { + use std::sync::Arc; + + let root = std::env::temp_dir().join(format!( + "dc-rust-bash-cancel-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&root).unwrap(); + let marker = root.join("orphan-marker.txt"); + let command = format!("(sleep 0.4; echo orphan > '{}') & wait", marker.display()); + let state = Arc::new(BashState::default()); + let run_state = state.clone(); + let task = tokio::spawn(async move { + run_bash( + BashInput { + command, + cwd: Some(root.to_string_lossy().to_string()), + timeout_ms: Some(5_000), + }, + "cancel-test".to_string(), + &run_state, + ) + .await + }); + + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert!(cancel_bash("cancel-test".to_string(), &state).await); + let result = task.await.unwrap().unwrap(); + assert!(result.cancelled); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + assert!(!marker.exists(), "descendant survived cancellation"); + let _ = std::fs::remove_dir_all(marker.parent().unwrap()); + } } // ── snapshot capture path ─────────────────────────────────────────────── diff --git a/apps/desktop/src/lib/mac-agent.ts b/apps/desktop/src/lib/mac-agent.ts index a0c6cda..034e286 100644 --- a/apps/desktop/src/lib/mac-agent.ts +++ b/apps/desktop/src/lib/mac-agent.ts @@ -203,7 +203,7 @@ export async function startAgentTurn(args: StartTurnArgs): Promise): Promise { + async execute(input: Record, ctx): Promise { try { const command = pickStr(input, 'command', 'cmd'); if (!command) { return { content: 'Error: missing command', isError: true }; } - const r = (await invoke('tool_bash', { - input: { - command, - cwd: pickStr(input, 'cwd', 'working_dir'), - timeout_ms: pickNum(input, 'timeout_ms', 'timeoutMs', 'timeout'), - }, - })) as { stdout: string; stderr: string; exitCode: number; timedOut: boolean }; + if (ctx.signal?.aborted) { + return { content: 'aborted by user', isError: true }; + } + const commandId = `bash-${Date.now().toString(36)}-${bashCommandSeq++}`; + const onAbort = (): void => { + void invoke('tool_bash_cancel', { commandId }); + }; + ctx.signal?.addEventListener('abort', onAbort, { once: true }); + let r: { + stdout: string; + stderr: string; + exitCode: number; + timedOut: boolean; + cancelled: boolean; + }; + try { + r = (await invoke('tool_bash', { + commandId, + input: { + command, + cwd: pickStr(input, 'cwd', 'working_dir'), + timeout_ms: pickNum(input, 'timeout_ms', 'timeoutMs', 'timeout'), + }, + })) as typeof r; + } finally { + ctx.signal?.removeEventListener('abort', onAbort); + } const combined = (r.stdout || '') + (r.stderr ? `\n[stderr]\n${r.stderr}` : ''); return { content: combined || `(no output, exit ${r.exitCode})`, - data: { exitCode: r.exitCode, timedOut: r.timedOut }, - isError: r.exitCode !== 0, + data: { exitCode: r.exitCode, timedOut: r.timedOut, cancelled: r.cancelled }, + isError: r.exitCode !== 0 || r.cancelled, }; } catch (err) { return { content: `Error: ${(err as Error).message ?? String(err)}`, isError: true }; diff --git a/apps/lsp/src/handler.test.ts b/apps/lsp/src/handler.test.ts index 6c2b404..d365019 100644 --- a/apps/lsp/src/handler.test.ts +++ b/apps/lsp/src/handler.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest'; -import { handleMessage, type LspMessage } from './handler.js'; +import { __test, handleMessage, type LspMessage } from './handler.js'; describe('handleMessage — initialize', () => { it('returns capabilities + serverInfo + supported commands', async () => { @@ -102,6 +102,26 @@ describe('handleMessage — executeCommand', () => { expect((out[0]!.result as { aborted: boolean }).aborted).toBe(false); }); + it('deepcode.abort aborts the active turn controller', async () => { + const controller = new AbortController(); + __test.state.activeTurns.set('active-turn', controller); + const out: LspMessage[] = []; + + await handleMessage( + { + jsonrpc: '2.0', + id: 41, + method: 'workspace/executeCommand', + params: { command: 'deepcode.abort', arguments: [{ turnId: 'active-turn' }] }, + }, + (m) => out.push(m), + ); + + expect((out[0]!.result as { aborted: boolean }).aborted).toBe(true); + expect(controller.signal.aborted).toBe(true); + __test.state.activeTurns.delete('active-turn'); + }); + it('errors on unknown command', async () => { const out: LspMessage[] = []; await handleMessage( diff --git a/apps/lsp/src/handler.ts b/apps/lsp/src/handler.ts index 94dee15..b642321 100644 --- a/apps/lsp/src/handler.ts +++ b/apps/lsp/src/handler.ts @@ -16,13 +16,13 @@ interface ServerState { initialized: boolean; /** Workspace root URI from initialize. */ rootUri?: string; - /** In-flight turn IDs so /abort can cancel them. */ - activeTurns: Set; + /** In-flight turn controllers so /abort cancels provider and tools. */ + activeTurns: Map; } const state: ServerState = { initialized: false, - activeTurns: new Set(), + activeTurns: new Map(), }; const SERVER_INFO = { @@ -116,7 +116,8 @@ async function handleRunAgent( ): Promise<{ turnId: string }> { if (!args.prompt) throw new Error('prompt is required'); const turnId = `lsp-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 6)}`; - state.activeTurns.add(turnId); + const abortController = new AbortController(); + state.activeTurns.set(turnId, abortController); // Stream events back via JSON-RPC notifications. // Wired to the real agent loop — same code that drives the CLI / Mac client. @@ -131,16 +132,17 @@ async function handleRunAgent( void (async () => { try { const [ - { runAgent }, + { RuntimeHost }, { DeepSeekProvider }, - { ToolRegistry, BUILTIN_TOOLS }, + { ToolRegistry, BUILTIN_TOOLS, SAFE_READONLY_TOOLS }, { resolveCredentials, CredentialsStore }, ] = await Promise.all([ - import('@deepcode/core').then((m) => ({ runAgent: m.runAgent })), + import('@deepcode/core').then((m) => ({ RuntimeHost: m.RuntimeHost })), import('@deepcode/core').then((m) => ({ DeepSeekProvider: m.DeepSeekProvider })), import('@deepcode/core').then((m) => ({ ToolRegistry: m.ToolRegistry, BUILTIN_TOOLS: m.BUILTIN_TOOLS, + SAFE_READONLY_TOOLS: m.SAFE_READONLY_TOOLS, })), import('@deepcode/core').then((m) => ({ resolveCredentials: m.resolveCredentials, @@ -161,13 +163,18 @@ async function handleRunAgent( baseURL: creds.baseURL, }); - const result = await runAgent({ + const runtime = new RuntimeHost({ provider, tools: new ToolRegistry(BUILTIN_TOOLS), + cwd: state.rootUri ? new URL(state.rootUri).pathname : process.cwd(), + mode: 'default', + permissions: { allow: [...SAFE_READONLY_TOOLS] }, + }); + const result = await runtime.run({ systemPrompt: 'You are DeepCode, an AI coding assistant powered by DeepSeek. Be concise.', userMessage: args.prompt!, model: args.model ?? 'deepseek-chat', - cwd: state.rootUri ? new URL(state.rootUri).pathname : process.cwd(), + signal: abortController.signal, onEvent: (e) => { send({ jsonrpc: '2.0', @@ -207,8 +214,10 @@ async function handleRunAgent( function handleAbort(args: { turnId?: string }): { aborted: boolean } { if (!args.turnId) throw new Error('turnId is required'); - const had = state.activeTurns.delete(args.turnId); - return { aborted: had }; + const controller = state.activeTurns.get(args.turnId); + if (!controller) return { aborted: false }; + controller.abort(); + return { aborted: true }; } async function handleListSkills(): Promise<{ skills: unknown[] }> { diff --git a/apps/vscode/src/extension.ts b/apps/vscode/src/extension.ts index 9fc5f16..c1e5099 100644 --- a/apps/vscode/src/extension.ts +++ b/apps/vscode/src/extension.ts @@ -89,13 +89,17 @@ async function runAgent( authToken: creds.authToken, baseURL: creds.baseURL, }); - await core.runAgent({ + const runtime = new core.RuntimeHost({ provider, tools: new core.ToolRegistry(core.BUILTIN_TOOLS), + cwd, + mode: 'default', + permissions: { allow: [...core.SAFE_READONLY_TOOLS] }, + }); + await runtime.run({ systemPrompt: 'You are DeepCode, an AI coding assistant powered by DeepSeek. Be concise.', userMessage, model: 'deepseek-chat', - cwd, onEvent: (e) => { if (e.type === 'text_delta') out.append(e.text); else if (e.type === 'tool_use') out.appendLine(`\n[${e.name}] ${formatInput(e.input)}`); @@ -155,13 +159,17 @@ class ChatViewProvider implements vscode.WebviewViewProvider { baseURL: creds.baseURL, }); let buffer = ''; - await core.runAgent({ + const runtime = new core.RuntimeHost({ provider, tools: new core.ToolRegistry(core.BUILTIN_TOOLS), + cwd: this.vscodeMod.workspace.workspaceFolders?.[0]?.uri.fsPath ?? process.cwd(), + mode: 'default', + permissions: { allow: [...core.SAFE_READONLY_TOOLS] }, + }); + await runtime.run({ systemPrompt: 'You are DeepCode, an AI coding assistant powered by DeepSeek. Be concise.', userMessage: msg.text, model: 'deepseek-chat', - cwd: this.vscodeMod.workspace.workspaceFolders?.[0]?.uri.fsPath ?? process.cwd(), onEvent: (e) => { if (e.type === 'text_delta') { buffer += e.text; diff --git a/docs/design/session-format-v1.md b/docs/design/session-format-v1.md new file mode 100644 index 0000000..b51976b --- /dev/null +++ b/docs/design/session-format-v1.md @@ -0,0 +1,54 @@ +# Session Format v1 + +Status: experimental, implemented by `@deepcode/core` and the Tauri desktop backend. + +## Goals + +- one append format across CLI, headless and desktop; +- lossless reads of historical core and desktop sessions; +- no in-place mutation of legacy files; +- explicit cross-process writer ownership; +- recover an interrupted final append, but never hide middle corruption. + +## Files + +For logical session `` under `~/.deepcode/sessions/`: + +- `.v1.jsonl` is the canonical stream; +- `.writer.lock` is held with create-new semantics for each metadata rewrite or append; +- `.jsonl` and `.meta.json` are legacy, read-only inputs; +- `/snapshots/` remains the session artifact directory. + +On the first write to a legacy session, DeepCode creates the canonical stream atomically and appends there. The legacy bytes remain unchanged. Explicit user archive/delete operations may move or remove both generations. + +## Records + +The first record is metadata: + +```json +{ + "type": "session_meta", + "schema_version": 1, + "id": "…", + "cwd": "/repo", + "created_at": "…", + "updated_at": "…", + "model": "deepseek-chat" +} +``` + +Every later record is a completed message envelope: + +```json +{ "type": "message", "schema_version": 1, "role": "assistant", "content": [], "timestamp": "…" } +``` + +Streaming deltas are not persisted. Tool calls and results are stored only after they become completed content blocks in the message history. + +## Recovery and ownership + +The lock filename and create-new behavior are identical in TypeScript and Rust, so two hosts cannot silently interleave writes. A conflicting writer receives an explicit error. The lock contains diagnostic owner information and is removed when the operation exits. + +Readers prefer v1, otherwise detect either legacy layout. An invalid final record without a newline is treated as an interrupted append and ignored. Invalid JSON or an invalid message in the middle is reported with its line number and blocks normalization. + +Crash-stale lock recovery is intentionally deferred to the single-owner app-server: clients must not guess that another process is dead and steal ownership. diff --git a/packages/core/README.md b/packages/core/README.md index 818d030..6e24127 100644 --- a/packages/core/README.md +++ b/packages/core/README.md @@ -6,7 +6,7 @@ DeepCode 的 TypeScript 内核包：agent loop、DeepSeek provider、tools、con ## 当前状态 -主要模块均已有实现与测试。当前最重要的已知限制不是“缺少骨架”，而是不同 host 对 `runAgent` 的组装不一致：CLI 传入完整 permissions/hooks/sandbox/session/task services，desktop、LSP 与 VS Code 只传入其中一部分。后续通过不可绕过的 `RuntimeHost` 收敛，而不是继续增加 host-specific wiring。 +主要模块均已有实现与测试。CLI、headless、LSP 与 VS Code 已通过 `RuntimeHost` 固定 provider、tools、permissions、hooks 与 sandbox 等安全服务；`runAgent` 保留为 core 内部循环和 desktop 迁移期兼容入口。当前剩余的主要 host 差异是 desktop renderer 仍直接运行 provider/loop，后续按 packaging ADR 迁出 WebView。 关键入口： diff --git a/packages/core/src/agent.test.ts b/packages/core/src/agent.test.ts index bca01a4..6e24040 100644 --- a/packages/core/src/agent.test.ts +++ b/packages/core/src/agent.test.ts @@ -3,7 +3,7 @@ import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { runAgent } from './agent.js'; +import { runAgent as runAgentCore, type RunAgentOptions } from './agent.js'; import { HookDispatcher } from './hooks/index.js'; import { SessionManager } from './sessions/index.js'; import { ToolRegistry } from './tools/registry.js'; @@ -16,6 +16,13 @@ import type { } from './types.js'; import type { Provider, ProviderResult, ProviderRunOpts } from './providers/types.js'; +type TestRunAgentOptions = Omit & { mode?: RunAgentOptions['mode'] }; + +/** Most loop tests predate policy dispatch and focus on orchestration behavior. */ +function runAgent(opts: TestRunAgentOptions) { + return runAgentCore({ mode: 'bypassPermissions', ...opts }); +} + /** * MockProvider — pulls scripted responses from a queue, allowing fully deterministic * agent loop tests with no real API calls. @@ -119,6 +126,12 @@ describe('runAgent', () => { expect(toolEvents).toHaveLength(1); const resultEvents = events.filter((e) => e.type === 'tool_result'); expect(resultEvents).toHaveLength(1); + const steps = events.filter((e) => e.type === 'model_step_complete'); + expect(steps).toHaveLength(2); + expect(steps.map((event) => event.step)).toEqual([1, 2]); + const completed = events.filter((e) => e.type === 'turn_complete'); + expect(completed).toHaveLength(1); + expect(completed[0]).toMatchObject({ stopReason: 'end_turn' }); }); it('handles unknown tool gracefully', async () => { @@ -190,6 +203,104 @@ describe('runAgent', () => { expect(result.turnsUsed).toBe(0); }); + it('classifies a provider AbortError as an aborted run', async () => { + const ac = new AbortController(); + let markEntered!: () => void; + const entered = new Promise((resolve) => { + markEntered = resolve; + }); + const provider: Provider = { + name: 'abortable', + runTurn: async () => { + markEntered(); + await new Promise((_resolve, reject) => { + ac.signal.addEventListener( + 'abort', + () => reject(Object.assign(new Error('cancelled'), { name: 'AbortError' })), + { once: true }, + ); + }); + throw new Error('unreachable'); + }, + }; + const pending = runAgent({ + provider, + tools: new ToolRegistry(), + systemPrompt: '', + userMessage: 'go', + model: 'deepseek-chat', + cwd, + signal: ac.signal, + }); + await entered; + ac.abort(); + await expect(pending).resolves.toMatchObject({ stopReason: 'aborted', turnsUsed: 1 }); + }); + + it('fails safe for a legacy caller that omits mode and permissions', async () => { + const provider = new MockProvider([ + toolUse('writing', { + type: 'tool_use', + id: 'write-1', + name: 'Write', + input: { file_path: 'blocked.txt', content: 'must not exist' }, + }), + endTurn('done'), + ]); + + const result = await runAgentCore({ + provider, + tools: new ToolRegistry(), + systemPrompt: '', + userMessage: 'write a file', + model: 'deepseek-chat', + cwd, + } as RunAgentOptions); + + expect(result.stopReason).toBe('end_turn'); + await expect(fs.access(join(cwd, 'blocked.txt'))).rejects.toThrow(); + const toolResult = result.history + .flatMap((message) => message.content) + .find((block) => block.type === 'tool_result'); + expect(toolResult).toMatchObject({ is_error: true }); + }); + + it('aborts while an approval prompt is pending', async () => { + const ac = new AbortController(); + let approvalStarted!: () => void; + const started = new Promise((resolve) => { + approvalStarted = resolve; + }); + const provider = new MockProvider([ + toolUse('writing', { + type: 'tool_use', + id: 'write-pending', + name: 'Write', + input: { file_path: 'pending.txt', content: 'must not exist' }, + }), + ]); + + const pending = runAgentCore({ + provider, + tools: new ToolRegistry(), + systemPrompt: '', + userMessage: 'write a file', + model: 'deepseek-chat', + cwd, + signal: ac.signal, + mode: 'default', + approval: async () => { + approvalStarted(); + return new Promise(() => {}); + }, + }); + await started; + ac.abort(); + + await expect(pending).resolves.toMatchObject({ stopReason: 'aborted', turnsUsed: 1 }); + await expect(fs.access(join(cwd, 'pending.txt'))).rejects.toThrow(); + }); + it('persists messages and captures snapshots when session is provided', async () => { await fs.writeFile(join(cwd, 'edit-me.txt'), 'before'); const sessionMgr = new SessionManager({ root: sessionsRoot }); diff --git a/packages/core/src/agent.ts b/packages/core/src/agent.ts index 9d81cc6..6f386cf 100644 --- a/packages/core/src/agent.ts +++ b/packages/core/src/agent.ts @@ -8,6 +8,7 @@ import { TaskManager, type TaskRunner } from './tasks/manager.js'; import type { HookDispatcher } from './hooks/index.js'; import type { Mode } from './types.js'; import type { Provider } from './providers/types.js'; +import { resolveRuntimePolicy } from './runtime/policy.js'; // NOTE: reminders + sessions are lazy-loaded inside the loop so a browser // build (Tauri renderer) that doesn't use them avoids pulling node:fs at // module-load time. See `loadRemindersIfEnabled` and `appendSessionIfSet`. @@ -60,9 +61,8 @@ export interface RunAgentOptions { session?: { manager: SessionManager; id: string }; /** Optional: snapshot files before/after Edit/Write tool calls. */ enableSnapshots?: boolean; - /** M3: dispatch gates (mode + permissions + hooks). When set, every tool call - * goes through the gate. When unset, all tool calls are allowed (M1 behavior). */ - mode?: Mode; + /** Required dispatch mode. Every tool call goes through the central gate. */ + mode: Mode; permissions?: PermissionRules; hooks?: HookDispatcher; approval?: ApprovalCallback; @@ -141,6 +141,35 @@ export interface RunAgentResult { const DEFAULT_MAX_TURNS = 16; +async function waitForApproval( + approval: ApprovalCallback, + toolName: string, + toolInput: Record, + verdict: DispatchVerdict, + signal?: AbortSignal, +): Promise { + if (!signal) return approval(toolName, toolInput, verdict); + if (signal.aborted) return false; + + return new Promise((resolve, reject) => { + let settled = false; + const finish = (decision: ApprovalDecision): void => { + if (settled) return; + settled = true; + signal.removeEventListener('abort', onAbort); + resolve(decision); + }; + const onAbort = (): void => finish(false); + signal.addEventListener('abort', onAbort, { once: true }); + Promise.resolve(approval(toolName, toolInput, verdict)).then(finish, (error: unknown) => { + if (settled) return; + settled = true; + signal.removeEventListener('abort', onAbort); + reject(error); + }); + }); +} + /** * Tools with no side effects whose results don't depend on each other — safe to * execute concurrently within a single turn. Everything else (Edit/Write/Bash/ @@ -155,6 +184,7 @@ const READ_ONLY_TOOLS = new Set(['Read', 'Grep', 'Glob', 'WebFetch', 'WebSearch' */ export async function runAgent(opts: RunAgentOptions): Promise { const maxTurns = opts.maxTurns ?? DEFAULT_MAX_TURNS; + const runtimePolicy = resolveRuntimePolicy(opts); let history: StoredMessage[] = [...(opts.history ?? [])]; let snapshotSeq = (await opts.session?.manager.snapshots(opts.session.id))?.length ?? 0; @@ -291,8 +321,8 @@ export async function runAgent(opts: RunAgentOptions): Promise { // A background task passes its own signal so TaskStop can cancel just // that task; foreground sub-agents inherit the main run's signal. signal: signal ?? opts.signal, - mode: opts.mode, - permissions: opts.permissions, + mode: runtimePolicy.mode, + permissions: runtimePolicy.permissions, hooks: opts.hooks, sandboxConfig: opts.sandboxConfig, autoMode: opts.autoMode, @@ -420,15 +450,16 @@ export async function runAgent(opts: RunAgentOptions): Promise { } }; + const finish = async (stopReason: RunAgentResult['stopReason']): Promise => { + await fireStop(stopReason); + const message = [...history].reverse().find((candidate) => candidate.role === 'assistant'); + opts.onEvent?.({ type: 'turn_complete', stopReason, message }); + return { history, turnsUsed, usage: totalUsage, stopReason, modeSignal }; + }; + for (let turn = 0; turn < maxTurns; turn++) { if (opts.signal?.aborted) { - return { - history, - turnsUsed, - usage: totalUsage, - stopReason: 'aborted', - modeSignal, - }; + return finish('aborted'); } turnsUsed++; @@ -450,9 +481,12 @@ export async function runAgent(opts: RunAgentOptions): Promise { }, }); } catch (err) { + if (opts.signal?.aborted || (err as { name?: string }).name === 'AbortError') { + return finish('aborted'); + } const message = (err as Error).message ?? 'unknown'; opts.onEvent?.({ type: 'error', error: message }); - return { history, turnsUsed, usage: totalUsage, stopReason: 'error', modeSignal }; + return finish('error'); } totalUsage.inputTokens += result.usage.inputTokens; @@ -475,7 +509,7 @@ export async function runAgent(opts: RunAgentOptions): Promise { history.push(assistantMsg); if (opts.session) await opts.session.manager.append(opts.session.id, assistantMsg); - opts.onEvent?.({ type: 'turn_complete', message: assistantMsg }); + opts.onEvent?.({ type: 'model_step_complete', step: turnsUsed, message: assistantMsg }); // Emit any tool_use events for (const block of result.content) { @@ -491,8 +525,7 @@ export async function runAgent(opts: RunAgentOptions): Promise { // If no tool calls, we're done if (result.stopReason !== 'tool_use') { - await fireStop('end_turn'); - return { history, turnsUsed, usage: totalUsage, stopReason: 'end_turn', modeSignal }; + return finish('end_turn'); } // Execute tool calls and append a single user-role message with tool_result @@ -522,42 +555,49 @@ export async function runAgent(opts: RunAgentOptions): Promise { continue; } - // M3: dispatch gate (mode + permissions + PreToolUse hook) - if (opts.mode) { - const verdict = await dispatchToolCall({ - tool: toolUse.name, - input: toolUse.input, - mode: opts.mode, - rules: opts.permissions, - hooks: opts.hooks, - cwd: opts.cwd, - autoMode: opts.autoMode, - autoModeProvider: opts.provider, - }); - let allowed = verdict.decision === 'allow'; - if (verdict.decision === 'ask' && opts.approval) { - const decision = await opts.approval(toolUse.name, toolUse.input, verdict); - // 'always' = host has (or will) persist a matcher; treat as allow-this-call. - allowed = decision === true || decision === 'always'; - } - if (!allowed) { - resultsById.set(toolUse.id, { - type: 'tool_result', - tool_use_id: toolUse.id, - content: `Tool call blocked: ${verdict.reason}`, - is_error: true, - }); - opts.onEvent?.({ - type: 'tool_result', - id: toolUse.id, - result: { - content: verdict.reason, - isError: true, - data: { dispatchSource: verdict.source, decision: verdict.decision }, - }, - }); - continue; + // Every call goes through the central mode + permissions + hook gate. + const verdict = await dispatchToolCall({ + tool: toolUse.name, + input: toolUse.input, + mode: runtimePolicy.mode, + rules: runtimePolicy.permissions, + hooks: opts.hooks, + cwd: opts.cwd, + autoMode: opts.autoMode, + autoModeProvider: opts.provider, + }); + let allowed = verdict.decision === 'allow'; + if (verdict.decision === 'ask' && opts.approval) { + const decision = await waitForApproval( + opts.approval, + toolUse.name, + toolUse.input, + verdict, + opts.signal, + ); + if (opts.signal?.aborted) { + return finish('aborted'); } + // 'always' = host has (or will) persist a matcher; treat as allow-this-call. + allowed = decision === true || decision === 'always'; + } + if (!allowed) { + resultsById.set(toolUse.id, { + type: 'tool_result', + tool_use_id: toolUse.id, + content: `Tool call blocked: ${verdict.reason}`, + is_error: true, + }); + opts.onEvent?.({ + type: 'tool_result', + id: toolUse.id, + result: { + content: verdict.reason, + isError: true, + data: { dispatchSource: verdict.source, decision: verdict.decision }, + }, + }); + continue; } ready.push({ toolUse, handler }); @@ -718,8 +758,7 @@ export async function runAgent(opts: RunAgentOptions): Promise { } } - await fireStop('max_turns'); - return { history, turnsUsed, usage: totalUsage, stopReason: 'max_turns', modeSignal }; + return finish('max_turns'); } export const AGENT_MODULE_VERSION = '0.1.0'; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a685970..d6aa5d8 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -58,6 +58,9 @@ export { SessionManager, defaultSessionsDir, newSessionId, + readSessionRecords, + SessionCorruptionError, + SessionWriterConflictError, captureSnapshot, captureGitCheckpoint, listSnapshots, @@ -65,6 +68,9 @@ export { type SessionMeta, type SessionFiles, type SessionManagerOpts, + type SessionDiagnostic, + type SessionFormat, + type SessionReadResult, type Snapshot, } from './sessions/index.js'; @@ -172,6 +178,18 @@ export { // Agent loop's approval callback type (M3b) export type { ApprovalCallback, ApprovalDecision } from './agent.js'; +// Runtime safety policy shared by non-interactive hosts. +export { + SAFE_DEFAULT_PERMISSIONS, + SAFE_READONLY_TOOLS, + RuntimeHost, + createRuntimeHost, + resolveRuntimePolicy, + type RuntimeHostOptions, + type RuntimeTurnOptions, + type RuntimePolicyInput, +} from './runtime/index.js'; + // Skills (M4 — SKILL.md frontmatter loading + system-prompt builder; M5 — Skill tool) export { loadSkills, diff --git a/packages/core/src/ipc/protocol.ts b/packages/core/src/ipc/protocol.ts index c7adf47..a7630f7 100644 --- a/packages/core/src/ipc/protocol.ts +++ b/packages/core/src/ipc/protocol.ts @@ -4,7 +4,7 @@ // Goals: // 1. Type-safe channel names + payload shapes (no string-typed `ipc.invoke`). // 2. Stream agent events (text_delta / tool_use / tool_result / usage / -// turn_complete / error) one-way from main → renderer. +// model_step_complete / turn_complete / error) one-way from main → renderer. // 3. Same shape works for the future web SDK if we host the agent loop // out-of-process (just swap the transport). // diff --git a/packages/core/src/runtime/host.test.ts b/packages/core/src/runtime/host.test.ts new file mode 100644 index 0000000..6d9976b --- /dev/null +++ b/packages/core/src/runtime/host.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, it } from 'vitest'; +import type { Provider, ProviderResult, ProviderRunOpts } from '../providers/types.js'; +import { ToolRegistry } from '../tools/registry.js'; +import type { ToolHandler } from '../types.js'; +import { RuntimeHost } from './host.js'; + +class ScriptedProvider implements Provider { + readonly name = 'scripted'; + constructor(private readonly results: ProviderResult[]) {} + async runTurn(_opts: ProviderRunOpts): Promise { + const result = this.results.shift(); + if (!result) throw new Error('no scripted result'); + return result; + } +} + +const usage = { inputTokens: 1, outputTokens: 1, reasoningTokens: 0, cacheReadTokens: 0 }; + +function writeThenDone(): ProviderResult[] { + return [ + { + content: [ + { + type: 'tool_use', + id: 'write-1', + name: 'Write', + input: { file_path: 'x', content: 'x' }, + }, + ], + stopReason: 'tool_use', + usage, + }, + { + content: [{ type: 'text', text: 'done' }], + stopReason: 'end_turn', + usage, + }, + ]; +} + +describe('RuntimeHost', () => { + it('fails closed when the host omits policy and approval', async () => { + let executions = 0; + const write: ToolHandler = { + name: 'Write', + definition: { name: 'Write', description: 'test', inputSchema: {} }, + execute: async () => { + executions++; + return { content: 'wrote' }; + }, + }; + const host = new RuntimeHost({ + provider: new ScriptedProvider(writeThenDone()), + tools: new ToolRegistry([write]), + cwd: '/tmp', + }); + + const result = await host.run({ + systemPrompt: '', + userMessage: 'write', + model: 'deepseek-chat', + systemReminders: false, + }); + + expect(executions).toBe(0); + expect(result.history.flatMap((message) => message.content)).toContainEqual( + expect.objectContaining({ type: 'tool_result', is_error: true }), + ); + }); + + it('keeps host policy while accepting an explicit turn mode override', async () => { + let executions = 0; + const write: ToolHandler = { + name: 'Write', + definition: { name: 'Write', description: 'test', inputSchema: {} }, + execute: async () => { + executions++; + return { content: 'wrote' }; + }, + }; + const host = new RuntimeHost({ + provider: new ScriptedProvider(writeThenDone()), + tools: new ToolRegistry([write]), + cwd: '/tmp', + mode: 'default', + }); + + await host.run({ + systemPrompt: '', + userMessage: 'write', + model: 'deepseek-chat', + systemReminders: false, + modeOverride: 'bypassPermissions', + }); + + expect(executions).toBe(1); + expect(host.mode).toBe('default'); + }); + + it('requires a cwd at either boundary', () => { + const host = new RuntimeHost({ + provider: new ScriptedProvider([]), + tools: new ToolRegistry(), + }); + expect(() => host.run({ systemPrompt: '', userMessage: 'x', model: 'deepseek-chat' })).toThrow( + /requires cwd/, + ); + }); +}); diff --git a/packages/core/src/runtime/host.ts b/packages/core/src/runtime/host.ts new file mode 100644 index 0000000..9b8d866 --- /dev/null +++ b/packages/core/src/runtime/host.ts @@ -0,0 +1,84 @@ +import { + runAgent, + type ApprovalCallback, + type RunAgentOptions, + type RunAgentResult, +} from '../agent.js'; +import type { AutoModeConfig, PermissionRules, SandboxConfig } from '../config/types.js'; +import type { HookDispatcher } from '../hooks/index.js'; +import type { Provider } from '../providers/types.js'; +import type { ToolRegistry } from '../tools/registry.js'; +import type { Mode } from '../types.js'; +import { resolveRuntimePolicy } from './policy.js'; + +export interface RuntimeHostOptions { + provider: Provider; + tools: ToolRegistry; + /** Default working directory; a turn may override it explicitly. */ + cwd?: string; + /** Safe fallback is `default`, even for untyped callers. */ + mode?: Mode; + permissions?: PermissionRules; + hooks?: HookDispatcher; + approval?: ApprovalCallback; + autoMode?: AutoModeConfig; + sandboxConfig?: SandboxConfig; + pluginDirs?: string[]; +} + +type HostBoundOption = + | 'provider' + | 'tools' + | 'mode' + | 'permissions' + | 'hooks' + | 'approval' + | 'autoMode' + | 'sandboxConfig' + | 'pluginDirs'; + +export type RuntimeTurnOptions = Omit & { + cwd?: string; + /** Explicit per-turn mode change; all other safety services remain host-owned. */ + modeOverride?: Mode; + /** Per-turn UI callback; omission remains fail-closed for `ask` decisions. */ + approval?: ApprovalCallback; +}; + +/** + * Host-owned assembly boundary for the agent runtime. Clients provide turn + * input, while provider/tool/policy/hook/sandbox services stay consistent. + */ +export class RuntimeHost { + readonly mode: Mode; + readonly permissions: PermissionRules; + + constructor(private readonly options: RuntimeHostOptions) { + const policy = resolveRuntimePolicy(options); + this.mode = policy.mode; + this.permissions = policy.permissions; + } + + run(turn: RuntimeTurnOptions): Promise { + const cwd = turn.cwd ?? this.options.cwd; + if (!cwd) throw new Error('RuntimeHost requires cwd in the host or turn options'); + const { modeOverride, approval, ...agentTurn } = turn; + return runAgent({ + ...agentTurn, + provider: this.options.provider, + tools: this.options.tools, + cwd, + mode: modeOverride ?? this.mode, + permissions: this.permissions, + hooks: this.options.hooks, + approval: approval ?? this.options.approval, + autoMode: this.options.autoMode, + sandboxConfig: this.options.sandboxConfig, + pluginDirs: this.options.pluginDirs, + }); + } +} + +export function createRuntimeHost(options: RuntimeHostOptions): RuntimeHost { + return new RuntimeHost(options); +} diff --git a/packages/core/src/runtime/index.ts b/packages/core/src/runtime/index.ts new file mode 100644 index 0000000..c803ad7 --- /dev/null +++ b/packages/core/src/runtime/index.ts @@ -0,0 +1,12 @@ +export { + SAFE_DEFAULT_PERMISSIONS, + SAFE_READONLY_TOOLS, + resolveRuntimePolicy, + type RuntimePolicyInput, +} from './policy.js'; +export { + RuntimeHost, + createRuntimeHost, + type RuntimeHostOptions, + type RuntimeTurnOptions, +} from './host.js'; diff --git a/packages/core/src/runtime/policy.ts b/packages/core/src/runtime/policy.ts new file mode 100644 index 0000000..7ce3403 --- /dev/null +++ b/packages/core/src/runtime/policy.ts @@ -0,0 +1,43 @@ +import type { PermissionRules } from '../config/types.js'; +import type { Mode } from '../types.js'; + +/** + * Tools that a host without an approval UI may safely expose by default. + * Unknown, write-capable, and extension-provided tools intentionally do not + * appear here, so they resolve to `ask` and are blocked when no approval + * callback is installed. + */ +export const SAFE_READONLY_TOOLS = Object.freeze([ + 'Read', + 'Grep', + 'Glob', + 'WebFetch', + 'WebSearch', + 'AskUserQuestion', + 'ExitPlanMode', + 'ToolSearch', +] as const); + +export const SAFE_DEFAULT_PERMISSIONS: Readonly = Object.freeze({ + allow: [...SAFE_READONLY_TOOLS], +}); + +export interface RuntimePolicyInput { + mode?: Mode; + permissions?: PermissionRules; +} + +/** + * Runtime fallback for untyped/legacy callers. Typed callers must still pass + * `mode`, but JavaScript and stale integrations fail safe instead of silently + * bypassing the dispatcher. + */ +export function resolveRuntimePolicy(input: RuntimePolicyInput): { + mode: Mode; + permissions: PermissionRules; +} { + return { + mode: input.mode ?? 'default', + permissions: input.permissions ?? { allow: [...SAFE_READONLY_TOOLS] }, + }; +} diff --git a/packages/core/src/sessions/index.ts b/packages/core/src/sessions/index.ts index bd70931..fd3667d 100644 --- a/packages/core/src/sessions/index.ts +++ b/packages/core/src/sessions/index.ts @@ -7,8 +7,14 @@ export type { SessionManagerOpts } from './manager.js'; export { defaultSessionsDir, newSessionId, + readSessionRecords, + SessionCorruptionError, + SessionWriterConflictError, type SessionMeta, type SessionFiles, + type SessionDiagnostic, + type SessionFormat, + type SessionReadResult, } from './storage.js'; export { captureSnapshot, diff --git a/packages/core/src/sessions/manager.ts b/packages/core/src/sessions/manager.ts index bd6b360..c789c1d 100644 --- a/packages/core/src/sessions/manager.ts +++ b/packages/core/src/sessions/manager.ts @@ -9,7 +9,6 @@ import { newSessionId, readMessages, readMeta, - touchSession, writeMeta, type SessionMeta, } from './storage.js'; @@ -54,7 +53,6 @@ export class SessionManager { async append(sessionId: string, msg: StoredMessage): Promise { await appendMessage(this.root, sessionId, msg); - await touchSession(this.root, sessionId); } async list(): Promise { diff --git a/packages/core/src/sessions/storage.test.ts b/packages/core/src/sessions/storage.test.ts index a903716..a7f09f4 100644 --- a/packages/core/src/sessions/storage.test.ts +++ b/packages/core/src/sessions/storage.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, rm } from 'node:fs/promises'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; @@ -8,6 +8,9 @@ import { newSessionId, readMessages, readMeta, + readSessionRecords, + SessionCorruptionError, + SessionWriterConflictError, sessionFiles, writeMeta, } from './storage.js'; @@ -61,12 +64,87 @@ describe('session storage', () => { expect(got[0]?.role).toBe('user'); expect(got[1]?.role).toBe('assistant'); if (got[0]?.content[0]?.type === 'text') expect(got[0].content[0].text).toBe('hello'); + const records = (await readFile(sessionFiles(root, id).jsonlPath, 'utf8')) + .trim() + .split('\n') + .map((line) => JSON.parse(line) as Record); + expect(records[0]).toMatchObject({ type: 'session_meta', schema_version: 1, id }); + expect(records.slice(1)).toEqual([ + expect.objectContaining({ type: 'message', schema_version: 1, role: 'user' }), + expect.objectContaining({ type: 'message', schema_version: 1, role: 'assistant' }), + ]); }); it('readMessages returns [] when jsonl missing', async () => { expect(await readMessages(root, 'nope')).toEqual([]); }); + it('reads desktop header + typed message JSONL without changing its bytes', async () => { + const id = 'desktop-old'; + const path = sessionFiles(root, id).legacyJsonlPath; + const original = [ + JSON.stringify({ + type: 'session_meta', + id, + cwd: '/desktop', + created_at: 1_767_225_600, + title: 'Legacy desktop', + }), + JSON.stringify({ + type: 'message', + role: 'user', + content: [{ type: 'text', text: 'hello from desktop' }], + timestamp: '2026-01-01T00:00:01.000Z', + }), + '', + ].join('\n'); + await writeFile(path, original, 'utf8'); + + const parsed = await readSessionRecords(root, id); + expect(parsed.format).toBe('desktop-v0'); + expect(parsed.meta).toMatchObject({ id, cwd: '/desktop', title: 'Legacy desktop' }); + expect(parsed.messages).toHaveLength(1); + expect(await readFile(path, 'utf8')).toBe(original); + await expect(readMeta(root, id)).resolves.toMatchObject({ id, cwd: '/desktop' }); + }); + + it('tolerates only an incomplete final JSONL record', async () => { + const id = 'truncated-tail'; + await writeFile( + sessionFiles(root, id).legacyJsonlPath, + `${JSON.stringify({ role: 'user', content: [{ type: 'text', text: 'complete' }] })}\n{"role":"assistant"`, + 'utf8', + ); + + const parsed = await readSessionRecords(root, id); + expect(parsed.messages).toHaveLength(1); + expect(parsed.diagnostics).toEqual([ + expect.objectContaining({ line: 2, code: 'truncated_tail', fatal: false }), + ]); + await expect(readMessages(root, id)).resolves.toHaveLength(1); + }); + + it('reports middle corruption instead of silently dropping history', async () => { + const id = 'middle-corrupt'; + await writeFile( + sessionFiles(root, id).legacyJsonlPath, + [ + JSON.stringify({ role: 'user', content: [{ type: 'text', text: 'before' }] }), + '{not-json}', + JSON.stringify({ role: 'assistant', content: [{ type: 'text', text: 'after' }] }), + '', + ].join('\n'), + 'utf8', + ); + + const parsed = await readSessionRecords(root, id); + expect(parsed.messages).toHaveLength(2); + expect(parsed.diagnostics).toEqual([ + expect.objectContaining({ line: 2, code: 'invalid_json', fatal: true }), + ]); + await expect(readMessages(root, id)).rejects.toBeInstanceOf(SessionCorruptionError); + }); + it('listSessions sorts newest first', async () => { await writeMeta(root, { id: 'a', @@ -84,10 +162,72 @@ describe('session storage', () => { expect(list.map((s) => s.id)).toEqual(['b', 'a']); }); + it('listSessions includes desktop-only JSONL sessions', async () => { + await writeFile( + sessionFiles(root, 'desktop-list').legacyJsonlPath, + `${JSON.stringify({ + type: 'session_meta', + id: 'desktop-list', + cwd: '/desktop', + created_at: 1_767_225_600, + })}\n`, + 'utf8', + ); + await expect(listSessions(root)).resolves.toEqual([ + expect.objectContaining({ id: 'desktop-list', cwd: '/desktop' }), + ]); + }); + it('sessionFiles returns sensible paths', () => { const f = sessionFiles('/root', 'abc'); expect(f.metaPath).toBe('/root/abc.meta.json'); - expect(f.jsonlPath).toBe('/root/abc.jsonl'); + expect(f.jsonlPath).toBe('/root/abc.v1.jsonl'); + expect(f.legacyJsonlPath).toBe('/root/abc.jsonl'); + expect(f.writerLockPath).toBe('/root/abc.writer.lock'); expect(f.snapshotsDir).toBe('/root/abc/snapshots'); }); + + it('normalizes a legacy core session without changing legacy bytes', async () => { + const id = 'legacy-normalize'; + const files = sessionFiles(root, id); + const legacyMeta = JSON.stringify( + { + id, + cwd: '/legacy', + createdAt: '2025-01-01T00:00:00.000Z', + updatedAt: '2025-01-01T00:00:00.000Z', + }, + null, + 2, + ); + const legacyJsonl = `${JSON.stringify({ + role: 'user', + content: [{ type: 'text', text: 'old' }], + })}\n`; + await writeFile(files.metaPath, legacyMeta, 'utf8'); + await writeFile(files.legacyJsonlPath, legacyJsonl, 'utf8'); + + await appendMessage(root, id, { + role: 'assistant', + content: [{ type: 'text', text: 'new' }], + }); + + expect(await readFile(files.metaPath, 'utf8')).toBe(legacyMeta); + expect(await readFile(files.legacyJsonlPath, 'utf8')).toBe(legacyJsonl); + const parsed = await readSessionRecords(root, id); + expect(parsed.format).toBe('canonical-v1'); + expect(parsed.meta).toMatchObject({ id, cwd: '/legacy' }); + expect(parsed.messages).toHaveLength(2); + }); + + it('rejects a second writer instead of interleaving records', async () => { + const id = 'writer-owned'; + const files = sessionFiles(root, id); + await writeFile(files.writerLockPath, '{"pid":1}', 'utf8'); + + await expect( + appendMessage(root, id, { role: 'user', content: [{ type: 'text', text: 'x' }] }), + ).rejects.toBeInstanceOf(SessionWriterConflictError); + await expect(readFile(files.jsonlPath, 'utf8')).rejects.toThrow(); + }); }); diff --git a/packages/core/src/sessions/storage.ts b/packages/core/src/sessions/storage.ts index 6e1faf1..32f958b 100644 --- a/packages/core/src/sessions/storage.ts +++ b/packages/core/src/sessions/storage.ts @@ -1,13 +1,49 @@ -// Session storage — jsonl persistence at ~/.deepcode/sessions/.jsonl -// Each line is one StoredMessage envelope. +// Session storage — canonical v1 JSONL plus read-only legacy compatibility. // Spec: docs/DEVELOPMENT_PLAN.md §3.5 -import { promises as fs, createReadStream } from 'node:fs'; +import { promises as fs } from 'node:fs'; import { homedir } from 'node:os'; import { dirname, join } from 'node:path'; -import { createInterface } from 'node:readline'; import type { StoredMessage } from '../types.js'; +export type SessionFormat = 'canonical-v1' | 'core-v0' | 'desktop-v0' | 'empty'; + +export interface SessionDiagnostic { + line: number; + code: 'truncated_tail' | 'invalid_json' | 'invalid_message'; + message: string; + fatal: boolean; +} + +export interface SessionReadResult { + format: SessionFormat; + meta: SessionMeta | null; + messages: StoredMessage[]; + diagnostics: SessionDiagnostic[]; +} + +export class SessionCorruptionError extends Error { + constructor( + readonly sessionId: string, + readonly diagnostics: SessionDiagnostic[], + ) { + super( + `Session ${sessionId} is corrupted at ${diagnostics + .filter((d) => d.fatal) + .map((d) => `line ${d.line}: ${d.message}`) + .join('; ')}`, + ); + this.name = 'SessionCorruptionError'; + } +} + +export class SessionWriterConflictError extends Error { + constructor(readonly sessionId: string) { + super(`Session ${sessionId} already has an active writer`); + this.name = 'SessionWriterConflictError'; + } +} + export interface SessionMeta { id: string; cwd: string; @@ -22,32 +58,53 @@ export function defaultSessionsDir(): string { } export interface SessionFiles { + /** Read-only core v0 metadata sidecar. */ metaPath: string; + /** Canonical v1 stream used for all new writes. */ jsonlPath: string; + /** Read-only core/desktop v0 stream. */ + legacyJsonlPath: string; + writerLockPath: string; snapshotsDir: string; } export function sessionFiles(root: string, sessionId: string): SessionFiles { return { metaPath: join(root, `${sessionId}.meta.json`), - jsonlPath: join(root, `${sessionId}.jsonl`), + jsonlPath: join(root, `${sessionId}.v1.jsonl`), + legacyJsonlPath: join(root, `${sessionId}.jsonl`), + writerLockPath: join(root, `${sessionId}.writer.lock`), snapshotsDir: join(root, sessionId, 'snapshots'), }; } export async function writeMeta(root: string, meta: SessionMeta): Promise { const files = sessionFiles(root, meta.id); - await fs.mkdir(dirname(files.metaPath), { recursive: true }); - await fs.writeFile(files.metaPath, JSON.stringify(meta, null, 2), 'utf8'); + await withWriterLock(files, meta.id, async () => { + let messages: StoredMessage[] = []; + try { + const parsed = await readRecordsFromPath(files.jsonlPath); + const fatal = parsed.diagnostics.filter((diagnostic) => diagnostic.fatal); + if (fatal.length > 0) throw new SessionCorruptionError(meta.id, fatal); + messages = parsed.messages; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + await writeCanonical(files.jsonlPath, meta, messages); + }); } export async function readMeta(root: string, sessionId: string): Promise { const files = sessionFiles(root, sessionId); + const records = await readSessionRecords(root, sessionId); + if (records.format === 'canonical-v1' && records.meta) return records.meta; try { const raw = await fs.readFile(files.metaPath, 'utf8'); return JSON.parse(raw) as SessionMeta; } catch (err) { - if ((err as NodeJS.ErrnoException).code === 'ENOENT') return null; + if ((err as NodeJS.ErrnoException).code === 'ENOENT') { + return records.meta; + } throw err; } } @@ -58,28 +115,247 @@ export async function appendMessage( message: StoredMessage, ): Promise { const files = sessionFiles(root, sessionId); - await fs.mkdir(dirname(files.jsonlPath), { recursive: true }); - await fs.appendFile(files.jsonlPath, JSON.stringify(message) + '\n', 'utf8'); + await withWriterLock(files, sessionId, async () => { + await ensureCanonical(sessionId, files); + await fs.appendFile(files.jsonlPath, JSON.stringify(messageRecord(message)) + '\n', 'utf8'); + }); } export async function readMessages(root: string, sessionId: string): Promise { - const files = sessionFiles(root, sessionId); + const result = await readSessionRecords(root, sessionId); + const fatal = result.diagnostics.filter((diagnostic) => diagnostic.fatal); + if (fatal.length > 0) throw new SessionCorruptionError(sessionId, fatal); + return result.messages; +} + +function isStoredMessage(value: unknown): value is StoredMessage { + if (!value || typeof value !== 'object') return false; + const record = value as Record; + return (record.role === 'user' || record.role === 'assistant') && Array.isArray(record.content); +} + +function desktopMeta(value: Record, updatedAt: string): SessionMeta | null { + if (value.type !== 'session_meta' || typeof value.id !== 'string') return null; + const createdAt = + typeof value.created_at === 'number' + ? new Date(value.created_at * 1000).toISOString() + : typeof value.created_at === 'string' + ? value.created_at + : updatedAt; + const normalizedUpdatedAt = + value.schema_version === 1 && typeof value.updated_at === 'string' + ? value.updated_at + : updatedAt; + return { + id: value.id, + cwd: typeof value.cwd === 'string' ? value.cwd : '', + createdAt, + updatedAt: normalizedUpdatedAt, + model: typeof value.model === 'string' ? value.model : undefined, + title: typeof value.title === 'string' ? value.title : undefined, + }; +} + +function metaRecord(meta: SessionMeta): Record { + return { + type: 'session_meta', + schema_version: 1, + id: meta.id, + cwd: meta.cwd, + created_at: meta.createdAt, + updated_at: meta.updatedAt, + ...(meta.model ? { model: meta.model } : {}), + ...(meta.title ? { title: meta.title } : {}), + }; +} + +function messageRecord(message: StoredMessage): Record { + return { type: 'message', schema_version: 1, ...message }; +} + +async function writeCanonical( + path: string, + meta: SessionMeta, + messages: StoredMessage[], +): Promise { + const tempPath = `${path}.${process.pid}.${Date.now()}.tmp`; + const body = [metaRecord(meta), ...messages.map(messageRecord)] + .map((record) => JSON.stringify(record)) + .join('\n'); + await fs.writeFile(tempPath, body + '\n', { encoding: 'utf8', flag: 'wx' }); + await fs.rename(tempPath, path); +} + +async function withWriterLock( + files: SessionFiles, + sessionId: string, + operation: () => Promise, +): Promise { + await fs.mkdir(dirname(files.writerLockPath), { recursive: true }); + let lock: Awaited>; + try { + lock = await fs.open(files.writerLockPath, 'wx'); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'EEXIST') { + throw new SessionWriterConflictError(sessionId); + } + throw error; + } + try { + await lock.writeFile(JSON.stringify({ pid: process.pid, createdAt: new Date().toISOString() })); + return await operation(); + } finally { + await lock.close(); + await fs.unlink(files.writerLockPath).catch(() => undefined); + } +} + +async function ensureCanonical(sessionId: string, files: SessionFiles): Promise { try { await fs.access(files.jsonlPath); + return; } catch { - return []; + // Normalize below while holding the writer lock. + } + + const legacy = await readRecordsFromPath(files.legacyJsonlPath).catch( + (error: NodeJS.ErrnoException) => { + if (error.code === 'ENOENT') { + return { format: 'empty', meta: null, messages: [], diagnostics: [] } as SessionReadResult; + } + throw error; + }, + ); + const fatal = legacy.diagnostics.filter((diagnostic) => diagnostic.fatal); + if (fatal.length > 0) throw new SessionCorruptionError(sessionId, fatal); + + let sidecar: SessionMeta | null = null; + try { + sidecar = JSON.parse(await fs.readFile(files.metaPath, 'utf8')) as SessionMeta; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + const now = new Date().toISOString(); + const meta = legacy.meta ?? + sidecar ?? { + id: sessionId, + cwd: '', + createdAt: now, + updatedAt: now, + }; + await writeCanonical(files.jsonlPath, meta, legacy.messages); +} + +/** Parse both historical JSONL layouts without modifying either one. */ +export async function readSessionRecords( + root: string, + sessionId: string, +): Promise { + const files = sessionFiles(root, sessionId); + try { + return await readRecordsFromPath(files.jsonlPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + try { + return await readRecordsFromPath(files.legacyJsonlPath); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') { + return { format: 'empty', meta: null, messages: [], diagnostics: [] }; + } + throw error; } - const out: StoredMessage[] = []; - const rl = createInterface({ input: createReadStream(files.jsonlPath, { encoding: 'utf8' }) }); - for await (const line of rl) { +} + +async function readRecordsFromPath(path: string): Promise { + const [raw, stat] = await Promise.all([fs.readFile(path, 'utf8'), fs.stat(path)]); + const updatedAt = stat.mtime.toISOString(); + + const lines = raw.split('\n'); + let lastContentIndex = -1; + for (let index = lines.length - 1; index >= 0; index--) { + if (lines[index]!.trim().length > 0) { + lastContentIndex = index; + break; + } + } + const messages: StoredMessage[] = []; + const diagnostics: SessionDiagnostic[] = []; + let meta: SessionMeta | null = null; + let format: SessionFormat = 'empty'; + + for (let index = 0; index < lines.length; index++) { + const line = lines[index]!; if (!line.trim()) continue; + let value: unknown; try { - out.push(JSON.parse(line) as StoredMessage); - } catch { - // skip malformed lines (forward-compat) + value = JSON.parse(line); + } catch (error) { + const isTruncatedTail = index === lastContentIndex && !raw.endsWith('\n'); + diagnostics.push({ + line: index + 1, + code: isTruncatedTail ? 'truncated_tail' : 'invalid_json', + message: isTruncatedTail + ? 'ignored an incomplete final JSONL record' + : `invalid JSON: ${(error as Error).message}`, + fatal: !isTruncatedTail, + }); + continue; + } + if (!value || typeof value !== 'object') { + diagnostics.push({ + line: index + 1, + code: 'invalid_message', + message: 'record must be a JSON object', + fatal: true, + }); + continue; + } + + const record = value as Record; + if (record.type === 'session_meta') { + format = record.schema_version === 1 ? 'canonical-v1' : 'desktop-v0'; + meta ??= desktopMeta(record, updatedAt); + continue; } + if (record.type === 'message') { + if (format !== 'canonical-v1') { + format = record.schema_version === 1 ? 'canonical-v1' : 'desktop-v0'; + } + if (isStoredMessage(record)) { + messages.push({ + role: record.role, + content: record.content, + timestamp: typeof record.timestamp === 'string' ? record.timestamp : undefined, + }); + } else { + diagnostics.push({ + line: index + 1, + code: 'invalid_message', + message: 'message record has an invalid role or content array', + fatal: true, + }); + } + continue; + } + if (record.type === undefined) { + format = 'core-v0'; + if (isStoredMessage(record)) messages.push(record); + else { + diagnostics.push({ + line: index + 1, + code: 'invalid_message', + message: 'bare record has an invalid role or content array', + fatal: true, + }); + } + continue; + } + // Unknown typed records are reserved for forward-compatible lifecycle + // items. They are not messages and are intentionally ignored. } - return out; + + return { format, meta, messages, diagnostics }; } export async function listSessions(root: string): Promise { @@ -89,12 +365,16 @@ export async function listSessions(root: string): Promise { return []; } const entries = await fs.readdir(root); - const metaFiles = entries.filter((f) => f.endsWith('.meta.json')); + const ids = new Set(); + for (const entry of entries) { + if (entry.endsWith('.meta.json')) ids.add(entry.slice(0, -'.meta.json'.length)); + else if (entry.endsWith('.v1.jsonl')) ids.add(entry.slice(0, -'.v1.jsonl'.length)); + else if (entry.endsWith('.jsonl')) ids.add(entry.slice(0, -'.jsonl'.length)); + } const metas = await Promise.all( - metaFiles.map(async (f) => { + [...ids].map(async (id) => { try { - const raw = await fs.readFile(join(root, f), 'utf8'); - return JSON.parse(raw) as SessionMeta; + return await readMeta(root, id); } catch { return null; } @@ -105,13 +385,6 @@ export async function listSessions(root: string): Promise { .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)); } -export async function touchSession(root: string, sessionId: string): Promise { - const meta = await readMeta(root, sessionId); - if (!meta) return; - meta.updatedAt = new Date().toISOString(); - await writeMeta(root, meta); -} - export function newSessionId(): string { // Short prefix + uuid-ish — collision risk is negligible at this scale. const ts = new Date() diff --git a/packages/core/src/tools/bash.test.ts b/packages/core/src/tools/bash.test.ts index b52ae08..71b33bc 100644 --- a/packages/core/src/tools/bash.test.ts +++ b/packages/core/src/tools/bash.test.ts @@ -1,5 +1,5 @@ import type { ChildProcess } from 'node:child_process'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { access, mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; @@ -48,6 +48,23 @@ describe('BashTool', () => { expect(r.content).toMatch(/killed by timeout/i); }, 5000); + it('aborts the foreground process tree', async () => { + if (process.platform === 'win32') return; + const marker = join(tmp, 'orphan-marker.txt'); + const ac = new AbortController(); + const pending = BashTool.execute( + { command: `(sleep 0.4; echo orphan > "${marker}") & wait`, timeout: 5_000 }, + { cwd: tmp, signal: ac.signal }, + ); + setTimeout(() => ac.abort(), 50); + + const result = await pending; + expect(result.isError).toBe(true); + expect(result.content).toMatch(/aborted by user/i); + await new Promise((resolve) => setTimeout(resolve, 500)); + await expect(access(marker)).rejects.toThrow(); + }, 5000); + it('run_in_background returns immediately with a log path that fills in', async () => { const r = await BashTool.execute( { command: 'echo bg-output-here', run_in_background: true }, diff --git a/packages/core/src/tools/bash.ts b/packages/core/src/tools/bash.ts index 36d3313..3b8c5dd 100644 --- a/packages/core/src/tools/bash.ts +++ b/packages/core/src/tools/bash.ts @@ -6,7 +6,7 @@ // that can't be set up (e.g. can't bind the DNS proxy on :53), fail CLOSED to // deny-all-net rather than running unrestricted. -import { spawn } from 'node:child_process'; +import { spawn, type ChildProcess } from 'node:child_process'; import { promises as fs } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -39,6 +39,7 @@ type SandboxCtx = ToolContext & { const DEFAULT_TIMEOUT_MS = 120_000; // 2 minutes const MAX_OUTPUT_BYTES = 30_000; +type TerminationReason = 'timeout' | 'aborted'; // Monotonic suffix so two background spawns in the same millisecond from the // same pid don't collide on a log filename. @@ -54,7 +55,7 @@ function capStream(s: string, label: string): string { function summarize( stdout: string, stderr: string, - killed: boolean, + terminationReason: TerminationReason | undefined, code: number | null, timeoutMs: number, note?: string, @@ -63,15 +64,39 @@ function summarize( if (note) parts.push(note); if (stdout) parts.push(`\n${stdout}\n`); if (stderr) parts.push(`\n${stderr}\n`); - if (killed) parts.push(`[killed by timeout after ${timeoutMs}ms]`); + if (terminationReason === 'timeout') parts.push(`[killed by timeout after ${timeoutMs}ms]`); + if (terminationReason === 'aborted') parts.push('[aborted by user]'); parts.push(`exit: ${code ?? 'unknown'}`); return { content: parts.join('\n'), - data: { exitCode: code, killed, stdoutBytes: stdout.length, stderrBytes: stderr.length }, - isError: killed || (code !== null && code !== 0), + data: { + exitCode: code, + killed: terminationReason !== undefined, + terminationReason, + stdoutBytes: stdout.length, + stderrBytes: stderr.length, + }, + isError: terminationReason !== undefined || (code !== null && code !== 0), }; } +/** Kill the whole foreground process group on POSIX, not just its shell. */ +function killProcessTree(child: ChildProcess, signal: NodeJS.Signals): void { + if (process.platform !== 'win32' && child.pid !== undefined) { + try { + process.kill(-child.pid, signal); + return; + } catch { + // The group may already have exited; fall back to the direct child. + } + } + try { + child.kill(signal); + } catch { + // Process already exited. + } +} + /** * Foreground run under the slirp4netns selective-network sandbox. Rejects with * NetworkSandboxUnavailable if setup fails (caller falls back to deny-all-net). @@ -87,7 +112,7 @@ async function runForegroundNet( return new Promise((resolve) => { let stdout = ''; let stderr = ''; - let killed = false; + let terminationReason: TerminationReason | undefined; let settled = false; const finish = (r: ToolResult): void => { if (!settled) { @@ -96,11 +121,11 @@ async function runForegroundNet( } }; const timer = setTimeout(() => { - killed = true; + terminationReason = 'timeout'; void handle.close(); }, timeoutMs); const onAbort = (): void => { - killed = true; + terminationReason = 'aborted'; void handle.close(); }; ctx.signal?.addEventListener('abort', onAbort, { once: true }); @@ -114,7 +139,7 @@ async function runForegroundNet( .then((code) => { clearTimeout(timer); ctx.signal?.removeEventListener('abort', onAbort); - finish(summarize(stdout, stderr, killed, code, timeoutMs)); + finish(summarize(stdout, stderr, terminationReason, code, timeoutMs)); }) .catch((err: unknown) => { clearTimeout(timer); @@ -153,6 +178,13 @@ export const BashTool: ToolHandler = { if (!input?.command || typeof input.command !== 'string') { return { content: 'Error: command is required (string).', isError: true }; } + if (ctx.signal?.aborted) { + return { + content: '[aborted by user]', + isError: true, + data: { terminationReason: 'aborted' }, + }; + } const timeoutMs = Math.max(1_000, input.timeout ?? DEFAULT_TIMEOUT_MS); // M3.5: wrap under platform sandbox if configured. ctx.sandboxConfig is @@ -239,20 +271,33 @@ export const BashTool: ToolHandler = { return new Promise((resolvePromise) => { const child = spawn(wrapped.command, wrapped.args, { cwd: ctx.cwd, - signal: ctx.signal, + detached: process.platform !== 'win32', }); let stdout = ''; let stderr = ''; - let killed = false; - const timer = setTimeout(() => { - killed = true; - // SIGKILL + destroy pipes — on Ubuntu CI, dash leaves orphaned children - // whose inherited stdout/stderr fds keep `close` from firing on the - // parent. Destroying the pipes forces close. - child.kill('SIGKILL'); + let terminationReason: TerminationReason | undefined; + let settled = false; + const finish = (result: ToolResult): void => { + if (settled) return; + settled = true; + clearTimeout(timer); + ctx.signal?.removeEventListener('abort', onAbort); + resolvePromise(result); + }; + const terminate = (reason: TerminationReason): void => { + if (terminationReason) return; + terminationReason = reason; + killProcessTree(child, 'SIGKILL'); + // Descendants can inherit these descriptors; destroying them also + // prevents an orphan from keeping the Promise open indefinitely. child.stdout?.destroy(); child.stderr?.destroy(); + }; + const timer = setTimeout(() => { + terminate('timeout'); }, timeoutMs); + const onAbort = (): void => terminate('aborted'); + ctx.signal?.addEventListener('abort', onAbort, { once: true }); child.stdout.on('data', (chunk: Buffer) => { stdout = capStream(stdout + chunk.toString('utf8'), 'stdout'); @@ -262,16 +307,14 @@ export const BashTool: ToolHandler = { }); child.on('error', (err) => { - clearTimeout(timer); - resolvePromise({ + finish({ content: `Error spawning command: ${err.message}`, isError: true, }); }); child.on('close', (code) => { - clearTimeout(timer); - resolvePromise(summarize(stdout, stderr, killed, code, timeoutMs, failNote)); + finish(summarize(stdout, stderr, terminationReason, code, timeoutMs, failNote)); }); }); }, diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index f49507d..5e9fbb7 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -181,7 +181,14 @@ export type AgentEvent = | { type: 'thinking_delta'; text: string } | { type: 'tool_use'; id: string; name: string; input: Record } | { type: 'tool_result'; id: string; result: ToolResult } - | { type: 'turn_complete'; message: StoredMessage } + /** One provider round-trip completed; a user turn may contain many steps. */ + | { type: 'model_step_complete'; step: number; message: StoredMessage } + /** The whole user turn reached one terminal state. Emitted exactly once. */ + | { + type: 'turn_complete'; + stopReason: 'end_turn' | 'max_turns' | 'aborted' | 'error'; + message?: StoredMessage; + } | { type: 'usage'; inputTokens: number; diff --git a/packages/core/src/worktree/index.test.ts b/packages/core/src/worktree/index.test.ts index c20edbf..f47e52f 100644 --- a/packages/core/src/worktree/index.test.ts +++ b/packages/core/src/worktree/index.test.ts @@ -73,6 +73,22 @@ describe('createWorktree / removeWorktree', () => { expect(await fs.readFile(join(h.path, 'a.txt'), 'utf8')).toBe('A'); await removeWorktree(h); await expect(fs.access(h.path)).rejects.toThrow(); + const branch = spawnSync('git', ['-C', src, 'rev-parse', '--verify', h.branch], { + encoding: 'utf8', + env: cleanGitEnv(), + }); + expect(branch.status).toBe(0); + }); + + it('refuses to remove a dirty worktree', async () => { + const h = await createWorktree({ source: src, parentDir: parent }); + const changed = join(h.path, 'a.txt'); + await fs.writeFile(changed, 'unsaved work'); + + await expect(removeWorktree(h)).rejects.toThrow(/worktree remove/); + expect(await fs.readFile(changed, 'utf8')).toBe('unsaved work'); + + runOrFail('git', ['worktree', 'remove', '--force', h.path], src); }); it('honors baseRef from config', async () => { diff --git a/packages/core/src/worktree/index.ts b/packages/core/src/worktree/index.ts index ba24fec..ab1483f 100644 --- a/packages/core/src/worktree/index.ts +++ b/packages/core/src/worktree/index.ts @@ -20,6 +20,8 @@ export interface WorktreeHandle { branch: string; /** Source repo path. */ source: string; + /** Untracked symlinks created by DeepCode and safe to unlink on removal. */ + managedSymlinks?: string[]; } export interface CreateWorktreeOpts { @@ -57,6 +59,7 @@ export async function createWorktree(opts: CreateWorktreeOpts): Promise { try { @@ -86,12 +91,17 @@ export async function removeWorktree(handle: WorktreeHandle): Promise { } catch { return; } - runGit(handle.source, ['worktree', 'remove', '--force', handle.path]); - // Delete the branch (best-effort) - spawnSync('git', ['-C', handle.source, 'branch', '-D', handle.branch], { - stdio: 'pipe', - env: gitSpawnEnv(), - }); + // These are the only untracked paths DeepCode itself creates. Remove them + // only if they are still symlinks; a user-replaced directory/file is data and + // must make the subsequent clean-worktree check fail. + for (const path of handle.managedSymlinks ?? []) { + try { + if ((await fs.lstat(path)).isSymbolicLink()) await fs.unlink(path); + } catch { + // Missing or unreadable managed link: let Git perform the final check. + } + } + runGit(handle.source, ['worktree', 'remove', handle.path]); } function runGit(cwd: string, args: string[]): void {