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/CODEX_ALIGNMENT_PLAN.md b/docs/CODEX_ALIGNMENT_PLAN.md index 3cf0eb3..8a03f43 100644 --- a/docs/CODEX_ALIGNMENT_PLAN.md +++ b/docs/CODEX_ALIGNMENT_PLAN.md @@ -267,11 +267,13 @@ model tool call ### PR 4 — Desktop runtime packaging ADR/spike -- 在 bundled Node sidecar、单可执行 sidecar、Rust runtime 与过渡 renderer loop 中做可发布选择。 -- 必须在签名后的 `.app`、无系统 Node 环境中证明启动、退出、取消、升级和恢复。 -- 同时决定单客户端 stdio 还是多客户端 daemon/socket，不提前承诺 active turn 跨端附着。 +- 已由 `docs/adr/0001-desktop-runtime-sidecar.md` 决定采用 Tauri 监督的 target-specific Node 22 + sidecar、单文件 app-server resource 与单客户端 stdio；不承诺 active turn 跨端附着。 +- 可复现 probe 必须在无系统 Node 的 PATH 中证明协议握手，并报告 runtime 体积与冷启动。 +- 签名、notarization、取消、升级和恢复仍是迁移 renderer 前的 release gate，不能用本地 ad-hoc + 签名冒充发布验证。 -验收：形成 ADR、可复现 spike、安装包体积/冷启动/签名结果和失败回滚路径。 +验收：形成 ADR、可复现 spike、本机构建体积/冷启动证据、发布签名 gate 和失败回滚路径。 ### PR 5 — App-server 垂直切片与 CLI diff --git a/docs/adr/0001-desktop-runtime-sidecar.md b/docs/adr/0001-desktop-runtime-sidecar.md new file mode 100644 index 0000000..f38f9fe --- /dev/null +++ b/docs/adr/0001-desktop-runtime-sidecar.md @@ -0,0 +1,120 @@ +# ADR 0001: Package the desktop runtime as a supervised Node sidecar + +- Status: Accepted +- Date: 2026-08-01 +- Decision owners: DeepCode runtime maintainers +- Roadmap: `docs/CODEX_ALIGNMENT_PLAN.md`, PR 4 + +## Context + +The Tauri renderer currently imports selected pieces of `@deepcode/core`, creates the +`DeepSeekProvider`, and calls `runAgent` inside the WebView. The renderer receives API credentials +from Rust, cannot load core modules that depend on Node APIs, and disables runtime features such as +hooks and system reminders. Each workaround widens the behavior gap between desktop and the other +clients. + +Moving the existing TypeScript runtime behind a process boundary is therefore required. An +installed desktop app cannot assume that Node is present on the user's `PATH`. + +Tauri 2 supports target-specific external binaries for precisely this class of dependency; its +[sidecar documentation](https://v2.tauri.app/develop/sidecar/) describes embedding executables so +users do not need to install runtimes such as Node or Python. Node also offers +[single-executable applications](https://nodejs.org/download/release/latest-jod/docs/api/single-executable-applications.html), +but the Node 22 feature is still marked active development, accepts one embedded CommonJS script, +and requires a separate platform-specific injection step. + +## Decision + +DeepCode will use a **Tauri-supervised, target-specific Node 22 sidecar** as the transitional +desktop runtime boundary. + +- The release build pins and checksum-verifies an official Node 22 binary for each supported + target. It does not copy an arbitrary developer-machine runtime into a release. +- The app server is bundled into one CommonJS file and included as an app resource. Its production + dependency graph must not resolve modules from a user-controlled working directory. +- Tauri packages the runtime through `bundle.externalBin`. The filename follows Tauri's required + target suffix convention, such as `deepcode-runtime-aarch64-apple-darwin`. +- Rust owns sidecar startup, shutdown, crash reporting, and stdio. The renderer communicates only + through the versioned line-delimited JSON protocol. +- The sidecar owns `RuntimeHost`, provider creation, credentials, configuration, hooks, MCP, + permissions, sandboxing, session persistence, and the agent lifecycle. Credentials never enter + WebView memory. +- Version 1 uses one child process per desktop app and a single-client stdio connection. A shared + multi-client daemon requires a later decision covering socket authentication, ownership, + subscriptions, and backpressure. +- Release signing explicitly signs the nested runtime before the outer `.app`; CI then performs + deep strict signature verification before notarization. + +The bundled runtime is an implementation detail behind the protocol. It may later become a Node +SEA, another compatible JavaScript runtime, or a native implementation without changing clients. + +## Spike evidence + +`pnpm spike:desktop-sidecar` creates an isolated temporary layout, copies and target-thins the +current runtime, strips it where supported, ad-hoc signs the resulting Mach-O, clears `PATH`, and +performs an `initialize` handshake over stdio. The script fails unless the child reports protocol +version 1 without discovering a system runtime. + +On the 2026-08-01 Apple Silicon development host: + +| Measurement | Result | +| ------------------------------ | ------------------- | +| Existing unsigned Tauri `.app` | 6,733,824 bytes | +| Local universal Node runtime | 237,619,616 bytes | +| Target-thin arm64 runtime | 117,655,968 bytes | +| Thin, stripped runtime | 108,412,080 bytes | +| Isolated protocol handshake | passed with no PATH | +| Cold isolated handshake | 1.02 seconds | + +These are topology measurements, not release promises. The local runtime is Homebrew's universal +Node 24 build, whereas production will use a pinned target-specific Node 22 distribution. A signed +and notarized artifact cannot be verified locally without release credentials, so that remains a +release-CI gate rather than a claimed spike result. + +## Options considered + +### Keep the agent loop in the WebView + +Rejected. It exposes credentials to renderer JavaScript, forces browser-compatible subsets of +core, duplicates host assembly, and cannot provide a trustworthy long-running backend. + +### Node single-executable application + +Deferred. SEA does not remove the Node runtime size, and Node 22 adds CommonJS bundling, blob +injection, fuse mutation, and post-injection signing to the release chain while the feature remains +in active development. Reconsider after the app-server bundle is stable and the SEA build is +reproducible on all release targets. + +### Bun-compiled sidecar + +Deferred. It could simplify single-file creation, but it adds a second JavaScript runtime and new +compatibility risk for Node-heavy core modules. It may be evaluated later against the same protocol +and test corpus. + +### Rewrite the runtime in Rust + +Rejected for this migration. It preserves a small app but duplicates the provider, tool, hook, +plugin, MCP, session, and policy implementations before their shared semantics are stable. + +### Require a system Node installation + +Rejected. It makes the desktop artifact non-self-contained and introduces unsupported version and +PATH variation. + +## Consequences and gates + +The installed app becomes materially larger. That cost is accepted to eliminate the higher-risk +renderer runtime split, but release PRs must report uncompressed `.app` size, compressed DMG size, +and cold handshake time. The first production sidecar is blocked if any of these gates fail: + +- a clean environment with an empty `PATH` cannot initialize the server; +- the bundled app resolves server code or dependencies outside its signed resources; +- credentials or provider calls remain in renderer bundles; +- interrupting the desktop turn does not stop the backend operation; +- nested and outer signatures fail `codesign --verify --deep --strict`; +- notarization or stapling fails; +- the final DMG exceeds 100 MB without a separate maintainers' decision. + +Rollback is a configuration-level switch to the existing renderer runtime during the experimental +phase. The fallback must be removed once sidecar parity, credential isolation, and signed release +gates pass; it is not a permanent dual architecture. diff --git a/docs/design/runtime-protocol-v1.md b/docs/design/runtime-protocol-v1.md new file mode 100644 index 0000000..fda1253 --- /dev/null +++ b/docs/design/runtime-protocol-v1.md @@ -0,0 +1,64 @@ +# Experimental runtime protocol v1 + +Status: experimental +Owner: runtime architecture +Implementation: `packages/protocol` + +## Purpose + +DeepCode clients currently integrate with the agent loop through surface-specific callbacks. The +experimental runtime protocol introduces a transport-neutral boundary that can be shared by the +CLI, desktop, VS Code, LSP, and a future local app server. It is intentionally independent of +Node.js, Tauri, React, and model providers. + +Version 1 proves lifecycle semantics and record/replay behavior. It is not yet a promise that +existing clients will migrate without a negotiated version check. + +## Lifecycle + +A thread contains ordered turns. A thread may have at most one `in_progress` turn. A turn starts +with a completed `user_message` item and reaches exactly one terminal state: + +```text + +-> completed +in_progress --------+-> interrupted + +-> failed +``` + +Terminal transitions are idempotent. Once a turn is terminal, later terminal requests return the +stored terminal snapshot and no second terminal event is emitted. Completed items cannot be added +to a terminal turn. + +## Durable and transient events + +Durable events describe state that can be reconstructed after a process restart: + +- `thread.started` +- `turn.started` +- `item.completed` +- `turn.completed` +- `turn.interrupted` +- `turn.failed` + +`item.delta` is transient. A delta is suitable for live UI streaming, but it is neither saved by +the thread store nor included in protocol recordings. A client that reconnects reads the latest +completed-item snapshot instead of replaying partial text. + +State is saved before its corresponding durable event is emitted. A consumer may therefore read +the referenced thread immediately after receiving an event. + +## Initialization and compatibility + +Clients call `initialize` before other methods and inspect both `protocolVersion` and advertised +capabilities. Version 1 advertises thread resume, turn interruption, completed-item persistence, +and transient deltas. + +Unknown methods and non-object request parameters are rejected by the line-oriented JSON codec. +Future incompatible lifecycle changes require a new protocol version; optional behavior should be +introduced through capabilities. + +## Current scope + +The in-memory store and codec are reference implementations used by contract tests. Production +transport, authorization, persistent storage, backpressure, and wiring to `RuntimeHost` belong to +the app-server phase of the alignment roadmap. 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/package.json b/package.json index 6f0a30f..ebe6be2 100644 --- a/package.json +++ b/package.json @@ -23,6 +23,7 @@ "format": "prettier --write \"**/*.{ts,tsx,json,md,yml,yaml}\"", "format:check": "prettier --check \"**/*.{ts,tsx,json,md,yml,yaml}\"", "docs:check": "node scripts/check-docs.mjs", + "spike:desktop-sidecar": "node scripts/spike-desktop-sidecar.mjs", "clean": "pnpm -r clean", "prepare": "husky || true" }, 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 { diff --git a/packages/protocol/README.md b/packages/protocol/README.md new file mode 100644 index 0000000..1b3f572 --- /dev/null +++ b/packages/protocol/README.md @@ -0,0 +1,10 @@ +# @deepcode/protocol + +Experimental, transport-neutral lifecycle contracts for DeepCode runtimes and clients. + +The package deliberately has no Node.js, Tauri, React, or model-provider dependency. Durable +events describe thread, turn, and completed-item state; streaming deltas are transient and are +excluded from record/replay snapshots. + +This is an internal experimental boundary. Consumers must negotiate `protocolVersion` through +`initialize` instead of assuming backwards compatibility. diff --git a/packages/protocol/package.json b/packages/protocol/package.json new file mode 100644 index 0000000..1675b7d --- /dev/null +++ b/packages/protocol/package.json @@ -0,0 +1,27 @@ +{ + "name": "@deepcode/protocol", + "version": "0.0.0", + "private": true, + "description": "Experimental provider- and transport-neutral DeepCode lifecycle protocol", + "license": "MIT", + "type": "module", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + } + }, + "scripts": { + "build": "tsc -p tsconfig.json", + "typecheck": "tsc -b", + "test": "vitest run", + "lint": "echo 'lint: configured at workspace root' && exit 0", + "clean": "rm -rf dist *.tsbuildinfo" + }, + "devDependencies": { + "typescript": "^5.7.0", + "vitest": "^2.1.0" + } +} diff --git a/packages/protocol/src/codec.test.ts b/packages/protocol/src/codec.test.ts new file mode 100644 index 0000000..456504c --- /dev/null +++ b/packages/protocol/src/codec.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from 'vitest'; + +import { decodeProtocolRequest, encodeProtocolMessage } from './codec.js'; + +describe('protocol codec', () => { + it('has a stable line-oriented JSON representation', () => { + const request = { + id: 1, + method: 'initialize' as const, + params: { client: 'protocol-test' }, + }; + + expect(encodeProtocolMessage(request)).toBe( + '{"id":1,"method":"initialize","params":{"client":"protocol-test"}}', + ); + expect(decodeProtocolRequest(encodeProtocolMessage(request))).toEqual(request); + }); + + it.each(['{}', '{"id":1,"method":"unknown"}', '{"id":1,"method":"initialize","params":[]}'])( + 'rejects an invalid request: %s', + (raw) => { + expect(() => decodeProtocolRequest(raw)).toThrow('invalid protocol request'); + }, + ); +}); diff --git a/packages/protocol/src/codec.ts b/packages/protocol/src/codec.ts new file mode 100644 index 0000000..58c18d6 --- /dev/null +++ b/packages/protocol/src/codec.ts @@ -0,0 +1,30 @@ +import type { ProtocolMethod, ProtocolRequest, ProtocolResponse } from './types.js'; + +const protocolMethods = new Set([ + 'initialize', + 'thread/start', + 'thread/read', + 'thread/resume', + 'turn/start', + 'turn/interrupt', +]); + +export function encodeProtocolMessage(message: ProtocolRequest | ProtocolResponse): string { + return JSON.stringify(message); +} + +export function decodeProtocolRequest(raw: string): ProtocolRequest { + const value = JSON.parse(raw) as Partial; + const validParams = + value.params === undefined || + (typeof value.params === 'object' && value.params !== null && !Array.isArray(value.params)); + if ( + (typeof value.id !== 'string' && typeof value.id !== 'number') || + typeof value.method !== 'string' || + !protocolMethods.has(value.method as ProtocolMethod) || + !validParams + ) { + throw new Error('invalid protocol request'); + } + return { id: value.id, method: value.method, params: value.params ?? {} } as ProtocolRequest; +} diff --git a/packages/protocol/src/index.ts b/packages/protocol/src/index.ts new file mode 100644 index 0000000..9679568 --- /dev/null +++ b/packages/protocol/src/index.ts @@ -0,0 +1,3 @@ +export * from './types.js'; +export * from './runtime.js'; +export * from './codec.js'; diff --git a/packages/protocol/src/runtime.test.ts b/packages/protocol/src/runtime.test.ts new file mode 100644 index 0000000..969f1f1 --- /dev/null +++ b/packages/protocol/src/runtime.test.ts @@ -0,0 +1,136 @@ +import { describe, expect, it } from 'vitest'; + +import { + MemoryThreadStore, + ProtocolInvariantError, + ProtocolRecorder, + ProtocolRuntime, +} from './runtime.js'; +import type { ProtocolEvent } from './types.js'; + +function deterministicRuntime( + store: MemoryThreadStore, + events: ProtocolEvent[] = [], +): ProtocolRuntime { + let tick = 0; + let sequence = 0; + return new ProtocolRuntime({ + store, + now: () => `2026-08-01T00:00:0${tick++}.000Z`, + newId: (prefix) => `${prefix}-${++sequence}`, + onEvent: (event) => events.push(event), + }); +} + +describe('ProtocolRuntime', () => { + it('advertises the versioned lifecycle capabilities', () => { + const runtime = deterministicRuntime(new MemoryThreadStore()); + + expect(runtime.initialize()).toEqual({ + protocolVersion: 1, + capabilities: { + threadResume: true, + turnInterrupt: true, + completedItemPersistence: true, + transientDeltas: true, + }, + }); + }); + + it('persists completed items, keeps deltas transient, and resumes after restart', async () => { + const store = new MemoryThreadStore(); + const events: ProtocolEvent[] = []; + const runtime = deterministicRuntime(store, events); + const thread = await runtime.startThread('/workspace'); + const turn = await runtime.startTurn(thread.id, { text: 'inspect the repository' }); + const assistant = await runtime.appendCompletedItem(thread.id, turn.id, 'assistant_message', { + text: 'working', + }); + + const savesBeforeDelta = store.saveCount; + runtime.publishDelta({ + threadId: thread.id, + turnId: turn.id, + itemId: assistant.id, + delta: '...', + }); + expect(store.saveCount).toBe(savesBeforeDelta); + + const completed = await runtime.completeTurn(thread.id, turn.id); + expect(store.saveCount).toBe(4); + expect(completed.status).toBe('completed'); + + const restartedRuntime = deterministicRuntime(store); + await expect(restartedRuntime.resumeThread(thread.id)).resolves.toEqual({ + ...thread, + updatedAt: completed.completedAt, + turns: [completed], + }); + expect(events.map((event) => event.type)).toEqual([ + 'thread.started', + 'turn.started', + 'item.completed', + 'item.completed', + 'item.delta', + 'turn.completed', + ]); + }); + + it('allows only one active turn per thread', async () => { + const runtime = deterministicRuntime(new MemoryThreadStore()); + const thread = await runtime.startThread('/workspace'); + await runtime.startTurn(thread.id, { text: 'first' }); + + await expect(runtime.startTurn(thread.id, { text: 'second' })).rejects.toThrow( + new ProtocolInvariantError(`Thread ${thread.id} already has an active turn`), + ); + }); + + it('makes terminal transitions idempotent and rejects late items', async () => { + const store = new MemoryThreadStore(); + const events: ProtocolEvent[] = []; + const runtime = deterministicRuntime(store, events); + const thread = await runtime.startThread('/workspace'); + const turn = await runtime.startTurn(thread.id, { text: 'stop me' }); + + const interrupted = await runtime.interruptTurn(thread.id, turn.id); + const savesAfterInterrupt = store.saveCount; + await expect(runtime.interruptTurn(thread.id, turn.id)).resolves.toEqual(interrupted); + await expect(runtime.completeTurn(thread.id, turn.id)).resolves.toEqual(interrupted); + expect(store.saveCount).toBe(savesAfterInterrupt); + expect(events.filter((event) => event.type === 'turn.interrupted')).toHaveLength(1); + expect(events.filter((event) => event.type === 'turn.completed')).toHaveLength(0); + await expect( + runtime.appendCompletedItem(thread.id, turn.id, 'assistant_message', { text: 'late' }), + ).rejects.toThrow(`Cannot append to terminal turn ${turn.id}`); + }); + + it('records and replays only durable events', async () => { + const recorder = new ProtocolRecorder(); + const runtime = new ProtocolRuntime({ + store: new MemoryThreadStore(), + now: () => '2026-08-01T00:00:00.000Z', + newId: (prefix) => `${prefix}-1`, + onEvent: (event) => recorder.record(event), + }); + const thread = await runtime.startThread('/workspace'); + const turn = await runtime.startTurn(thread.id, { text: 'hello' }); + runtime.publishDelta({ + threadId: thread.id, + turnId: turn.id, + itemId: 'item-streaming', + delta: 'hel', + }); + await runtime.completeTurn(thread.id, turn.id); + + const replayed: ProtocolEvent[] = []; + recorder.replay((event) => replayed.push(event)); + expect(replayed).toEqual(recorder.snapshot()); + expect(replayed.map((event) => event.type)).toEqual([ + 'thread.started', + 'turn.started', + 'item.completed', + 'turn.completed', + ]); + }); +}); diff --git a/packages/protocol/src/runtime.ts b/packages/protocol/src/runtime.ts new file mode 100644 index 0000000..326a75e --- /dev/null +++ b/packages/protocol/src/runtime.ts @@ -0,0 +1,217 @@ +import { + PROTOCOL_VERSION, + type CompletedItem, + type CompletedItemType, + type DurableProtocolEvent, + type InitializeResult, + type ProtocolEvent, + type ThreadSnapshot, + type TransientDeltaEvent, + type TurnSnapshot, + type TurnStatus, +} from './types.js'; + +function clone(value: T): T { + return JSON.parse(JSON.stringify(value)) as T; +} + +export interface ThreadStore { + load(threadId: string): Promise; + save(thread: ThreadSnapshot): Promise; +} + +export class MemoryThreadStore implements ThreadStore { + private readonly threads = new Map(); + saveCount = 0; + + async load(threadId: string): Promise { + const thread = this.threads.get(threadId); + return thread ? clone(thread) : null; + } + + async save(thread: ThreadSnapshot): Promise { + this.saveCount++; + this.threads.set(thread.id, clone(thread)); + } +} + +export interface ProtocolRuntimeOptions { + store: ThreadStore; + now?: () => string; + newId?: (prefix: 'thread' | 'turn' | 'item') => string; + onEvent?: (event: ProtocolEvent) => void; +} + +export class ProtocolInvariantError extends Error { + constructor(message: string) { + super(message); + this.name = 'ProtocolInvariantError'; + } +} + +export class ProtocolRuntime { + private readonly now: () => string; + private readonly newId: (prefix: 'thread' | 'turn' | 'item') => string; + + constructor(private readonly options: ProtocolRuntimeOptions) { + this.now = options.now ?? (() => new Date().toISOString()); + this.newId = + options.newId ?? + ((prefix) => + `${prefix}-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`); + } + + initialize(): InitializeResult { + return { + protocolVersion: PROTOCOL_VERSION, + capabilities: { + threadResume: true, + turnInterrupt: true, + completedItemPersistence: true, + transientDeltas: true, + }, + }; + } + + async startThread(cwd: string): Promise { + const now = this.now(); + const thread: ThreadSnapshot = { + id: this.newId('thread'), + cwd, + createdAt: now, + updatedAt: now, + turns: [], + }; + await this.options.store.save(thread); + this.emit({ type: 'thread.started', thread: clone(thread) }); + return clone(thread); + } + + readThread(threadId: string): Promise { + return this.options.store.load(threadId); + } + + async resumeThread(threadId: string): Promise { + return this.requireThread(threadId); + } + + async startTurn(threadId: string, input: Record): Promise { + const thread = await this.requireThread(threadId); + if (thread.turns.some((turn) => turn.status === 'in_progress')) { + throw new ProtocolInvariantError(`Thread ${threadId} already has an active turn`); + } + const now = this.now(); + const inputItem = this.completedItem('user_message', input, now); + const turn: TurnSnapshot = { + id: this.newId('turn'), + threadId, + status: 'in_progress', + startedAt: now, + items: [inputItem], + }; + thread.turns.push(turn); + thread.updatedAt = now; + await this.options.store.save(thread); + this.emit({ type: 'turn.started', threadId, turn: clone(turn) }); + this.emit({ type: 'item.completed', threadId, turnId: turn.id, item: clone(inputItem) }); + return clone(turn); + } + + async appendCompletedItem( + threadId: string, + turnId: string, + type: CompletedItemType, + payload: Record, + ): Promise { + const thread = await this.requireThread(threadId); + const turn = this.requireTurn(thread, turnId); + if (turn.status !== 'in_progress') { + throw new ProtocolInvariantError(`Cannot append to terminal turn ${turnId}`); + } + const item = this.completedItem(type, payload, this.now()); + turn.items.push(item); + thread.updatedAt = item.completedAt; + await this.options.store.save(thread); + this.emit({ type: 'item.completed', threadId, turnId, item: clone(item) }); + return clone(item); + } + + publishDelta(event: Omit): void { + this.emit({ type: 'item.delta', ...event }); + } + + completeTurn(threadId: string, turnId: string): Promise { + return this.finishTurn(threadId, turnId, 'completed'); + } + + interruptTurn(threadId: string, turnId: string): Promise { + return this.finishTurn(threadId, turnId, 'interrupted'); + } + + failTurn(threadId: string, turnId: string): Promise { + return this.finishTurn(threadId, turnId, 'failed'); + } + + private async finishTurn( + threadId: string, + turnId: string, + requested: Exclude, + ): Promise { + const thread = await this.requireThread(threadId); + const turn = this.requireTurn(thread, turnId); + if (turn.status !== 'in_progress') return clone(turn); + const now = this.now(); + turn.status = requested; + turn.completedAt = now; + thread.updatedAt = now; + await this.options.store.save(thread); + const type = + requested === 'completed' + ? 'turn.completed' + : requested === 'interrupted' + ? 'turn.interrupted' + : 'turn.failed'; + this.emit({ type, threadId, turn: clone(turn) } as DurableProtocolEvent); + return clone(turn); + } + + private async requireThread(threadId: string): Promise { + const thread = await this.options.store.load(threadId); + if (!thread) throw new ProtocolInvariantError(`Thread not found: ${threadId}`); + return thread; + } + + private requireTurn(thread: ThreadSnapshot, turnId: string): TurnSnapshot { + const turn = thread.turns.find((candidate) => candidate.id === turnId); + if (!turn) throw new ProtocolInvariantError(`Turn not found: ${turnId}`); + return turn; + } + + private completedItem( + type: CompletedItemType, + payload: Record, + completedAt: string, + ): CompletedItem { + return { id: this.newId('item'), type, payload: clone(payload), completedAt }; + } + + private emit(event: ProtocolEvent): void { + this.options.onEvent?.(event); + } +} + +export class ProtocolRecorder { + private readonly records: DurableProtocolEvent[] = []; + + record(event: ProtocolEvent): void { + if (event.type !== 'item.delta') this.records.push(clone(event)); + } + + replay(consumer: (event: DurableProtocolEvent) => void): void { + for (const event of this.records) consumer(clone(event)); + } + + snapshot(): DurableProtocolEvent[] { + return clone(this.records); + } +} diff --git a/packages/protocol/src/types.ts b/packages/protocol/src/types.ts new file mode 100644 index 0000000..9c1c99a --- /dev/null +++ b/packages/protocol/src/types.ts @@ -0,0 +1,83 @@ +export const PROTOCOL_VERSION = 1 as const; + +export type TurnStatus = 'in_progress' | 'completed' | 'interrupted' | 'failed'; +export type CompletedItemType = + | 'user_message' + | 'assistant_message' + | 'tool_call' + | 'tool_result' + | 'approval' + | 'ask_user' + | 'error'; + +export interface CompletedItem { + id: string; + type: CompletedItemType; + payload: Record; + completedAt: string; +} + +export interface TurnSnapshot { + id: string; + threadId: string; + status: TurnStatus; + startedAt: string; + completedAt?: string; + items: CompletedItem[]; +} + +export interface ThreadSnapshot { + id: string; + cwd: string; + createdAt: string; + updatedAt: string; + turns: TurnSnapshot[]; +} + +export type DurableProtocolEvent = + | { type: 'thread.started'; thread: ThreadSnapshot } + | { type: 'turn.started'; threadId: string; turn: TurnSnapshot } + | { type: 'item.completed'; threadId: string; turnId: string; item: CompletedItem } + | { type: 'turn.completed'; threadId: string; turn: TurnSnapshot } + | { type: 'turn.interrupted'; threadId: string; turn: TurnSnapshot } + | { type: 'turn.failed'; threadId: string; turn: TurnSnapshot }; + +export interface TransientDeltaEvent { + type: 'item.delta'; + threadId: string; + turnId: string; + itemId: string; + delta: string; +} + +export type ProtocolEvent = DurableProtocolEvent | TransientDeltaEvent; + +export interface InitializeResult { + protocolVersion: typeof PROTOCOL_VERSION; + capabilities: { + threadResume: true; + turnInterrupt: true; + completedItemPersistence: true; + transientDeltas: true; + }; +} + +export type ProtocolMethod = + | 'initialize' + | 'thread/start' + | 'thread/read' + | 'thread/resume' + | 'turn/start' + | 'turn/interrupt'; + +export interface ProtocolRequest { + id: string | number; + method: ProtocolMethod; + params: Record; +} + +export interface ProtocolResponse { + id: string | number; + result?: unknown; + error?: { code: string; message: string }; +} diff --git a/packages/protocol/tsconfig.json b/packages/protocol/tsconfig.json new file mode 100644 index 0000000..571f2cf --- /dev/null +++ b/packages/protocol/tsconfig.json @@ -0,0 +1,11 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "./dist", + "rootDir": "./src", + "composite": true, + "tsBuildInfoFile": "./dist/.tsbuildinfo" + }, + "include": ["src/**/*"], + "exclude": ["node_modules", "dist", "**/*.test.ts"] +} diff --git a/packages/protocol/vitest.config.ts b/packages/protocol/vitest.config.ts new file mode 100644 index 0000000..7067968 --- /dev/null +++ b/packages/protocol/vitest.config.ts @@ -0,0 +1,5 @@ +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + test: { include: ['src/**/*.test.ts'], environment: 'node' }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b25967a..7d23aa1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -167,6 +167,15 @@ importers: specifier: ^2.1.0 version: 2.1.9(@types/node@22.19.19) + packages/protocol: + devDependencies: + typescript: + specifier: ^5.7.0 + version: 5.9.3 + vitest: + specifier: ^2.1.0 + version: 2.1.9(@types/node@22.19.19) + packages/shared-ui: devDependencies: typescript: diff --git a/scripts/spike-desktop-sidecar.mjs b/scripts/spike-desktop-sidecar.mjs new file mode 100644 index 0000000..f53a586 --- /dev/null +++ b/scripts/spike-desktop-sidecar.mjs @@ -0,0 +1,126 @@ +#!/usr/bin/env node + +import { copyFile, mkdtemp, rename, rm, stat, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { spawn, spawnSync } from 'node:child_process'; +import { performance } from 'node:perf_hooks'; +import process from 'node:process'; +import { createInterface } from 'node:readline'; + +const probeSource = String.raw` +const readline = require('node:readline'); +const lines = readline.createInterface({ input: process.stdin, crlfDelay: Infinity }); +lines.on('line', (line) => { + const request = JSON.parse(line); + const result = request.method === 'initialize' + ? { + protocolVersion: 1, + capabilities: { + threadResume: true, + turnInterrupt: true, + completedItemPersistence: true, + transientDeltas: true, + }, + runtime: { execPath: process.execPath, path: process.env.PATH ?? null }, + } + : undefined; + process.stdout.write(JSON.stringify({ id: request.id, result }) + '\n'); + lines.close(); +}); +`; + +async function runProbe(runtimePath, serverPath) { + const startedAt = performance.now(); + const child = spawn(runtimePath, [serverPath], { + env: { PATH: '' }, + stdio: ['pipe', 'pipe', 'inherit'], + }); + const output = createInterface({ input: child.stdout, crlfDelay: Infinity }); + const exited = new Promise((resolve, reject) => { + child.once('error', reject); + child.once('exit', (code, signal) => resolve({ code, signal })); + }); + const response = new Promise((resolve, reject) => { + output.once('line', (line) => { + try { + resolve(JSON.parse(line)); + } catch (error) { + reject(error); + } + }); + }); + child.stdin.end('{"id":1,"method":"initialize","params":{}}\n'); + const result = await Promise.race([ + response, + exited.then(({ code, signal }) => { + throw new Error(`sidecar probe exited before responding (code=${code}, signal=${signal})`); + }), + ]); + const { code, signal } = await exited; + if (code !== 0) throw new Error(`sidecar probe exited with code=${code}, signal=${signal}`); + return { result, handshakeMilliseconds: performance.now() - startedAt }; +} + +const temporaryRoot = await mkdtemp(join(tmpdir(), 'deepcode-sidecar-')); +const runtimePath = join(temporaryRoot, 'deepcode-runtime'); +const serverPath = join(temporaryRoot, 'app-server.cjs'); + +try { + await copyFile(process.execPath, runtimePath); + await writeFile(serverPath, probeSource); + const sourceRuntimeBytes = (await stat(runtimePath)).size; + let thinned = false; + if (process.platform === 'darwin' && ['arm64', 'x64'].includes(process.arch)) { + const thinPath = `${runtimePath}.thin`; + const architecture = process.arch === 'x64' ? 'x86_64' : process.arch; + const thin = spawnSync('/usr/bin/lipo', [ + runtimePath, + '-thin', + architecture, + '-output', + thinPath, + ]); + if (thin.status === 0) { + await rename(thinPath, runtimePath); + thinned = true; + } + } + const targetRuntimeBytes = (await stat(runtimePath)).size; + const strip = + process.platform === 'darwin' ? spawnSync('/usr/bin/strip', ['-S', runtimePath]) : null; + const stripped = strip?.status === 0; + const sign = + process.platform === 'darwin' + ? spawnSync('/usr/bin/codesign', ['--force', '--sign', '-', runtimePath]) + : null; + const signed = sign?.status === 0; + const afterStrip = (await stat(runtimePath)).size; + const { result: response, handshakeMilliseconds } = await runProbe(runtimePath, serverPath); + if (response?.result?.protocolVersion !== 1 || response?.result?.runtime?.path !== '') { + throw new Error('sidecar did not complete an isolated protocol handshake'); + } + + process.stdout.write( + `${JSON.stringify( + { + status: 'ok', + platform: `${process.platform}-${process.arch}`, + sourceRuntimeBytes, + targetRuntimeBytes, + runtimeBytesAfterStrip: afterStrip, + thinned, + stripped, + signed, + protocolVersion: response.result.protocolVersion, + handshakeMilliseconds: Math.round(handshakeMilliseconds * 100) / 100, + childExecPath: response.result.runtime.execPath, + childPath: response.result.runtime.path, + }, + null, + 2, + )}\n`, + ); +} finally { + await rm(temporaryRoot, { recursive: true, force: true }); +} diff --git a/tsconfig.json b/tsconfig.json index b2e6051..e448224 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -3,6 +3,7 @@ "include": [], "references": [ { "path": "./packages/core" }, + { "path": "./packages/protocol" }, { "path": "./packages/shared-ui" }, { "path": "./apps/cli" }, { "path": "./apps/desktop" },