From 09e1f622992d873ef0d294b348772b9d9ffeadea Mon Sep 17 00:00:00 2001 From: Myles Anderson Date: Thu, 22 Oct 2026 14:42:56 -0700 Subject: [PATCH 1/5] Decide project-delete refusals before purging native chat history A refused project delete (busy run artifacts, data-dir move, runs in flight, artifact cleanup failure, or contended native lock) no longer kills OpenCode history while the chat rows survive. The store now takes the native-deletion lock only after the run-artifacts guard and move check, then cleans artifacts, purges native sessions, and deletes rows, so the native lock is never held across the run-artifacts wait. Part of OR-569. Co-Authored-By: Claude Opus 5.5 --- src/commands/up.rs | 20 ++++----- src/local/chat/mod.rs | 12 +++-- src/store.rs | 100 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 115 insertions(+), 17 deletions(-) diff --git a/src/commands/up.rs b/src/commands/up.rs index 64689a70b..0e20ef233 100644 --- a/src/commands/up.rs +++ b/src/commands/up.rs @@ -2108,20 +2108,16 @@ async fn delete_project_inner(state: AppState, id: String, query: DeleteProjectQ .iter() .map(|session| session.id.clone()) .collect::>(); - // Other deletions fail fast on this lock, so it covers only the purge and the row delete. - let native_deletion = if session_ids.is_empty() { - None - } else { - let lock = store.lock_native_deletion()?; - local::chat::ChatHost::delete_native_sessions(&session_ids).await?; - Some(lock) - }; let project_id = id.clone(); // 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 || store.delete_local_project(&project_id)) - .await - .map_err(|error| anyhow!("project row deletion task failed: {error}"))??; - drop(native_deletion); + tokio::task::spawn_blocking(move || { + let native_purge = (!session_ids.is_empty()).then_some(|| { + local::chat::ChatHost::delete_native_sessions_blocking(&store, &session_ids) + }); + store.delete_local_project_with(&project_id, native_purge) + }) + .await + .map_err(|error| anyhow!("project row deletion task failed: {error}"))??; for session in &sessions { state.chat.claude.forget_session(&session.id).await; local::chat::cleanup_session_transcript_artifacts(&session.id); diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index 9ab06da2b..34f63bd4a 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -6718,12 +6718,16 @@ impl ChatHost { } pub async fn delete_native_sessions(session_ids: &[String]) -> Result<()> { - let (ids, retained) = Store::open()?.opencode_sessions_for_deletion(session_ids)?; + let session_ids = session_ids.to_vec(); tokio::task::spawn_blocking(move || { - crate::local::native_store::delete_opencode_sessions(&ids, &retained) + Self::delete_native_sessions_blocking(&Store::open()?, &session_ids) }) - .await??; - Ok(()) + .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) } } diff --git a/src/store.rs b/src/store.rs index 96b1ea447..c8d86b624 100644 --- a/src/store.rs +++ b/src/store.rs @@ -2130,9 +2130,19 @@ impl Store { )?) } + #[cfg(test)] + pub fn delete_local_project(&self, id: &str) -> Result<()> { + self.delete_local_project_with(id, None:: Result<()>>) + } + /// Delete a project, its rows and its run artifacts; its artifacts folder is the caller's. GitHub repo and cache clone are kept. /// Blocks up to [`RUN_ARTIFACTS_WAIT`] while a submission or orphan scan holds the artifacts lock. - pub fn delete_local_project(&self, id: &str) -> Result<()> { + /// `native_purge` runs under the native-deletion lock after run-artifact cleanup, just before the rows go. + pub fn delete_local_project_with( + &self, + id: &str, + native_purge: Option Result<()>>, + ) -> Result<()> { let mut lock = self.run_artifacts_lock()?; let deadline = std::time::Instant::now() + RUN_ARTIFACTS_WAIT; let _guard = loop { @@ -2156,8 +2166,15 @@ impl Store { "The data directory is being moved; retry project deletion afterwards." )); } + let _native_deletion = native_purge + .is_some() + .then(|| self.lock_native_deletion()) + .transpose()?; // Keep rows on filesystem errors so deletion can be retried with the same handles. self.cleanup_project_run_artifacts(id)?; + if let Some(purge) = native_purge { + purge()?; + } let tx = self.begin()?; tx.execute( "DELETE FROM overleaf_links WHERE project_id = ?1", @@ -4329,6 +4346,87 @@ mod tests { std::fs::remove_dir_all(dir).unwrap(); } + #[test] + fn refused_project_deletion_never_reaches_the_native_purge() { + let dir = std::env::temp_dir().join(format!("orx-delete-order-{}", uuid::Uuid::new_v4())); + let store = Store::open_at(dir.clone()).unwrap(); + project_with_run_artifacts(&store, "project", "run", &"e".repeat(64)); + let mut session = chat_session_fixture("chat"); + session.project_id = "project".into(); + session.harness = "opencode".into(); + session.native_session_id = Some("native".into()); + store.create_chat_session(&session).unwrap(); + let purged = std::cell::Cell::new(false); + let purge = || { + purged.set(true); + Ok(()) + }; + let refuse = |reason: &str| { + let error = store + .delete_local_project_with("project", Some(purge)) + .unwrap_err(); + assert!(error.to_string().contains(reason), "{error}"); + assert!(!purged.get()); + assert!(store.get_chat_session("chat").unwrap().is_some()); + }; + + store + .conn + .execute("UPDATE runs SET status = 'running'", []) + .unwrap(); + refuse("in flight"); + store + .conn + .execute("UPDATE runs SET status = 'done'", []) + .unwrap(); + + let staging_lock = store.run_artifacts_lock().unwrap(); + let staging = staging_lock.read().unwrap(); + refuse("busy"); + drop(staging); + + let moving = store.acquire_data_dir_move_lock().unwrap(); + assert!(store.claim_data_dir_move("move").unwrap()); + refuse("being moved"); + store.release_data_dir_move("move").unwrap(); + drop(moving); + + let native_deletion = store.lock_native_deletion().unwrap(); + refuse("native chat deletion"); + drop(native_deletion); + assert!(dir.join("local-runs/run/repo/output").is_file()); + + #[cfg(unix)] + { + std::fs::rename(dir.join("local-runs"), dir.join("outside")).unwrap(); + std::os::unix::fs::symlink(dir.join("outside"), dir.join("local-runs")).unwrap(); + refuse("Refusing artifact cleanup"); + std::fs::remove_file(dir.join("local-runs")).unwrap(); + std::fs::rename(dir.join("outside"), dir.join("local-runs")).unwrap(); + } + + assert!(store + .delete_local_project_with("project", Some(|| Err(anyhow!("purge failed")))) + .is_err()); + assert!(store.get_chat_session("chat").unwrap().is_some()); + + store + .delete_local_project_with( + "project", + Some(|| { + assert!(store.run_artifacts_lock()?.try_read().is_err()); + assert!(store.lock_native_deletion().is_err()); + purge() + }), + ) + .unwrap(); + assert!(purged.get()); + assert!(store.get_chat_session("chat").unwrap().is_none()); + assert!(store.get_local_project("project").unwrap().is_none()); + drop(store); + std::fs::remove_dir_all(dir).unwrap(); + } + #[test] fn project_deletion_rejects_unsafe_artifact_identifiers() { let dir = std::env::temp_dir().join(format!("orx-delete-unsafe-{}", uuid::Uuid::new_v4())); From 147181ff52951e3d5c1deab0a6eb20a1069825d9 Mon Sep 17 00:00:00 2001 From: Myles Anderson Date: Thu, 22 Oct 2026 15:06:46 -0700 Subject: [PATCH 2/5] Delete chat attachments with their chats and sweep orphans (OR-574) Chat and project deletes now remove the attachment files their messages showed once no remaining message shows them, skipping while a data-dir move is claimed. The periodic reclaim task also sweeps att-* files no message shows after a 24h settle, within the orphan-pass budget. Co-Authored-By: Claude Opus 5.5 --- src/commands/up.rs | 11 +++ src/local/chat/mod.rs | 13 ++- src/store.rs | 205 +++++++++++++++++++++++++++++++++++++++++- 3 files changed, 227 insertions(+), 2 deletions(-) diff --git a/src/commands/up.rs b/src/commands/up.rs index 0e20ef233..f49652704 100644 --- a/src/commands/up.rs +++ b/src/commands/up.rs @@ -4680,6 +4680,17 @@ 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}") + } + Err(error) => eprintln!("orx up: chat attachment cleanup failed: {error}"), + } } } let pause = if more { diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index 34f63bd4a..0eaf7d30d 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -986,8 +986,10 @@ pub struct SavedAttachment { pub is_pdf: bool, } +pub const ATTACHMENTS_DIR: &str = "chat-attachments"; + pub fn attachments_dir() -> Result { - let dir = crate::store::data_dir().join("chat-attachments"); + let dir = crate::store::data_dir().join(ATTACHMENTS_DIR); std::fs::create_dir_all(&dir) .map_err(|e| anyhow!("Could not create {}: {}", dir.display(), e))?; Ok(dir) @@ -1038,6 +1040,15 @@ fn safe_attachment_name(original: Option<&str>, ext: &str) -> String { format!("{safe}.{ext}") } +/// Matches the `att-__.` names `save_images` mints. +pub(crate) fn is_attachment_name(name: &str) -> bool { + name.starts_with("att-") + && !name.contains("..") + && name + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.')) +} + /// Decode pasted/uploaded attachments to the attachments dir. The original file /// name (when present) is preserved after a `__` marker so the transcript and the /// agent see a meaningful name; the uuid prefix keeps names collision-free. diff --git a/src/store.rs b/src/store.rs index c8d86b624..1de3c2c75 100644 --- a/src/store.rs +++ b/src/store.rs @@ -479,6 +479,8 @@ pub(crate) const RUN_ARTIFACTS_POLL: Duration = Duration::from_millis(250); const ORPHAN_PASS_BUDGET: Duration = Duration::from_secs(5); /// Finished runs older than each project's newest few local runs lose their rebuildable environments. 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); const SLIMMED_RUN_PATHS: [&[&str]; 2] = [&["repo", ".venv"], &["repo", ".cache", "uv"]]; const SUBMISSION_HANDLE: (&str, &str) = ("submission-handles", ".json"); @@ -2135,7 +2137,7 @@ impl Store { self.delete_local_project_with(id, None:: Result<()>>) } - /// Delete a project, its rows and its run artifacts; its artifacts folder is the caller's. GitHub repo and cache clone are kept. + /// Delete a project, its rows, run artifacts and chat attachments; its artifacts folder is the caller's. GitHub repo and cache clone are kept. /// Blocks up to [`RUN_ARTIFACTS_WAIT`] while a submission or orphan scan holds the artifacts lock. /// `native_purge` runs under the native-deletion lock after run-artifact cleanup, just before the rows go. pub fn delete_local_project_with( @@ -2176,6 +2178,12 @@ impl Store { purge()?; } let tx = self.begin()?; + let attachments = self + .chat_attachment_names_where( + "m.session_id IN (SELECT id FROM chat_sessions WHERE project_id = ?1)", + [id], + ) + .unwrap_or_default(); tx.execute( "DELETE FROM overleaf_links WHERE project_id = ?1", params![id], @@ -2208,6 +2216,7 @@ impl Store { self.conn .execute("DELETE FROM local_projects WHERE id = ?1", params![id])?; tx.commit()?; + self.remove_unreferenced_chat_attachments(attachments); Ok(()) } @@ -2621,6 +2630,9 @@ impl Store { if has_children { return Err(anyhow!("delete side chats before deleting their parent")); } + let attachments = self + .chat_attachment_names_where("m.session_id = ?1", [id]) + .unwrap_or_default(); tx.execute( "DELETE FROM chat_queued_messages WHERE session_id = ?1", params![id], @@ -2632,9 +2644,92 @@ impl Store { )?; tx.execute("DELETE FROM chat_sessions WHERE id = ?1", params![id])?; tx.commit()?; + self.remove_unreferenced_chat_attachments(attachments); Ok(()) } + /// Attachment file names that image parts of matching messages show; only user messages carry them. + fn chat_attachment_names_where( + &self, + filter: &'static str, + params: impl rusqlite::Params, + ) -> Result> { + let mut query = self.conn.prepare(&format!( + "SELECT DISTINCT json_extract(p.value, '$.text') + FROM chat_messages m, json_each(m.parts_json) p + WHERE m.role = 'user' AND json_extract(p.value, '$.type') = 'image' + AND json_extract(p.value, '$.text') IS NOT NULL AND {filter}" + ))?; + let names = query + .query_map(params, |row| row.get::<_, String>(0))? + .collect::>()?; + Ok(names) + } + + /// Best effort once the rows are gone; the periodic sweep retries what is left. + fn remove_unreferenced_chat_attachments(&self, names: std::collections::HashSet) { + // Another dashboard may be copying this data directory. + if names.is_empty() || self.data_dir_move_claimed().unwrap_or(true) { + return; + } + let Ok(referenced) = self.chat_attachment_names_where("1", []) else { + return; + }; + let dir = self.data_dir.join(crate::local::chat::ATTACHMENTS_DIR); + for name in names.difference(&referenced) { + if crate::local::chat::is_attachment_name(name) { + let _ = std::fs::remove_file(dir.join(name)); + } + } + } + + /// Remove attachment files no message shows; returns whether the budget ran out first. + pub(crate) fn cleanup_orphaned_chat_attachments(&self) -> Result { + // Another dashboard may be copying this data directory. + if self.data_dir_move_claimed()? { + return Ok(false); + } + let entries = + match std::fs::read_dir(self.data_dir.join(crate::local::chat::ATTACHMENTS_DIR)) { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(error.into()), + }; + let mut started = std::time::Instant::now(); + let mut referenced = None; + for entry in entries.flatten() { + if started.elapsed() >= ORPHAN_PASS_BUDGET { + return Ok(true); + } + let Some(name) = entry.file_name().to_str().map(str::to_string) else { + continue; + }; + let settled = entry.metadata().ok().filter(|metadata| { + metadata.is_file() + && metadata + .modified() + .is_ok_and(|at| at.elapsed().unwrap_or_default() >= CHAT_ATTACHMENT_SETTLE) + }); + if !crate::local::chat::is_attachment_name(&name) || settled.is_none() { + continue; + } + // Built only once a candidate exists, and kept out of the removal budget. + let referenced = match &mut referenced { + Some(referenced) => referenced, + None => { + let names = self.chat_attachment_names_where("1", [])?; + started = std::time::Instant::now(); + referenced.insert(names) + } + }; + if !referenced.contains(&name) { + // A file that cannot be removed is retried next pass without stalling the rest. + let _ = std::fs::remove_file(entry.path()); + } + } + Ok(false) + } + /// `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<()> { @@ -4427,6 +4522,114 @@ mod tests { std::fs::remove_dir_all(dir).unwrap(); } + #[test] + fn deleting_chats_removes_only_attachments_no_other_message_shows() { + let dir = std::env::temp_dir().join(format!("orx-attachments-{}", uuid::Uuid::new_v4())); + let store = Store::open_at(dir.clone()).unwrap(); + project_with_run_artifacts(&store, "project", "run", &"a".repeat(64)); + let attachments = dir.join(crate::local::chat::ATTACHMENTS_DIR); + std::fs::create_dir_all(&attachments).unwrap(); + let chats: [(&str, &str, &[&str]); 3] = [ + ( + "first", + "project", + &["att-1__own.png", "att-2__shared.pdf", "keep.txt"], + ), + ("second", "project", &["att-3__own.png"]), + ("other", "elsewhere", &["att-2__shared.pdf"]), + ]; + for (chat, project, files) in chats { + let mut session = chat_session_fixture(chat); + session.project_id = project.into(); + store.create_chat_session(&session).unwrap(); + let parts = files + .iter() + .enumerate() + .map(|(i, file)| serde_json::json!({ "id": format!("img{i}"), "type": "image", "text": file })) + .collect::>(); + store + .upsert_chat_message(&StoredChatMessage { + id: format!("msg_{chat}"), + session_id: chat.into(), + role: "user".into(), + parts_json: serde_json::Value::from(parts).to_string(), + created_at: 1, + completed_at: None, + parent_id: None, + base_native_session_id: None, + result_native_session_id: None, + }) + .unwrap(); + } + for file in [ + "att-1__own.png", + "att-2__shared.pdf", + "att-3__own.png", + "keep.txt", + ] { + std::fs::write(attachments.join(file), "bytes").unwrap(); + } + + store.delete_chat_session("first").unwrap(); + assert!(!attachments.join("att-1__own.png").exists()); + assert!(attachments.join("att-2__shared.pdf").exists()); + store.delete_local_project("project").unwrap(); + assert!(!attachments.join("att-3__own.png").exists()); + assert!(attachments.join("att-2__shared.pdf").exists()); + store.delete_chat_session("other").unwrap(); + assert!(!attachments.join("att-2__shared.pdf").exists()); + assert!(attachments.join("keep.txt").exists()); + drop(store); + std::fs::remove_dir_all(dir).unwrap(); + } + + #[test] + fn attachment_sweep_spares_shown_recent_and_foreign_files() { + let dir = + std::env::temp_dir().join(format!("orx-attachment-sweep-{}", uuid::Uuid::new_v4())); + let store = Store::open_at(dir.clone()).unwrap(); + assert!(!store.cleanup_orphaned_chat_attachments().unwrap()); + store + .create_chat_session(&chat_session_fixture("chat")) + .unwrap(); + store + .upsert_chat_message(&StoredChatMessage { + id: "msg".into(), + session_id: "chat".into(), + role: "user".into(), + parts_json: r#"[{"id":"img0","type":"image","text":"att-1__shown.png"}]"#.into(), + created_at: 1, + completed_at: None, + parent_id: None, + base_native_session_id: None, + result_native_session_id: None, + }) + .unwrap(); + let attachments = dir.join(crate::local::chat::ATTACHMENTS_DIR); + std::fs::create_dir_all(&attachments).unwrap(); + let old = std::time::SystemTime::now() - 2 * CHAT_ATTACHMENT_SETTLE; + for file in ["att-1__shown.png", "att-2__orphan.png", "notes.txt"] { + let file = std::fs::File::create(attachments.join(file)).unwrap(); + file.set_modified(old).unwrap(); + } + std::fs::write(attachments.join("att-3__unsent.png"), "bytes").unwrap(); + + let moving = store.acquire_data_dir_move_lock().unwrap(); + assert!(store.claim_data_dir_move("move").unwrap()); + store.cleanup_orphaned_chat_attachments().unwrap(); + assert!(attachments.join("att-2__orphan.png").exists()); + store.release_data_dir_move("move").unwrap(); + drop(moving); + + assert!(!store.cleanup_orphaned_chat_attachments().unwrap()); + assert!(!attachments.join("att-2__orphan.png").exists()); + for kept in ["att-1__shown.png", "att-3__unsent.png", "notes.txt"] { + assert!(attachments.join(kept).exists(), "{kept}"); + } + drop(store); + std::fs::remove_dir_all(dir).unwrap(); + } + #[test] fn project_deletion_rejects_unsafe_artifact_identifiers() { let dir = std::env::temp_dir().join(format!("orx-delete-unsafe-{}", uuid::Uuid::new_v4())); From 07bc545e58b2dae82b9331e6829d39d5e96fa715 Mon Sep 17 00:00:00 2001 From: Myles Anderson Date: Thu, 22 Oct 2026 15:27:33 -0700 Subject: [PATCH 3/5] Purge Claude Code, Codex and Cursor native history and sweep orphans (OR-573) Chat and project deletes now remove the chat's sessions from orx's isolated Claude Code, Codex and Cursor homes, using every native id the chat recorded, and never touch the user's own homes. OpenCode deletes zero deleted content in place and checkpoint the WAL. The periodic reclaim task sweeps isolated sessions no chat records after a 24h settle (keeping Codex sub-agents of referenced threads), and VACUUMs orx's OpenCode database only while no OpenCode holds it. Co-Authored-By: Claude Opus 5.5 --- src/commands/up.rs | 32 ++-- src/local/chat/mod.rs | 16 +- src/local/native_store.rs | 371 ++++++++++++++++++++++++++++++++++++++ src/local/opencode_db.rs | 93 +++++++++- src/store.rs | 83 +++++++-- 5 files changed, 558 insertions(+), 37 deletions(-) diff --git a/src/commands/up.rs b/src/commands/up.rs index f49652704..3353393ed 100644 --- a/src/commands/up.rs +++ b/src/commands/up.rs @@ -2112,7 +2112,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) }) @@ -4653,7 +4653,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; @@ -4680,16 +4680,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 0eaf7d30d..14616f063 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -6711,7 +6711,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; @@ -6728,17 +6728,21 @@ 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. + let (ids, retained) = + store.native_sessions_for_deletion(session_ids, &["claude-code", "codex", "cursor"])?; + crate::local::native_store::delete_isolated_sessions(&ids, &retained) } } diff --git a/src/local/native_store.rs b/src/local/native_store.rs index db49903d4..6e81bfabd 100644 --- a/src/local/native_store.rs +++ b/src/local/native_store.rs @@ -196,6 +196,245 @@ 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, + retained: &std::collections::BTreeSet, +) -> Result<()> { + if ids.is_empty() { + return Ok(()); + } + remove_isolated_sessions( + &crate::store::data_dir().join("agents"), + ids.difference(retained), + ) +} + +fn remove_isolated_sessions<'a>( + agents: &Path, + ids: impl Iterator, +) -> Result<()> { + let files = isolated_session_files(agents)?; + for id in ids { + for path in files.get(id).into_iter().flatten() { + remove_session_path(path)?; + } + } + Ok(()) +} + +/// 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 codex = agents.join("codex"); + // Without the spawn edges a referenced thread's sub-agents would look orphaned. + let codex_readable = match codex_spawned_threads(&codex, &referenced) { + Ok(spawned) => { + referenced.extend(spawned); + true + } + Err(_) => false, + }; + let settled = |path: &PathBuf| { + std::fs::symlink_metadata(path) + .and_then(|metadata| metadata.modified()) + .is_ok_and(|at| at.elapsed().unwrap_or_default() >= settle) + }; + let sessions = isolated_session_files(agents)?; + let started = std::time::Instant::now(); + for (id, paths) in sessions { + if started.elapsed() >= budget { + return Ok(true); + } + if referenced.contains(&id) + || !paths.iter().all(settled) + || (!codex_readable && paths.iter().any(|path| path.starts_with(&codex))) + { + continue; + } + for path in &paths { + // A path that cannot be removed is retried next pass without stalling the rest. + let _ = remove_session_path(path); + } + } + // After the files, so an unhealthy OpenCode database never holds them back. + let database = agents.join("opencode/opencode.db"); + if !same_home(&database, &opencode_db(NativeStore::Legacy)) { + let orphans = opencode_database::stale_root_sessions(&database, settle)? + .into_iter() + .filter(|id| !referenced.contains(id)) + .collect::>(); + if !orphans.is_empty() { + let retained = referenced.into_iter().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)?; + } + Ok(false) +} + +/// 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(); + 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) + }; + let claude = agents.join("claude"); + if !same_home(&claude, &claude_home(NativeStore::Legacy)) { + 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); + } + } + } + let codex = agents.join("codex"); + if !same_home(&codex, &codex_home(NativeStore::Legacy)) { + 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, + ); + } + } + } + let cursor = agents.join("cursor"); + if !same_home(&cursor, &cursor_home(NativeStore::Legacy)) { + 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) +} + +/// Threads Codex spawned as sub-agents of the given ones, transitively. +fn codex_spawned_threads( + codex: &Path, + parents: &std::collections::HashSet, +) -> Result> { + let database = codex.join("state_5.sqlite"); + if !database.exists() { + return Ok(Vec::new()); + } + 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 +990,136 @@ 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 state = rusqlite::Connection::open(agents.join("codex/state_5.sqlite")).unwrap(); + state + .execute_batch(&format!( + "CREATE TABLE thread_spawn_edges (parent_thread_id TEXT, child_thread_id TEXT, status TEXT); + INSERT INTO thread_spawn_edges VALUES ('{}', '{}', 'done');", + id(6), + id(7) + )) + .unwrap(); + drop(state); + + remove_isolated_sessions(&agents, [id(1)].iter()).unwrap(); + assert!(deleted.iter().all(|path| !path.exists())); + assert!(!project.join(id(1)).exists()); + assert!(orphan.exists()); + + 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, + ] { + 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..80797a34a 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,67 @@ 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) +} + +/// Checkpoints the WAL, then reclaims free pages once they are a tenth of the file while 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)?; + 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"))?; + if lock.try_lock().is_err() || ensure_no_external_users(&path).is_err() { + return Ok(()); + } + 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 +874,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 1de3c2c75..e479cffd3 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);", + )?; 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( @@ -2579,9 +2586,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, @@ -2589,13 +2597,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 @@ -2603,13 +2611,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); @@ -2730,6 +2744,30 @@ 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 { + // Another dashboard may be copying this data directory. + if self.data_dir_move_claimed()? { + return Ok(false); + } + let mut query = self.conn.prepare( + "SELECT native_session_id FROM chat_sessions WHERE native_session_id IS NOT NULL + UNION SELECT base_native_session_id FROM chat_messages + WHERE base_native_session_id IS NOT NULL + UNION SELECT result_native_session_id FROM chat_messages + WHERE result_native_session_id IS NOT NULL", + )?; + let referenced = query + .query_map([], |row| row.get::<_, String>(0))? + .collect::>()?; + crate::local::native_store::sweep_isolated_sessions( + &self.data_dir.join("agents"), + referenced, + NATIVE_SESSION_SETTLE, + ORPHAN_PASS_BUDGET, + ) + } + /// `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<()> { @@ -5702,9 +5740,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(); } @@ -5718,12 +5760,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, @@ -5749,7 +5794,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()); @@ -5780,7 +5825,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()); From d1cc352607d1bbb22f83622c67651abb33691d6f Mon Sep 17 00:00:00 2001 From: Myles Anderson Date: Thu, 22 Oct 2026 18:05:14 -0700 Subject: [PATCH 4/5] Hold off data-dir moves during attachment cleanup; keep sweep progress Attachment removal now holds a shared lock on the data-dir move lock, which a move takes exclusively before claiming its lease, so another dashboard cannot start copying mid-removal. Only removals count against the sweep budget, so a pass that walks many retained files still reaches later orphans. Co-Authored-By: Claude Opus 5.5 --- src/commands/up.rs | 3 +- src/store.rs | 82 ++++++++++++++++++++++++++++++++++++---------- 2 files changed, 66 insertions(+), 19 deletions(-) diff --git a/src/commands/up.rs b/src/commands/up.rs index 16295ac8d..c753d5d4a 100644 --- a/src/commands/up.rs +++ b/src/commands/up.rs @@ -5150,7 +5150,8 @@ async fn move_data_dir(State(state): State, Json(req): Json Result { self.clear_stale_data_dir_move_lease()?; + self.data_dir_move_lease_exists() + } + + fn data_dir_move_lease_exists(&self) -> Result { Ok(self.conn.query_row( "SELECT EXISTS (SELECT 1 FROM data_dir_move_lease WHERE id = 1)", [], @@ -2667,10 +2671,12 @@ impl Store { /// Best effort once the rows are gone; the periodic sweep retries what is left. fn remove_unreferenced_chat_attachments(&self, names: std::collections::HashSet) { - // Another dashboard may be copying this data directory. - if names.is_empty() || self.data_dir_move_claimed().unwrap_or(true) { + if names.is_empty() { return; } + let Ok(Some(_move_hold)) = self.hold_off_data_dir_move() else { + return; + }; let Ok(referenced) = self.chat_attachment_names_where("1", []) else { return; }; @@ -2682,24 +2688,40 @@ impl Store { } } + /// Keeps any dashboard from starting a data-dir move until dropped; `None` while a move holds or claims it. + fn hold_off_data_dir_move(&self) -> Result> { + if self.data_dir_move_claimed()? { + return Ok(None); + } + // A move takes this lock exclusively before claiming its lease. + let hold = open_lifecycle_lock_at(&self.data_dir_move_lock_path)?.into_inner(); + match hold.try_lock_shared() { + Ok(()) => {} + Err(std::fs::TryLockError::WouldBlock) => return Ok(None), + Err(std::fs::TryLockError::Error(error)) => return Err(error.into()), + } + Ok((!self.data_dir_move_lease_exists()?).then_some(hold)) + } + /// Remove attachment files no message shows; returns whether the budget ran out first. pub(crate) fn cleanup_orphaned_chat_attachments(&self) -> Result { - // Another dashboard may be copying this data directory. - if self.data_dir_move_claimed()? { + self.cleanup_orphaned_chat_attachments_within(ORPHAN_PASS_BUDGET) + } + + /// The budget starts at the first removal, so every pass makes progress; removed files leave the listing. + fn cleanup_orphaned_chat_attachments_within(&self, budget: Duration) -> Result { + let Some(_move_hold) = self.hold_off_data_dir_move()? else { return Ok(false); - } + }; let entries = match std::fs::read_dir(self.data_dir.join(crate::local::chat::ATTACHMENTS_DIR)) { Ok(entries) => entries, Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), Err(error) => return Err(error.into()), }; - let mut started = std::time::Instant::now(); + let mut started = None; let mut referenced = None; for entry in entries.flatten() { - if started.elapsed() >= ORPHAN_PASS_BUDGET { - return Ok(true); - } let Some(name) = entry.file_name().to_str().map(str::to_string) else { continue; }; @@ -2712,18 +2734,18 @@ impl Store { if !crate::local::chat::is_attachment_name(&name) || settled.is_none() { continue; } - // Built only once a candidate exists, and kept out of the removal budget. let referenced = match &mut referenced { Some(referenced) => referenced, - None => { - let names = self.chat_attachment_names_where("1", [])?; - started = std::time::Instant::now(); - referenced.insert(names) - } + None => referenced.insert(self.chat_attachment_names_where("1", [])?), }; - if !referenced.contains(&name) { - // A file that cannot be removed is retried next pass without stalling the rest. - let _ = std::fs::remove_file(entry.path()); + if referenced.contains(&name) { + continue; + } + let started = *started.get_or_insert_with(std::time::Instant::now); + // A file that cannot be removed is retried next pass without stalling the rest. + let _ = std::fs::remove_file(entry.path()); + if started.elapsed() >= budget { + return Ok(true); } } Ok(false) @@ -4648,11 +4670,35 @@ mod tests { store.release_data_dir_move("move").unwrap(); drop(moving); + let hold = store.hold_off_data_dir_move().unwrap().unwrap(); + assert!(store.acquire_data_dir_move_lock().is_err()); + drop(hold); + assert!(!store.cleanup_orphaned_chat_attachments().unwrap()); assert!(!attachments.join("att-2__orphan.png").exists()); for kept in ["att-1__shown.png", "att-3__unsent.png", "notes.txt"] { assert!(attachments.join(kept).exists(), "{kept}"); } + + let orphans = ["att-4__a.png", "att-5__b.png", "att-6__c.png"]; + for file in orphans { + let file = std::fs::File::create(attachments.join(file)).unwrap(); + file.set_modified(old).unwrap(); + } + let passes = (0..=orphans.len()) + .take_while(|_| { + store + .cleanup_orphaned_chat_attachments_within(Duration::ZERO) + .unwrap() + }) + .count(); + assert_eq!( + passes, + orphans.len(), + "each exhausted pass still removes one" + ); + assert!(orphans.iter().all(|file| !attachments.join(file).exists())); + assert!(attachments.join("att-1__shown.png").exists()); drop(store); std::fs::remove_dir_all(dir).unwrap(); } From fe1bc79842a0fba67a6550bda88de6ecbb63d2cd Mon Sep 17 00:00:00 2001 From: Myles Anderson Date: Thu, 22 Oct 2026 18:16:41 -0700 Subject: [PATCH 5/5] Settle on the newest file, purge Codex sub-agents, bound the sweep (OR-573) The sweep treats a session as settled only when its newest descendant is, starts its deadline before discovery and rotates which store goes first, with each store progressing at least one removal and failing on its own. Deleting a Codex chat also removes its sub-agent rollouts unless another chat still records them. OpenCode compaction closes its own connection before the Windows exclusive-open check and never waits on readers. Co-Authored-By: Claude Opus 5.5 --- src/local/chat/mod.rs | 8 +- src/local/native_store.rs | 320 +++++++++++++++++++++++++++----------- src/local/opencode_db.rs | 6 +- src/store.rs | 68 ++++++-- 4 files changed, 295 insertions(+), 107 deletions(-) diff --git a/src/local/chat/mod.rs b/src/local/chat/mod.rs index 5e66aeae9..97c7cd904 100644 --- a/src/local/chat/mod.rs +++ b/src/local/chat/mod.rs @@ -6762,9 +6762,13 @@ impl ChatHost { 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. - let (ids, retained) = + // 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, &retained) + 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 6e81bfabd..2bd0dd8fe 100644 --- a/src/local/native_store.rs +++ b/src/local/native_store.rs @@ -199,23 +199,38 @@ pub fn delete_opencode_sessions( /// 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, - retained: &std::collections::BTreeSet, + surviving: &std::collections::HashSet, ) -> Result<()> { if ids.is_empty() { return Ok(()); } - remove_isolated_sessions( - &crate::store::data_dir().join("agents"), - ids.difference(retained), - ) + remove_isolated_sessions(&crate::store::data_dir().join("agents"), ids, surviving) } -fn remove_isolated_sessions<'a>( +/// `surviving` holds every id other chats still record; their Codex sub-agents stay too. +fn remove_isolated_sessions( agents: &Path, - ids: impl Iterator, + 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 ids { + for id in &deleting { for path in files.get(id).into_iter().flatten() { remove_session_path(path)?; } @@ -223,6 +238,35 @@ fn remove_isolated_sessions<'a>( 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, @@ -230,29 +274,53 @@ pub(crate) fn sweep_isolated_sessions( settle: std::time::Duration, budget: std::time::Duration, ) -> Result { - let codex = agents.join("codex"); + 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(&codex, &referenced) { + let codex_readable = match codex_spawned_threads(&agents.join("codex"), &referenced) { Ok(spawned) => { referenced.extend(spawned); true } Err(_) => false, }; - let settled = |path: &PathBuf| { - std::fs::symlink_metadata(path) - .and_then(|metadata| metadata.modified()) - .is_ok_and(|at| at.elapsed().unwrap_or_default() >= settle) - }; - let sessions = isolated_session_files(agents)?; - let started = std::time::Instant::now(); - for (id, paths) in sessions { - if started.elapsed() >= budget { + 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(settled) - || (!codex_readable && paths.iter().any(|path| path.starts_with(&codex))) + || !paths.iter().all(|path| { + newest_modified(path).is_some_and(|at| at.elapsed().unwrap_or_default() >= settle) + }) { continue; } @@ -260,34 +328,57 @@ pub(crate) fn sweep_isolated_sessions( // 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); + } } - // After the files, so an unhealthy OpenCode database never holds them back. + 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)) { - let orphans = opencode_database::stale_root_sessions(&database, settle)? - .into_iter() - .filter(|id| !referenced.contains(id)) - .collect::>(); - if !orphans.is_empty() { - let retained = referenced.into_iter().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); - } + 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)?; } - Ok(false) + 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. @@ -302,6 +393,19 @@ fn same_home(isolated: &Path, legacy: &Path) -> bool { /// 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| { @@ -315,62 +419,73 @@ fn isolated_session_files( .and_then(|name| name.to_str()) .map(str::to_string) }; - let claude = agents.join("claude"); - if !same_home(&claude, &claude_home(NativeStore::Legacy)) { - 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, - ); + match home { + IsolatedHome::Claude => { + let claude = agents.join("claude"); + if same_home(&claude, &claude_home(NativeStore::Legacy)) { + return Ok(sessions); } - } - for tree in ["session-env", "file-history"] { - for path in child_paths(&claude.join(tree))? { - add(name(&path).as_deref(), path); + 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); + } } } - } - let codex = agents.join("codex"); - if !same_home(&codex, &codex_home(NativeStore::Legacy)) { - 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::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, + ); + } } } - } - let cursor = agents.join("cursor"); - if !same_home(&cursor, &cursor_home(NativeStore::Legacy)) { - for workspace in child_dirs(&cursor.join("chats"))? { - for path in child_paths(&workspace)? { + 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); } } - for path in child_paths(&cursor.join("acp-sessions"))? { - add(name(&path).as_deref(), path); - } } Ok(sessions) } -/// Threads Codex spawned as sub-agents of the given ones, transitively. +/// The given threads plus every sub-agent Codex spawned from them, transitively. fn codex_spawned_threads( codex: &Path, parents: &std::collections::HashSet, -) -> Result> { +) -> Result> { let database = codex.join("state_5.sqlite"); if !database.exists() { - return Ok(Vec::new()); + return Ok(parents.clone()); } let connection = rusqlite::Connection::open_with_flags( database, @@ -1042,21 +1157,45 @@ mod tests { 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(&format!( - "CREATE TABLE thread_spawn_edges (parent_thread_id TEXT, child_thread_id TEXT, status TEXT); - INSERT INTO thread_spawn_edges VALUES ('{}', '{}', 'done');", - id(6), - id(7) - )) + .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); - remove_isolated_sessions(&agents, [id(1)].iter()).unwrap(); + 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!( @@ -1071,6 +1210,7 @@ mod tests { &referenced_rollout, &spawned_rollout, ¬_a_session, + &active_chat, ] { assert!(kept.exists(), "{}", kept.display()); } diff --git a/src/local/opencode_db.rs b/src/local/opencode_db.rs index 80797a34a..1962f5642 100644 --- a/src/local/opencode_db.rs +++ b/src/local/opencode_db.rs @@ -300,13 +300,14 @@ pub(super) fn stale_root_sessions( Ok(ids) } -/// Checkpoints the WAL, then reclaims free pages once they are a tenth of the file while no OpenCode holds it. +/// 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()", @@ -323,9 +324,12 @@ pub(super) fn compact(path: &Path) -> Result<()> { .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(())); diff --git a/src/store.rs b/src/store.rs index adf42bbda..d900a5ca9 100644 --- a/src/store.rs +++ b/src/store.rs @@ -841,7 +841,7 @@ impl Store { // 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);", + 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;")?; @@ -2767,26 +2767,38 @@ impl Store { /// Reclaim orx-launched agent history no chat records; returns whether the budget ran out first. pub(crate) fn cleanup_orphaned_native_sessions(&self) -> Result { - // Another dashboard may be copying this data directory. - if self.data_dir_move_claimed()? { + 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 + "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", + WHERE result_native_session_id IS NOT NULL + AND session_id NOT IN (SELECT value FROM json_each(?1))", )?; - let referenced = query - .query_map([], |row| row.get::<_, String>(0))? + let ids = query + .query_map([serde_json::to_string(chat_ids)?], |row| { + row.get::<_, String>(0) + })? .collect::>()?; - crate::local::native_store::sweep_isolated_sessions( - &self.data_dir.join("agents"), - referenced, - NATIVE_SESSION_SETTLE, - ORPHAN_PASS_BUDGET, - ) + Ok(ids) } /// `None` clears it, which a fork of the session's very first turn needs: @@ -5856,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";