diff --git a/src/commands/up.rs b/src/commands/up.rs index c753d5d4a..1807f9091 100644 --- a/src/commands/up.rs +++ b/src/commands/up.rs @@ -2113,7 +2113,7 @@ async fn delete_project_inner(state: AppState, id: String, query: DeleteProjectQ // Row deletion can wait up to RUN_ARTIFACTS_WAIT for the run-artifacts lock, so keep it off the async workers. tokio::task::spawn_blocking(move || { let native_purge = (!session_ids.is_empty()).then_some(|| { - local::chat::ChatHost::delete_native_sessions_blocking(&store, &session_ids) + local::chat::ChatHost::purge_native_history_blocking(&store, &session_ids) }); store.delete_local_project_with(&project_id, native_purge) }) @@ -4656,7 +4656,7 @@ async fn reclaim_orphaned_run_artifacts( ) { // Demo prewarm skips itself when the gate is contended at startup. tokio::time::sleep(Duration::from_secs(5)).await; - // Orphans whose failure is already logged, so a stuck one logs once. + // Orphans and sweep failures already logged, so a stuck one logs once. let mut reported = HashSet::new(); loop { let mut more = false; @@ -4683,16 +4683,26 @@ async fn reclaim_orphaned_run_artifacts( eprintln!("orx up: orphaned run artifact cleanup failed: {error}") } } - match tokio::task::spawn_blocking(|| { - Store::open()?.cleanup_orphaned_chat_attachments() - }) - .await - { - Ok(Ok(left)) => more |= left, - Ok(Err(error)) => { - eprintln!("orx up: could not clean orphaned chat attachments: {error}") + let sweeps = [ + ( + "chat attachments", + Store::cleanup_orphaned_chat_attachments as fn(&Store) -> Result, + ), + ( + "native chat history", + Store::cleanup_orphaned_native_sessions, + ), + ]; + for (label, sweep) in sweeps { + match tokio::task::spawn_blocking(move || sweep(&Store::open()?)).await { + Ok(Ok(left)) => more |= left, + Ok(Err(error)) => { + if reported.insert(format!("{label}: {error}")) { + eprintln!("orx up: could not clean orphaned {label}: {error}"); + } + } + Err(error) => eprintln!("orx up: orphaned {label} cleanup failed: {error}"), } - Err(error) => eprintln!("orx up: chat attachment cleanup failed: {error}"), } } } diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index a6a5bcf79..97c7cd904 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -6733,7 +6733,7 @@ impl ChatHost { let native_deletion = store.lock_native_deletion()?; let session = store.get_chat_session(session_id)?; let worktree_owner = store.chat_worktree_owner(session_id)?; - Self::delete_native_sessions(&[session_id.to_string()]).await?; + Self::purge_native_history(&[session_id.to_string()]).await?; store.delete_chat_session(session_id)?; drop(native_deletion); self.claude.forget_session(session_id).await; @@ -6750,17 +6750,25 @@ impl ChatHost { Ok(()) } - pub async fn delete_native_sessions(session_ids: &[String]) -> Result<()> { + pub async fn purge_native_history(session_ids: &[String]) -> Result<()> { let session_ids = session_ids.to_vec(); tokio::task::spawn_blocking(move || { - Self::delete_native_sessions_blocking(&Store::open()?, &session_ids) + Self::purge_native_history_blocking(&Store::open()?, &session_ids) }) .await? } - pub fn delete_native_sessions_blocking(store: &Store, session_ids: &[String]) -> Result<()> { - let (ids, retained) = store.opencode_sessions_for_deletion(session_ids)?; - crate::local::native_store::delete_opencode_sessions(&ids, &retained) + pub fn purge_native_history_blocking(store: &Store, session_ids: &[String]) -> Result<()> { + let (ids, retained) = store.native_sessions_for_deletion(session_ids, &["opencode"])?; + crate::local::native_store::delete_opencode_sessions(&ids, &retained)?; + // ACP agents keep history in homes orx does not own, so it is left alone. + // Survivors span every harness: a Codex sub-agent may be recorded by any other chat. + let (ids, _) = + store.native_sessions_for_deletion(session_ids, &["claude-code", "codex", "cursor"])?; + crate::local::native_store::delete_isolated_sessions( + &ids, + &store.native_ids_outside(session_ids)?, + ) } } diff --git a/src/local/native_store.rs b/src/local/native_store.rs index db49903d4..2bd0dd8fe 100644 --- a/src/local/native_store.rs +++ b/src/local/native_store.rs @@ -196,6 +196,360 @@ pub fn delete_opencode_sessions( opencode_database::delete_sessions(&[isolated, legacy], ids, retained) } +/// Removes Claude Code, Codex and Cursor history from orx's isolated homes only, never the user's own. +pub fn delete_isolated_sessions( + ids: &std::collections::BTreeSet, + surviving: &std::collections::HashSet, +) -> Result<()> { + if ids.is_empty() { + return Ok(()); + } + remove_isolated_sessions(&crate::store::data_dir().join("agents"), ids, surviving) +} + +/// `surviving` holds every id other chats still record; their Codex sub-agents stay too. +fn remove_isolated_sessions( + agents: &Path, + ids: &std::collections::BTreeSet, + surviving: &std::collections::HashSet, +) -> Result<()> { + let mut deleting = ids + .iter() + .filter(|id| !surviving.contains(*id)) + .cloned() + .collect::>(); + let codex = agents.join("codex"); + // Unreadable spawn edges leave sub-agents to the sweep, which skips Codex until they can be read. + if let (Ok(spawned), Ok(kept)) = ( + codex_spawned_threads(&codex, &deleting), + codex_spawned_threads(&codex, surviving), + ) { + deleting = spawned + .into_iter() + .filter(|id| !kept.contains(id)) + .collect(); + } + let files = isolated_session_files(agents)?; + for id in &deleting { + for path in files.get(id).into_iter().flatten() { + remove_session_path(path)?; + } + } + Ok(()) +} + +#[derive(Clone, Copy)] +enum IsolatedHome { + Claude, + Codex, + Cursor, +} + +const ISOLATED_HOMES: [IsolatedHome; 3] = [ + IsolatedHome::Claude, + IsolatedHome::Codex, + IsolatedHome::Cursor, +]; + +#[derive(Clone, Copy)] +enum SweepTarget { + Home(IsolatedHome), + OpenCode, +} + +const SWEEP_ORDER: [SweepTarget; 4] = [ + SweepTarget::Home(IsolatedHome::Claude), + SweepTarget::Home(IsolatedHome::Codex), + SweepTarget::Home(IsolatedHome::Cursor), + SweepTarget::OpenCode, +]; + +/// Rotates which target a sweep pass starts with, so one slow target cannot starve the others. +static NEXT_SWEEP_TARGET: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); + +/// Remove orx-launched sessions no chat records once they have settled; returns whether the budget ran out. +pub(crate) fn sweep_isolated_sessions( + agents: &Path, + mut referenced: std::collections::HashSet, + settle: std::time::Duration, + budget: std::time::Duration, +) -> Result { + let deadline = std::time::Instant::now() + budget; + // Without the spawn edges a referenced thread's sub-agents would look orphaned. + let codex_readable = match codex_spawned_threads(&agents.join("codex"), &referenced) { + Ok(spawned) => { + referenced.extend(spawned); + true + } + Err(_) => false, + }; + let first = NEXT_SWEEP_TARGET.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let mut failure = None; + for step in 0..SWEEP_ORDER.len() { + // More work keeps the fast re-run; a failure is reported by a pass that finishes. + if std::time::Instant::now() >= deadline { + return Ok(true); + } + // Each target fails on its own, so an unhealthy one never holds the others back. + let swept = match SWEEP_ORDER[(first + step) % SWEEP_ORDER.len()] { + SweepTarget::Home(IsolatedHome::Codex) if !codex_readable => Ok(false), + SweepTarget::Home(home) => { + sweep_isolated_home(agents, home, &referenced, settle, deadline) + } + SweepTarget::OpenCode => sweep_opencode(agents, &referenced, settle).map(|()| false), + }; + match swept { + Ok(true) => return Ok(true), + Ok(false) => {} + Err(error) => { + failure.get_or_insert(error); + } + } + } + failure.map_or(Ok(false), Err) +} + +fn sweep_isolated_home( + agents: &Path, + home: IsolatedHome, + referenced: &std::collections::HashSet, + settle: std::time::Duration, + deadline: std::time::Instant, +) -> Result { + for (id, paths) in home_session_files(agents, home)? { + if referenced.contains(&id) + || !paths.iter().all(|path| { + newest_modified(path).is_some_and(|at| at.elapsed().unwrap_or_default() >= settle) + }) + { + continue; + } + for path in &paths { + // A path that cannot be removed is retried next pass without stalling the rest. + let _ = remove_session_path(path); + } + // Checked after a removal, so a home whose walk outlasts the budget still progresses. + if std::time::Instant::now() >= deadline { + return Ok(true); + } + } + Ok(false) +} + +fn sweep_opencode( + agents: &Path, + referenced: &std::collections::HashSet, + settle: std::time::Duration, +) -> Result<()> { + let database = agents.join("opencode/opencode.db"); + if same_home(&database, &opencode_db(NativeStore::Legacy)) { + return Ok(()); + } + let orphans = opencode_database::stale_root_sessions(&database, settle)? + .into_iter() + .filter(|id| !referenced.contains(id)) + .collect::>(); + if !orphans.is_empty() { + let retained = referenced.iter().cloned().collect(); + let purged = opencode_database::delete_sessions( + std::slice::from_ref(&database), + &orphans, + &retained, + ); + // A busy database is retried next pass. + if let Err(error) = purged { + if error + .downcast_ref::() + .is_none() + { + return Err(error); + } + } + } + opencode_database::compact(&database) +} + +/// A directory's own mtime misses writes to files already inside it, so the newest one counts. +fn newest_modified(path: &Path) -> Option { + let metadata = std::fs::symlink_metadata(path).ok()?; + let mut newest = metadata.modified().ok()?; + if metadata.is_dir() { + for child in child_paths(path).ok()? { + newest = newest.max(newest_modified(&child)?); + } + } + Some(newest) +} + +/// A user home pointed at orx's isolated one is the user's, so it is never purged. +fn same_home(isolated: &Path, legacy: &Path) -> bool { + isolated == legacy + || matches!( + (crate::paths::canonicalize(isolated), crate::paths::canonicalize(legacy)), + (Ok(isolated), Ok(legacy)) if isolated == legacy + ) +} + +/// Session files in orx's isolated Claude Code, Codex and Cursor homes, by native id. +fn isolated_session_files( + agents: &Path, +) -> Result>> { + let mut sessions = std::collections::BTreeMap::>::new(); + for home in ISOLATED_HOMES { + for (id, paths) in home_session_files(agents, home)? { + sessions.entry(id).or_default().extend(paths); + } + } + Ok(sessions) +} + +fn home_session_files( + agents: &Path, + home: IsolatedHome, +) -> Result>> { + let mut sessions = std::collections::BTreeMap::>::new(); + let mut add = |id: Option<&str>, path: PathBuf| { + // Hyphenated only: Claude, Codex and Cursor ids, and nothing else in those trees. + if let Some(id) = id.filter(|id| id.len() == 36 && uuid::Uuid::parse_str(id).is_ok()) { + sessions.entry(id.to_string()).or_default().push(path); + } + }; + let name = |path: &Path| { + path.file_name() + .and_then(|name| name.to_str()) + .map(str::to_string) + }; + match home { + IsolatedHome::Claude => { + let claude = agents.join("claude"); + if same_home(&claude, &claude_home(NativeStore::Legacy)) { + return Ok(sessions); + } + for project in child_dirs(&claude.join("projects"))? { + // `.jsonl` is the transcript; a sibling `/` holds its subagents and tool results. + for path in child_paths(&project)? { + let name = name(&path); + add( + name.as_deref() + .map(|name| name.strip_suffix(".jsonl").unwrap_or(name)), + path, + ); + } + } + for tree in ["session-env", "file-history"] { + for path in child_paths(&claude.join(tree))? { + add(name(&path).as_deref(), path); + } + } + } + IsolatedHome::Codex => { + let codex = agents.join("codex"); + if same_home(&codex, &codex_home(NativeStore::Legacy)) { + return Ok(sessions); + } + for tree in ["sessions", "archived_sessions"] { + for path in tree_files(&codex.join(tree))? { + // Rollouts are `rollout--.jsonl`. + let stem = name(&path) + .and_then(|name| name.strip_suffix(".jsonl").map(str::to_string)); + add( + stem.as_deref() + .and_then(|stem| stem.get(stem.len().saturating_sub(36)..)), + path, + ); + } + } + } + IsolatedHome::Cursor => { + let cursor = agents.join("cursor"); + if same_home(&cursor, &cursor_home(NativeStore::Legacy)) { + return Ok(sessions); + } + for workspace in child_dirs(&cursor.join("chats"))? { + for path in child_paths(&workspace)? { + add(name(&path).as_deref(), path); + } + } + for path in child_paths(&cursor.join("acp-sessions"))? { + add(name(&path).as_deref(), path); + } + } + } + Ok(sessions) +} + +/// The given threads plus every sub-agent Codex spawned from them, transitively. +fn codex_spawned_threads( + codex: &Path, + parents: &std::collections::HashSet, +) -> Result> { + let database = codex.join("state_5.sqlite"); + if !database.exists() { + return Ok(parents.clone()); + } + let connection = rusqlite::Connection::open_with_flags( + database, + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, + )?; + let mut query = connection.prepare( + "WITH RECURSIVE spawned(id) AS ( + SELECT value FROM json_each(?1) + UNION SELECT child_thread_id FROM thread_spawn_edges JOIN spawned ON parent_thread_id = spawned.id + ) SELECT id FROM spawned", + )?; + let spawned = query + .query_map([serde_json::to_string(parents)?], |row| row.get(0))? + .collect::>()?; + Ok(spawned) +} + +/// Entries that vanish while listing are skipped: a harness may be moving them. +fn child_paths(root: &Path) -> Result> { + match std::fs::read_dir(root) { + Ok(entries) => Ok(entries.flatten().map(|entry| entry.path()).collect()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()), + Err(error) => Err(error.into()), + } +} + +fn child_dirs(root: &Path) -> Result> { + Ok(child_paths(root)? + .into_iter() + .filter(|path| std::fs::symlink_metadata(path).is_ok_and(|metadata| metadata.is_dir())) + .collect()) +} + +fn tree_files(root: &Path) -> Result> { + let mut files = Vec::new(); + for path in child_paths(root)? { + let Ok(metadata) = std::fs::symlink_metadata(&path) else { + continue; + }; + if metadata.is_dir() { + files.extend(tree_files(&path)?); + } else if metadata.is_file() { + files.push(path); + } + } + Ok(files) +} + +/// Symlinks are removed, never followed. +fn remove_session_path(path: &Path) -> Result<()> { + let result = match std::fs::symlink_metadata(path) { + Ok(metadata) if metadata.is_dir() => std::fs::remove_dir_all(path), + Ok(_) => std::fs::remove_file(path), + Err(error) => Err(error), + }; + match result { + Err(error) if error.kind() != std::io::ErrorKind::NotFound => Err(anyhow!( + "Could not remove native chat history {}: {error}", + path.display() + )), + _ => Ok(()), + } +} + pub fn codex_sqlite_override(store: NativeStore, home: &Path) -> Option { (store == NativeStore::Isolated).then(|| format!("sqlite_home={}", toml_string(home))) } @@ -751,4 +1105,161 @@ mod tests { assert!(!opencode_has_session(&db, "missing").unwrap()); std::fs::remove_dir_all(root).ok(); } + + fn write_aged(path: &Path, age: std::time::Duration) { + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, "private transcript").unwrap(); + std::fs::File::options() + .write(true) + .open(path) + .unwrap() + .set_modified(std::time::SystemTime::now() - age) + .unwrap(); + } + + #[test] + fn isolated_history_is_purged_by_id_and_swept_only_when_unreferenced_and_settled() { + let root = std::env::temp_dir().join(format!("orx-native-sweep-{}", uuid::Uuid::new_v4())); + let agents = root.join("agents"); + let day = std::time::Duration::from_secs(24 * 60 * 60); + let id = |n: u8| format!("00000000-0000-4000-8000-{n:012}"); + let project = agents.join("claude/projects/-tmp-chat"); + let rollout = |dir: &str, n: u8| { + agents.join(format!( + "codex/{dir}/rollout-2026-09-29T16-57-41-{}.jsonl", + id(n) + )) + }; + let deleted = [ + project.join(format!("{}.jsonl", id(1))), + project.join(id(1)).join("subagents/agent-a.jsonl"), + agents.join("claude/session-env").join(id(1)).join("env"), + agents + .join("cursor/chats/abc123") + .join(id(1)) + .join("store.db"), + ]; + let orphan = project.join(format!("{}.jsonl", id(2))); + let fresh = project.join(format!("{}.jsonl", id(3))); + let referenced_claude = project.join(format!("{}.jsonl", id(4))); + let orphan_rollout = rollout("sessions/2026/09/29", 5); + let referenced_rollout = rollout("archived_sessions", 6); + let spawned_rollout = rollout("sessions/2026/09/29", 7); + let not_a_session = project.join("notes.jsonl"); + for path in deleted.iter().chain([ + &orphan, + &referenced_claude, + &orphan_rollout, + &referenced_rollout, + &spawned_rollout, + ¬_a_session, + ]) { + write_aged(path, 2 * day); + } + write_aged(&fresh, std::time::Duration::ZERO); + let active_chat = agents.join("cursor/chats/abc123").join(id(8)); + write_aged(&active_chat.join("store.db"), std::time::Duration::ZERO); + std::fs::File::open(&active_chat) + .unwrap() + .set_modified(std::time::SystemTime::now() - 2 * day) + .unwrap(); + let (parent, child, kept_child) = ( + rollout("sessions/2026/09/29", 9), + rollout("sessions/2026/09/29", 10), + rollout("sessions/2026/09/29", 11), + ); + for path in [&parent, &child, &kept_child] { + write_aged(path, 2 * day); + } + let state = rusqlite::Connection::open(agents.join("codex/state_5.sqlite")).unwrap(); + state + .execute_batch("CREATE TABLE thread_spawn_edges (parent_thread_id TEXT, child_thread_id TEXT, status TEXT);") + .unwrap(); + for (parent, child) in [(6, 7), (9, 10), (9, 11)] { + state + .execute( + "INSERT INTO thread_spawn_edges VALUES (?1, ?2, 'done')", + [id(parent), id(child)], + ) + .unwrap(); + } + drop(state); + + let none = std::collections::HashSet::new(); + remove_isolated_sessions(&agents, &[id(1)].into(), &none).unwrap(); + assert!(deleted.iter().all(|path| !path.exists())); + assert!(!project.join(id(1)).exists()); + assert!(orphan.exists()); + remove_isolated_sessions(&agents, &[id(9)].into(), &[id(11)].into()).unwrap(); + assert!( + !parent.exists() && !child.exists(), + "sub-agents go with their parent" + ); + assert!(kept_child.exists(), "a surviving chat still records it"); + + let referenced = || [id(4), id(6)].into_iter().collect(); + assert!( + sweep_isolated_sessions(&agents, referenced(), day, std::time::Duration::ZERO).unwrap() + ); + assert!(orphan.exists(), "an exhausted budget removes nothing"); + assert!(!sweep_isolated_sessions(&agents, referenced(), day, day).unwrap()); + assert!(!orphan.exists() && !orphan_rollout.exists()); + for kept in [ + &fresh, + &referenced_claude, + &referenced_rollout, + &spawned_rollout, + ¬_a_session, + &active_chat, + ] { + assert!(kept.exists(), "{}", kept.display()); + } + std::fs::remove_dir_all(root).ok(); + } + + #[test] + fn opencode_sweep_purges_settled_unreferenced_roots_and_compacts() { + let root = + std::env::temp_dir().join(format!("orx-opencode-sweep-{}", uuid::Uuid::new_v4())); + let agents = root.join("agents"); + let db = agents.join("opencode/opencode.db"); + std::fs::create_dir_all(db.parent().unwrap()).unwrap(); + let day = std::time::Duration::from_secs(24 * 60 * 60); + let old = crate::store::now_ms() - 2 * 24 * 60 * 60 * 1000; + let connection = rusqlite::Connection::open(&db).unwrap(); + connection + .execute_batch(&format!( + "CREATE TABLE session (id TEXT PRIMARY KEY, directory TEXT, parent_id TEXT, time_updated INTEGER); + CREATE TABLE message (id TEXT, session_id TEXT, data TEXT); + CREATE TABLE part (id TEXT, message_id TEXT, session_id TEXT, data TEXT); + INSERT INTO session VALUES ('orphan', '/tmp', NULL, {old}), ('child', '/tmp', 'orphan', {old}), + ('kept', '/tmp', NULL, {old}), ('fresh', '/tmp', NULL, {now});", + now = crate::store::now_ms() + )) + .unwrap(); + for id in ["orphan", "child", "kept", "fresh"] { + connection + .execute( + "INSERT INTO message VALUES (?1, ?1, ?2)", + rusqlite::params![id, "x".repeat(64 * 1024)], + ) + .unwrap(); + } + drop(connection); + let before = std::fs::metadata(&db).unwrap().len(); + let referenced = ["kept".to_string()].into_iter().collect(); + sweep_isolated_sessions(&agents, referenced, day, day).unwrap(); + let connection = rusqlite::Connection::open(&db).unwrap(); + let mut query = connection + .prepare("SELECT id FROM session ORDER BY id") + .unwrap(); + let remaining = query + .query_map([], |row| row.get::<_, String>(0)) + .unwrap() + .collect::>>() + .unwrap(); + assert_eq!(remaining, ["fresh", "kept"]); + assert!(std::fs::metadata(&db).unwrap().len() < before); + std::fs::remove_dir_all(root).ok(); + } } diff --git a/src/local/opencode_db.rs b/src/local/opencode_db.rs index bff9a3103..1962f5642 100644 --- a/src/local/opencode_db.rs +++ b/src/local/opencode_db.rs @@ -215,7 +215,8 @@ fn delete_session_rows( ) -> Result<()> { let mut connection = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_WRITE)?; connection.busy_timeout(std::time::Duration::from_secs(5))?; - connection.execute_batch("PRAGMA foreign_keys = ON;")?; + // Zero deleted content in place so chat text does not linger in free pages. + connection.execute_batch("PRAGMA foreign_keys = ON; PRAGMA secure_delete = ON;")?; let transaction = connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let mut deleting = ids.difference(retained).cloned().collect::>(); @@ -267,6 +268,71 @@ fn delete_session_rows( } } transaction.commit()?; + // Older WAL frames still hold the deleted text; waiting out a live reader would stall + // OpenCode's writers, so the periodic `compact` of orx's own database retries it. + connection.busy_timeout(std::time::Duration::ZERO)?; + let _ = connection.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(())); + Ok(()) +} + +/// Top-level sessions untouched for `settle`; children go with their parent. +pub(super) fn stale_root_sessions( + path: &Path, + settle: std::time::Duration, +) -> Result> { + let mut ids = BTreeSet::new(); + if inspect(path)? == DatabaseState::Empty { + return Ok(ids); + } + let connection = open_readonly(path)?; + let cutoff = crate::store::now_ms() - i64::try_from(settle.as_millis())?; + for table in ["session", "session_v2"] { + if !has_columns(&connection, table, &["id", "parent_id", "time_updated"])? { + continue; + } + let mut query = connection.prepare(&format!( + "SELECT id FROM {table} WHERE parent_id IS NULL AND time_updated < ?1" + ))?; + for id in query.query_map([cutoff], |row| row.get::<_, String>(0))? { + ids.insert(id?); + } + } + Ok(ids) +} + +/// Truncates the WAL without waiting on readers, then VACUUMs once free pages are a tenth of the file and no OpenCode holds it. +pub(super) fn compact(path: &Path) -> Result<()> { + if inspect(path)? == DatabaseState::Empty { + return Ok(()); + } + let path = normalize_path(path)?; + let connection = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_WRITE)?; + connection.busy_timeout(std::time::Duration::ZERO)?; + let _ = connection.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(())); + let (free, pages): (i64, i64) = connection.query_row( + "SELECT freelist_count, page_count FROM pragma_freelist_count(), pragma_page_count()", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + )?; + if free == 0 || free * 10 < pages { + return Ok(()); + } + // orx's OpenCode children hold a shared lease, so an exclusive one means none is running. + let lock = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(sidecar(&path, ".orx-lock"))?; + // Windows' exclusive-open check would see this process's own connection as another user. + drop(connection); + if lock.try_lock().is_err() || ensure_no_external_users(&path).is_err() { + return Ok(()); + } + let connection = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_WRITE)?; + connection.busy_timeout(std::time::Duration::from_secs(5))?; + connection.execute_batch("VACUUM;")?; + let _ = connection.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(())); Ok(()) } @@ -812,6 +878,35 @@ mod tests { } } + #[test] + fn deleted_session_text_does_not_linger_in_the_file_or_wal() { + let fixture = Fixture::new(); + let path = fixture.path(); + let connection = Connection::open(&path).unwrap(); + connection + .execute_batch(&format!( + "PRAGMA journal_mode=WAL; {V1} + INSERT INTO message VALUES ('msg', 'history', 'deleted-secret-marker');" + )) + .unwrap(); + delete_sessions( + std::slice::from_ref(&path), + &BTreeSet::from(["history".into()]), + &BTreeSet::new(), + ) + .unwrap(); + drop(connection); + for suffix in ["", "-wal"] { + let bytes = std::fs::read(sidecar(&path, suffix)).unwrap_or_default(); + assert!( + !bytes + .windows(b"deleted-secret-marker".len()) + .any(|window| window == b"deleted-secret-marker"), + "{suffix}" + ); + } + } + #[test] fn native_deletion_rolls_back_on_failure_and_does_not_create_missing_databases() { let fixture = Fixture::new(); diff --git a/src/store.rs b/src/store.rs index 64d49a3e8..d900a5ca9 100644 --- a/src/store.rs +++ b/src/store.rs @@ -481,6 +481,8 @@ const ORPHAN_PASS_BUDGET: Duration = Duration::from_secs(5); const FULL_RUNS_PER_PROJECT: i64 = 3; /// Uploads are written before the message naming them, so the sweep spares recent files. const CHAT_ATTACHMENT_SETTLE: Duration = Duration::from_secs(24 * 60 * 60); +/// A harness writes its session before orx records the id, so the sweep spares recent ones. +const NATIVE_SESSION_SETTLE: Duration = Duration::from_secs(24 * 60 * 60); const SLIMMED_RUN_PATHS: [&[&str]; 2] = [&["repo", ".venv"], &["repo", ".cache", "uv"]]; const SUBMISSION_HANDLE: (&str, &str) = ("submission-handles", ".json"); @@ -836,6 +838,11 @@ impl Store { ] { let _ = conn.execute(ddl, []); } + // Covers the periodic native-id scan so it never reads the large parts_json. + conn.execute_batch( + "CREATE INDEX IF NOT EXISTS idx_chat_messages_native_ids + ON chat_messages(base_native_session_id, result_native_session_id, session_id);", + )?; conn.execute_batch("CREATE TRIGGER IF NOT EXISTS delete_chat_invocation_identities AFTER DELETE ON chat_sessions BEGIN DELETE FROM native_invocation_identities WHERE session_id = OLD.id; END;")?; conn.execute_batch("CREATE TRIGGER IF NOT EXISTS delete_acp_session AFTER DELETE ON chat_sessions BEGIN DELETE FROM acp_sessions WHERE session_id = OLD.id; END;")?; conn.execute( @@ -2582,9 +2589,10 @@ impl Store { Ok(lock) } - pub fn opencode_sessions_for_deletion( + pub fn native_sessions_for_deletion( &self, chat_ids: &[String], + harnesses: &[&str], ) -> Result<( std::collections::BTreeSet, std::collections::BTreeSet, @@ -2592,13 +2600,13 @@ impl Store { let mut query = self.conn.prepare( "WITH refs AS ( SELECT id AS chat_id, native_session_id AS native_id FROM chat_sessions - WHERE harness = 'opencode' + WHERE harness IN (SELECT value FROM json_each(?2)) UNION - SELECT s.id, m.base_native_session_id FROM chat_sessions s - JOIN chat_messages m ON m.session_id = s.id WHERE s.harness = 'opencode' + SELECT s.id, m.base_native_session_id FROM chat_sessions s JOIN chat_messages m + ON m.session_id = s.id WHERE s.harness IN (SELECT value FROM json_each(?2)) UNION - SELECT s.id, m.result_native_session_id FROM chat_sessions s - JOIN chat_messages m ON m.session_id = s.id WHERE s.harness = 'opencode' + SELECT s.id, m.result_native_session_id FROM chat_sessions s JOIN chat_messages m + ON m.session_id = s.id WHERE s.harness IN (SELECT value FROM json_each(?2)) ) SELECT native_id, MAX(chat_id IN (SELECT value FROM json_each(?1))), MAX(chat_id NOT IN (SELECT value FROM json_each(?1))) FROM refs @@ -2606,13 +2614,19 @@ impl Store { )?; let mut deleting = std::collections::BTreeSet::new(); let mut retained = std::collections::BTreeSet::new(); - for row in query.query_map([serde_json::to_string(chat_ids)?], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, bool>(1)?, - row.get::<_, bool>(2)?, - )) - })? { + for row in query.query_map( + [ + serde_json::to_string(chat_ids)?, + serde_json::to_string(harnesses)?, + ], + |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, bool>(1)?, + row.get::<_, bool>(2)?, + )) + }, + )? { let (native_id, belongs_to_chat, shared) = row?; if shared { retained.insert(native_id); @@ -2751,6 +2765,42 @@ impl Store { Ok(false) } + /// Reclaim orx-launched agent history no chat records; returns whether the budget ran out first. + pub(crate) fn cleanup_orphaned_native_sessions(&self) -> Result { + let Some(_move_hold) = self.hold_off_data_dir_move()? else { + return Ok(false); + }; + crate::local::native_store::sweep_isolated_sessions( + &self.data_dir.join("agents"), + self.native_ids_outside(&[])?, + NATIVE_SESSION_SETTLE, + ORPHAN_PASS_BUDGET, + ) + } + + /// Every native id recorded by chats other than `chat_ids`. + pub fn native_ids_outside( + &self, + chat_ids: &[String], + ) -> Result> { + let mut query = self.conn.prepare( + "SELECT native_session_id FROM chat_sessions + WHERE native_session_id IS NOT NULL AND id NOT IN (SELECT value FROM json_each(?1)) + UNION SELECT base_native_session_id FROM chat_messages + WHERE base_native_session_id IS NOT NULL + AND session_id NOT IN (SELECT value FROM json_each(?1)) + UNION SELECT result_native_session_id FROM chat_messages + WHERE result_native_session_id IS NOT NULL + AND session_id NOT IN (SELECT value FROM json_each(?1))", + )?; + let ids = query + .query_map([serde_json::to_string(chat_ids)?], |row| { + row.get::<_, String>(0) + })? + .collect::>()?; + Ok(ids) + } + /// `None` clears it, which a fork of the session's very first turn needs: /// that turn ran with no harness session to resume from. pub fn set_chat_session_native_id(&self, id: &str, native_id: Option<&str>) -> Result<()> { @@ -5775,9 +5825,13 @@ mod tests { fn native_deletion_includes_inactive_branches_and_preserves_shared_sessions() { let dir = crate::local::git::TemporaryDirectory::new("orx-native-delete-refs").unwrap(); let store = Store::open_at(dir.path().to_path_buf()).unwrap(); - for (id, native_id) in [("delete", "current"), ("keep", "shared")] { + for (id, harness, native_id) in [ + ("delete", "opencode", "current"), + ("keep", "claude-code", "shared"), + ("cursor", "cursor", "current"), + ] { let mut session = chat_session_fixture(id); - session.harness = "opencode".into(); + session.harness = harness.into(); session.native_session_id = Some(native_id.into()); store.create_chat_session(&session).unwrap(); } @@ -5791,12 +5845,15 @@ mod tests { ) .unwrap(); let (deleting, retained) = store - .opencode_sessions_for_deletion(&["delete".into()]) + .native_sessions_for_deletion(&["delete".into()], &["opencode", "claude-code"]) .unwrap(); assert_eq!(deleting, ["current".into(), "old".into()].into()); assert_eq!(retained, ["kept-branch".into(), "shared".into()].into()); let (deleting, retained) = store - .opencode_sessions_for_deletion(&["delete".into(), "keep".into()]) + .native_sessions_for_deletion( + &["delete".into(), "keep".into()], + &["opencode", "claude-code"], + ) .unwrap(); assert_eq!( deleting, @@ -5811,6 +5868,34 @@ mod tests { assert!(retained.is_empty()); } + #[test] + fn native_ids_outside_skips_only_the_given_chats() { + let dir = crate::local::git::TemporaryDirectory::new("orx-native-outside").unwrap(); + let store = Store::open_at(dir.path().to_path_buf()).unwrap(); + for (id, native_id) in [("gone", "own"), ("kept", "other")] { + let mut session = chat_session_fixture(id); + session.native_session_id = Some(native_id.into()); + store.create_chat_session(&session).unwrap(); + } + store + .conn + .execute_batch( + "INSERT INTO chat_messages (id, session_id, role, parts_json, created_at, + base_native_session_id, result_native_session_id) + VALUES ('a', 'gone', 'assistant', '[]', 1, 'own-base', 'own-result'), + ('b', 'kept', 'assistant', '[]', 1, 'other-base', 'other-result');", + ) + .unwrap(); + let outside = store.native_ids_outside(&["gone".into()]).unwrap(); + assert_eq!( + outside, + ["other", "other-base", "other-result"] + .map(String::from) + .into() + ); + assert_eq!(store.native_ids_outside(&[]).unwrap().len(), 6); + } + #[test] fn native_deletion_lock_serializes_shared_owners_across_processes() { const CHILD_DIR: &str = "ORX_TEST_NATIVE_DELETE_DIR"; @@ -5822,7 +5907,7 @@ mod tests { } else { let _lock = store.lock_native_deletion().unwrap(); let (deleting, retained) = store - .opencode_sessions_for_deletion(&["second".into()]) + .native_sessions_for_deletion(&["second".into()], &["opencode", "claude-code"]) .unwrap(); assert_eq!(deleting, ["shared".into()].into()); assert!(retained.is_empty()); @@ -5853,7 +5938,7 @@ mod tests { }; let lock = store.lock_native_deletion().unwrap(); let (deleting, retained) = store - .opencode_sessions_for_deletion(&["first".into()]) + .native_sessions_for_deletion(&["first".into()], &["opencode", "claude-code"]) .unwrap(); assert!(deleting.is_empty()); assert_eq!(retained, ["shared".into()].into());