diff --git a/.changeset/session-catchup-hook-trust.md b/.changeset/session-catchup-hook-trust.md new file mode 100644 index 000000000..0889f5cf7 --- /dev/null +++ b/.changeset/session-catchup-hook-trust.md @@ -0,0 +1,5 @@ +--- +"tracedecay": patch +--- + +Coalesce transcript catch-ups, route Hermes history through one shared source sweep, harden branch database recovery markers, emit Codex hook trust state in the exact form required for non-interactive approval, and reduce test-profile link overhead while preserving line-table backtraces. diff --git a/Cargo.toml b/Cargo.toml index c0d1ed275..088bb3160 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -183,6 +183,12 @@ jsonschema = { version = "0.46.8", default-features = false } name = "large_repos" harness = false +# Full debuginfo makes each large integration-test binary 500-700 MB and +# dominates local/CI link time. Level 1 preserves source line tables for +# useful panic backtraces while substantially shrinking linker input/output. +[profile.test] +debug = 1 + # Compile the SQLite stack with optimizations even in dev/test builds. The # bundled SQLite C sources in libsql-ffi are otherwise built at -O0 (cc honors # cargo's OPT_LEVEL), and nearly every test creates databases and runs many diff --git a/src/agents/codex.rs b/src/agents/codex.rs index c0c781149..e69c9a861 100644 --- a/src/agents/codex.rs +++ b/src/agents/codex.rs @@ -1041,10 +1041,38 @@ fn sync_codex_hook_trust(home: &Path, tracedecay_bin: &str) -> Result Result<()> { + super::backup_file(config_path)?; + let contents = toml::to_string_pretty(config).map_err(|error| TraceDecayError::Config { + message: format!("failed to serialize {}: {error}", config_path.display()), + })?; + let Some(child_offset) = contents.find("[hooks.state.\"") else { + return Err(TraceDecayError::Config { + message: "Codex hook trust state serialized without hook entries".to_string(), + }); + }; + let mut updated = String::with_capacity(contents.len() + "[hooks.state]\n\n".len()); + updated.push_str(&contents[..child_offset]); + updated.push_str("[hooks.state]\n\n"); + updated.push_str(&contents[child_offset..]); + std::fs::write(config_path, updated).map_err(|error| TraceDecayError::Config { + message: format!("failed to write {}: {error}", config_path.display()), + })?; + eprintln!("\x1b[32m✔\x1b[0m Wrote {}", config_path.display()); + Ok(()) +} + +fn codex_hook_state_table_is_explicit(contents: &str) -> bool { + contents.lines().any(|line| line.trim() == "[hooks.state]") +} + /// Auto-trust the installed plugin's hooks, printing a concise confirmation on /// full success and falling back to [`print_hook_trust_guidance`] whenever a /// hook is skipped by the safety valve or the config could not be written. @@ -1855,8 +1883,17 @@ fn doctor_check_hooks( let entries = codex_hook_trust_entries_for_marketplace(&hooks, marketplace_name); match load_toml_file(config_path) { Ok(config) => match codex_plugin_hook_trust_state(&config, &entries) { - CodexHookTrustState::Trusted => dc.pass(&format!( - "Codex hook trust entries recorded and current in {}", + CodexHookTrustState::Trusted + if std::fs::read_to_string(config_path) + .is_ok_and(|contents| codex_hook_state_table_is_explicit(&contents)) => + { + dc.pass(&format!( + "Codex hook trust entries recorded and current in {}", + config_path.display() + )); + } + CodexHookTrustState::Trusted => dc.warn(&format!( + "Codex hook trust records in {} lack an explicit [hooks.state] table, so Codex still requests review; run `tracedecay update-plugin` to repair and auto-trust them", config_path.display() )), CodexHookTrustState::Missing(missing) => dc.info(&format!( diff --git a/src/agents/codex/tests.rs b/src/agents/codex/tests.rs index 21c48dc70..097777bab 100644 --- a/src/agents/codex/tests.rs +++ b/src/agents/codex/tests.rs @@ -263,10 +263,33 @@ trusted_hash = "sha256:foreign" "#, ) .unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&config_path, std::fs::Permissions::from_mode(0o600)).unwrap(); + } let outcome = sync_codex_hook_trust(home.path(), TEST_BIN).unwrap(); assert_eq!(outcome.trusted, CODEX_MANAGED_HOOKS.len()); assert!(outcome.skipped.is_empty()); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + assert_eq!( + std::fs::metadata(&config_path) + .unwrap() + .permissions() + .mode() + & 0o777, + 0o600 + ); + } + + let config_text = std::fs::read_to_string(&config_path).unwrap(); + assert!( + config_text.lines().any(|line| line == "[hooks.state]"), + "Codex requires an explicit [hooks.state] parent table before trusting child records" + ); let entries = managed_entries(TEST_BIN); let config = load_toml_file(&config_path).unwrap(); diff --git a/src/db/connection.rs b/src/db/connection.rs index 820d9ff26..4a9bb0e68 100644 --- a/src/db/connection.rs +++ b/src/db/connection.rs @@ -256,13 +256,45 @@ impl Database { /// This ensures all committed transactions are merged into the main DB /// before the process exits, preventing a stale WAL file on next startup. pub async fn checkpoint(&self) -> Result<()> { - self.conn - .execute_batch("PRAGMA wal_checkpoint(TRUNCATE);") + let mut rows = self + .conn + .query("PRAGMA wal_checkpoint(TRUNCATE);", ()) .await .map_err(|e| TraceDecayError::Database { message: format!("failed to checkpoint WAL: {e}"), operation: "checkpoint".to_string(), })?; + let row = rows + .next() + .await + .map_err(|e| TraceDecayError::Database { + message: format!("failed to read WAL checkpoint status: {e}"), + operation: "checkpoint".to_string(), + })? + .ok_or_else(|| TraceDecayError::Database { + message: "WAL checkpoint returned no status row".to_string(), + operation: "checkpoint".to_string(), + })?; + let busy: i64 = row.get(0).map_err(|e| TraceDecayError::Database { + message: format!("failed to read WAL checkpoint busy status: {e}"), + operation: "checkpoint".to_string(), + })?; + let log_frames: i64 = row.get(1).map_err(|e| TraceDecayError::Database { + message: format!("failed to read WAL checkpoint frame count: {e}"), + operation: "checkpoint".to_string(), + })?; + let checkpointed_frames: i64 = row.get(2).map_err(|e| TraceDecayError::Database { + message: format!("failed to read WAL checkpoint completion count: {e}"), + operation: "checkpoint".to_string(), + })?; + if busy != 0 || checkpointed_frames < log_frames { + return Err(TraceDecayError::Database { + message: format!( + "WAL checkpoint incomplete: busy={busy}, log_frames={log_frames}, checkpointed_frames={checkpointed_frames}" + ), + operation: "checkpoint".to_string(), + }); + } Ok(()) } diff --git a/src/doctor.rs b/src/doctor.rs index 5b0a25e5d..b1b2b5055 100644 --- a/src/doctor.rs +++ b/src/doctor.rs @@ -309,7 +309,10 @@ fn database_recovery_guidance(db_path: &Path) -> String { let wal_path = db_path.with_extension("db-wal"); let shm_path = db_path.with_extension("db-shm"); let data_root = db_path.parent().unwrap_or_else(|| Path::new(".")); - let dirty_path = data_root.join("dirty"); + let mut graph_dirty = db_path.as_os_str().to_os_string(); + graph_dirty.push(".dirty"); + let graph_dirty = PathBuf::from(graph_dirty); + let legacy_dirty = data_root.join("dirty"); let sessions_path = data_root.join(crate::storage::SESSIONS_DB_FILENAME); format!( @@ -318,7 +321,8 @@ fn database_recovery_guidance(db_path: &Path) -> String { DB: {}\n\ WAL: {}\n\ SHM: {}\n\ - dirty sentinel: {}\n\ + graph dirty sentinel: {}\n\ + legacy dirty sentinel (if present): {}\n\ `sessions.db` is separate and must not be removed: {}\n\ Facts are stored in the graph database; automatic rebuild is intentionally blocked because it cannot preserve them generically.\n\ Do not run `tracedecay init`, `tracedecay sync --force`, or `tracedecay wipe` until that recovery set is safely copied.\n\ @@ -326,7 +330,8 @@ fn database_recovery_guidance(db_path: &Path) -> String { db_path.display(), wal_path.display(), shm_path.display(), - dirty_path.display(), + graph_dirty.display(), + legacy_dirty.display(), sessions_path.display(), ) } diff --git a/src/doctor/tests.rs b/src/doctor/tests.rs index 0acf7c3b6..7f36df691 100644 --- a/src/doctor/tests.rs +++ b/src/doctor/tests.rs @@ -215,10 +215,15 @@ fn database_recovery_guidance_names_the_preserved_recovery_set() { let db_path = PathBuf::from("/profile/projects/proj_test/tracedecay.db"); let guidance = database_recovery_guidance(&db_path); - assert!(guidance.contains("/profile/projects/proj_test/tracedecay.db")); - assert!(guidance.contains("/profile/projects/proj_test/tracedecay.db-wal")); - assert!(guidance.contains("/profile/projects/proj_test/tracedecay.db-shm")); - assert!(guidance.contains("/profile/projects/proj_test/dirty")); + for path in [ + db_path.clone(), + db_path.with_extension("db-wal"), + db_path.with_extension("db-shm"), + PathBuf::from(format!("{}.dirty", db_path.display())), + db_path.parent().unwrap().join("dirty"), + ] { + assert!(guidance.contains(&path.display().to_string())); + } assert!(guidance.contains("stop all TraceDecay daemon and MCP processes")); assert!( guidance.contains( diff --git a/src/global_db.rs b/src/global_db.rs index 512d01d05..27a03c8e4 100644 --- a/src/global_db.rs +++ b/src/global_db.rs @@ -971,20 +971,38 @@ impl GlobalDb { /// in-process and retried briefly to also cover a racing *external* /// process (e.g. two MCP servers starting simultaneously). pub async fn open_at(db_path: &std::path::Path) -> Option { + Self::open_at_with_backfill(db_path, true).await + } + + /// Opens and ensures a writable session store without starting detached + /// structured backfill. Bulk multi-store catch-up uses this to avoid + /// launching one competing backfill task per registered project. + pub async fn open_at_without_structured_backfill(db_path: &std::path::Path) -> Option { + Self::open_at_with_backfill(db_path, false).await + } + + async fn open_at_with_backfill( + db_path: &std::path::Path, + spawn_structured_backfill: bool, + ) -> Option { static OPEN_ENSURE_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); let _guard = OPEN_ENSURE_LOCK.lock().await; for attempt in 0..3_u64 { if attempt > 0 { tokio::time::sleep(std::time::Duration::from_millis(50 * attempt)).await; } - if let Some(db) = Self::open_at_unsynchronized(db_path).await { + if let Some(db) = Self::open_at_unsynchronized(db_path, spawn_structured_backfill).await + { return Some(db); } } None } - async fn open_at_unsynchronized(db_path: &std::path::Path) -> Option { + async fn open_at_unsynchronized( + db_path: &std::path::Path, + spawn_structured_backfill: bool, + ) -> Option { if let Some(parent) = db_path.parent() { std::fs::create_dir_all(parent).ok()?; } @@ -1204,7 +1222,9 @@ impl GlobalDb { // runs on every open (per hook event, per CLI/MCP invocation), so it // must not block: schedule it on a detached background task rather than // synchronously reading and re-parsing a batch of multi-MB transcripts. - db.spawn_structured_backfill(); + if spawn_structured_backfill { + db.spawn_structured_backfill(); + } Some(db) } @@ -3477,10 +3497,17 @@ impl GlobalDb { return false; } } - if !self - .set_parse_offset_in_existing_tx(parse_offset_path, parse_offset) - .await - { + let cursor_set = match mode { + TranscriptWriteMode::Full => { + self.set_parse_offset_in_existing_tx(parse_offset_path, parse_offset) + .await + } + TranscriptWriteMode::ProjectionOnly => { + self.set_parse_offset_monotonic_in_existing_tx(parse_offset_path, parse_offset) + .await + } + }; + if !cursor_set { let _ = self.conn.execute("ROLLBACK", ()).await; return false; } @@ -4851,6 +4878,39 @@ impl GlobalDb { let _ = self.set_parse_offset_in_existing_tx(path, offset).await; } + /// Advances a row-style parse cursor without allowing an overlapping, + /// older sweep to move it backwards. + pub async fn advance_parse_offset(&self, path: &str, offset: ParseOffset) { + let _ = self + .set_parse_offset_monotonic_in_existing_tx(path, offset) + .await; + } + + async fn set_parse_offset_monotonic_in_existing_tx( + &self, + path: &str, + offset: ParseOffset, + ) -> bool { + self.conn + .execute( + "INSERT INTO parse_offsets (file_path, byte_offset, mtime, file_id) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(file_path) DO UPDATE SET + byte_offset = excluded.byte_offset, + mtime = excluded.mtime, + file_id = excluded.file_id + WHERE excluded.byte_offset >= parse_offsets.byte_offset", + params![ + path, + offset.byte_offset as i64, + offset.mtime as i64, + offset.file_id as i64 + ], + ) + .await + .is_ok() + } + async fn set_parse_offset_in_existing_tx(&self, path: &str, offset: ParseOffset) -> bool { if self .conn diff --git a/src/mcp/tools/handlers/session.rs b/src/mcp/tools/handlers/session.rs index ae8de0d87..f774c92be 100644 --- a/src/mcp/tools/handlers/session.rs +++ b/src/mcp/tools/handlers/session.rs @@ -1,7 +1,7 @@ -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::fmt::Write as _; use std::path::{Path, PathBuf}; -use std::sync::{LazyLock, Mutex}; +use std::sync::{Arc, LazyLock, Mutex}; use serde_json::{Map, Value, json}; @@ -50,6 +50,53 @@ const MAX_LCM_EXPAND_QUERY_SYNTHESIS_PROMPT_CHARS: usize = 2_048; const MESSAGE_SEARCH_SNIPPET_CHARS: usize = 240; +static MESSAGE_CATCH_UPS: LazyLock>>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +struct MessageCatchUpLeader { + key: String, + done: Arc>, +} + +impl Drop for MessageCatchUpLeader { + fn drop(&mut self) { + if let Ok(mut catch_ups) = MESSAGE_CATCH_UPS.lock() + && catch_ups + .get(&self.key) + .is_some_and(|current| Arc::ptr_eq(current, &self.done)) + { + catch_ups.remove(&self.key); + } + let _ = self.done.send(true); + } +} + +enum MessageCatchUpClaim { + Leader(MessageCatchUpLeader), + Wait(tokio::sync::watch::Receiver), +} + +fn claim_message_catch_up(key: String) -> MessageCatchUpClaim { + let mut catch_ups = MESSAGE_CATCH_UPS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(done) = catch_ups.get(&key) { + return MessageCatchUpClaim::Wait(done.subscribe()); + } + let (done, _) = tokio::sync::watch::channel(false); + let done = Arc::new(done); + catch_ups.insert(key.clone(), Arc::clone(&done)); + MessageCatchUpClaim::Leader(MessageCatchUpLeader { key, done }) +} + +async fn wait_for_message_catch_up(mut done: tokio::sync::watch::Receiver) { + while !*done.borrow_and_update() { + if done.changed().await.is_err() { + break; + } + } +} + /// Renders `tracedecay_message_search` results as compact markdown. Each hit /// shows provider, session (id + title), role, timestamp, and score with a /// plain-text snippet of the message body — deliberately dropping the raw @@ -1744,6 +1791,17 @@ async fn open_session_db_with_cached_ensure(db_path: &Path) -> Option Some(db) } +async fn open_session_db_for_bulk_catch_up(db_path: &Path) -> Option { + if schema_already_ensured(db_path) { + if let Some(db) = GlobalDb::open_at_assuming_schema(db_path).await { + return Some(db); + } + } + let db = GlobalDb::open_at_without_structured_backfill(db_path).await?; + mark_schema_ensured(db_path); + Some(db) +} + enum LcmStorageResolution { Available(Box), Unavailable(ToolResult), @@ -2254,8 +2312,24 @@ pub(super) async fn handle_message_search( .ok_or_else(|| TraceDecayError::Config { message: "could not resolve tracedecay profile root".to_string(), })?; - let mut results = Vec::new(); - let mut searched_project_count = 0usize; + let catch_up_leader = if request.catch_up { + let key = format!( + "all-registered:{}:{}", + profile_root.display(), + request.provider_scope.response_label() + ); + match claim_message_catch_up(key) { + MessageCatchUpClaim::Wait(done) => { + wait_for_message_catch_up(done).await; + None + } + MessageCatchUpClaim::Leader(leader) => Some(leader), + } + } else { + None + }; + let perform_catch_up = catch_up_leader.is_some(); + let mut destinations = Vec::new(); let mut skipped_project_count = 0usize; for project in global.list_code_projects(usize::MAX).await { let Some(context) = global @@ -2276,8 +2350,8 @@ pub(super) async fn handle_message_search( skipped_project_count += 1; continue; }; - let db = if request.catch_up { - open_session_db_with_cached_ensure(&db_path).await + let db = if perform_catch_up { + open_session_db_for_bulk_catch_up(&db_path).await } else { GlobalDb::open_read_only_at(&db_path).await }; @@ -2285,21 +2359,47 @@ pub(super) async fn handle_message_search( skipped_project_count += 1; continue; }; - searched_project_count += 1; - if request.catch_up { - let display_root = Path::new(&context.project.display_root); - let project_root = if display_root.is_absolute() { - display_root - } else { - Path::new(&context.project.canonical_root) - }; - let _ = crate::sessions::ingest_global_sources_for_provider( - &db, - project_root, - request.provider_scope.provider(), - ) - .await; + let display_root = Path::new(&context.project.display_root); + let project_root = if display_root.is_absolute() { + display_root.to_path_buf() + } else { + PathBuf::from(&context.project.canonical_root) + }; + destinations.push((db, project_root)); + } + if perform_catch_up { + let provider = request.provider_scope.provider(); + let _ = crate::sessions::ingest_user_global_sources_for_provider(provider).await; + if provider.is_none() || provider == Some(crate::sessions::SessionProvider::Hermes) { + let hermes_destinations = + destinations + .iter() + .map(|(db, project_root)| { + crate::sessions::hermes::ProjectIngestDestination { db, project_root } + }) + .collect::>(); + let _ = crate::sessions::hermes::ingest_for_projects(&hermes_destinations).await; + } + if provider == Some(crate::sessions::SessionProvider::Hermes) { + for (db, project_root) in &destinations { + crate::sessions::finalize_project_ingest(db, project_root).await; + } + } else { + for (db, project_root) in &destinations { + let _ = crate::sessions::ingest_project_sources_for_provider( + db, + project_root, + provider, + false, + ) + .await; + } } + } + drop(catch_up_leader); + let searched_project_count = destinations.len(); + let mut results = Vec::new(); + for (db, _) in &destinations { let mut project_results = search_session_messages_in_db(&db, &request).await; results.append(&mut project_results); } @@ -2347,7 +2447,29 @@ pub(super) async fn handle_message_search( }), )); }; - let Some(db) = open_session_db_with_cached_ensure(&db_path).await else { + let catch_up_leader = if request.catch_up { + let key = format!( + "project:{}:{}", + db_path.display(), + request.provider_scope.response_label() + ); + match claim_message_catch_up(key) { + MessageCatchUpClaim::Wait(done) => { + wait_for_message_catch_up(done).await; + None + } + MessageCatchUpClaim::Leader(leader) => Some(leader), + } + } else { + None + }; + let perform_catch_up = catch_up_leader.is_some(); + let db = if request.catch_up && !perform_catch_up { + GlobalDb::open_read_only_at(&db_path).await + } else { + open_session_db_with_cached_ensure(&db_path).await + }; + let Some(db) = db else { return Ok(tool_json( Some(cg.project_root()), &args, @@ -2360,7 +2482,7 @@ pub(super) async fn handle_message_search( )); }; let catch_up_performed = request.catch_up; - if catch_up_performed { + if perform_catch_up { let _ = crate::sessions::ingest_global_sources_for_provider( &db, &target_root, @@ -2368,6 +2490,7 @@ pub(super) async fn handle_message_search( ) .await; } + drop(catch_up_leader); // Build the workflow-run scope filter and, separately, resolve the run's // parent thread purely for the echoed `workflow_run_parent_session` field // (the scope itself is authoritative via the `workflow_agents` EXISTS diff --git a/src/mcp/tools/handlers/session/tests.rs b/src/mcp/tools/handlers/session/tests.rs index af5713fb6..446b4b991 100644 --- a/src/mcp/tools/handlers/session/tests.rs +++ b/src/mcp/tools/handlers/session/tests.rs @@ -2,6 +2,30 @@ use super::*; use crate::mcp::response_handles::lock_response_handle_store; use crate::sessions::{SessionMessageRecord, SessionRecord}; +#[tokio::test] +async fn identical_message_catch_ups_share_one_leader() { + let key = format!("test-singleflight-{}", std::process::id()); + let leader = match claim_message_catch_up(key.clone()) { + MessageCatchUpClaim::Leader(leader) => leader, + MessageCatchUpClaim::Wait(_) => panic!("fresh key unexpectedly had a leader"), + }; + let waiter = match claim_message_catch_up(key.clone()) { + MessageCatchUpClaim::Wait(waiter) => waiter, + MessageCatchUpClaim::Leader(_) => panic!("identical catch-up did not coalesce"), + }; + drop(leader); + tokio::time::timeout( + std::time::Duration::from_secs(1), + wait_for_message_catch_up(waiter), + ) + .await + .expect("waiter was not released when leader completed"); + assert!(matches!( + claim_message_catch_up(key), + MessageCatchUpClaim::Leader(_) + )); +} + /// Build a search hit with a given relevance score, message timestamp, and /// stable ids — enough to exercise the cross-project merge ordering contract. fn merge_result( diff --git a/src/sessions/hermes.rs b/src/sessions/hermes.rs index 9404239f0..528a1257f 100644 --- a/src/sessions/hermes.rs +++ b/src/sessions/hermes.rs @@ -31,12 +31,13 @@ use std::collections::{BTreeSet, HashMap}; use std::path::{Path, PathBuf}; +use rayon::prelude::*; use serde_json::{Map, Value}; use crate::agents::hermes::read_config_pinned_project_root; use crate::global_db::{GlobalDb, ParseOffset, TranscriptBatch}; use crate::sessions::shared::{ - NewRows, StoredCursor, TranscriptIngestStats, TranscriptLocation, + NewRows, ProjectRootMatcher, StoredCursor, TranscriptIngestStats, TranscriptLocation, TranscriptLocationMetadataKeys, append_location_metadata, content_storage_text_and_tools, path_belongs_to_project, preview_title, title_from_messages, }; @@ -66,6 +67,55 @@ pub async fn ingest_for_project(db: &GlobalDb, project_root: &Path) -> Transcrip ingest_homes(db, &homes, project_root).await } +/// One project-store destination for a shared Hermes source sweep. +#[derive(Clone, Copy)] +pub struct ProjectIngestDestination<'a> { + pub db: &'a GlobalDb, + pub project_root: &'a Path, +} + +/// Ingests Hermes history for several registered projects while opening and +/// scanning each profile `state.db` only once. Every destination retains its +/// own durable row cursor and advances it in the same transaction as its +/// projection writes. +pub async fn ingest_for_projects( + destinations: &[ProjectIngestDestination<'_>], +) -> TranscriptIngestStats { + let homes = crate::sessions::home_dir() + .map(|home| vec![home.join(".hermes")]) + .unwrap_or_default(); + ingest_homes_for_projects(&homes, destinations).await +} + +/// Test seam for [`ingest_for_projects`]. +pub async fn ingest_homes_for_projects( + hermes_homes: &[PathBuf], + destinations: &[ProjectIngestDestination<'_>], +) -> TranscriptIngestStats { + let mut stats = TranscriptIngestStats::default(); + for source in all_profile_sources(hermes_homes) { + let eligible = destinations + .iter() + .copied() + .filter(|destination| { + source_is_candidate_for_project(&source, destination.project_root) + }) + .collect::>(); + if eligible.is_empty() { + continue; + } + match try_ingest_state_db_for_projects(&source, &eligible).await { + Ok(source_stats) => stats = stats.merge(source_stats), + Err(error) => tracing::debug!( + state_db = %source.state_db.display(), + error, + "skipping shared Hermes transcript source" + ), + } + } + stats +} + /// [`ingest_for_project`] with explicit Hermes home directories — the test /// seam for pointing the sweep at a temporary home instead of the real /// `~/.hermes`. @@ -245,6 +295,19 @@ fn candidate_state_dbs(hermes_homes: &[PathBuf], project_root: &Path) -> Vec bool { + if source + .legacy_project_pin + .as_deref() + .is_some_and(|pin| !path_belongs_to_project(pin, project_root)) + { + return false; + } + source.legacy_project_pin.is_some() + || crate::worktree::git_worktree_root(project_root).is_some() + || crate::config::has_project_database(project_root) +} + /// One joined `messages` × `sessions` row read past the cursor. struct HermesRow { id: i64, @@ -355,7 +418,6 @@ async fn try_ingest_state_db( &message_columns(&conn).await, &table_columns(&conn, "sessions").await, ); - loop { let new = read_new_rows_strict(&conn, &select_sql, cursor).await?; let row_count = new.items.len(); @@ -376,7 +438,7 @@ async fn try_ingest_state_db( if batches.is_empty() { // Only non-conversation rows (e.g. `session_meta`) — still advance // the cursor so the next sweep does not re-read them. - db.set_parse_offset(&cursor_path, offset).await; + db.advance_parse_offset(&cursor_path, offset).await; } else { let message_count: u64 = batches .iter() @@ -404,6 +466,168 @@ async fn try_ingest_state_db( } } +struct ProjectDestinationState<'a> { + destination: ProjectIngestDestination<'a>, + cursor: StoredCursor, + sessions_seen: BTreeSet, + writable: bool, + cursor_pending: bool, +} + +/// Shared-source equivalent of [`try_ingest_state_db`]. Source rows are read +/// from the lowest destination cursor; destinations already ahead skip the +/// prefix and independently commit their projection plus cursor. +async fn try_ingest_state_db_for_projects( + source: &HermesProfileSource, + destinations: &[ProjectIngestDestination<'_>], +) -> Result { + let state_db = &source.state_db; + let conn = open_read_only_strict(state_db).await?; + let path_str = state_db.to_string_lossy().to_string(); + let cursor_path = format!("{path_str}#{CORRELATION_CURSOR_VERSION}"); + let mut states = Vec::with_capacity(destinations.len()); + for destination in destinations { + let prev = destination + .db + .get_parse_offset(&cursor_path) + .await + .unwrap_or_default(); + states.push(ProjectDestinationState { + destination: *destination, + cursor: StoredCursor { + position: prev.byte_offset, + mtime: prev.mtime, + file_id: prev.file_id, + }, + sessions_seen: BTreeSet::new(), + writable: true, + cursor_pending: false, + }); + } + let select_sql = select_new_messages_sql( + &message_columns(&conn).await, + &table_columns(&conn, "sessions").await, + ); + let destination_matchers = states + .par_iter() + .map(|state| ProjectRootMatcher::new(state.destination.project_root)) + .collect::>(); + let mut destination_routes = HashMap::>::new(); + let mut read_cursor = StoredCursor { + position: states + .iter() + .map(|state| state.cursor.position) + .min() + .unwrap_or_default(), + mtime: 0, + file_id: 0, + }; + let mut stats = TranscriptIngestStats::default(); + + loop { + let new = read_new_rows_strict(&conn, &select_sql, read_cursor).await?; + let row_count = new.items.len(); + if row_count == 0 { + break; + } + let source_position = new.new_cursor.position; + let mtime = file_mtime_secs(state_db); + let destination_locations = turn_project_locations_for_destinations( + &new.items, + &destination_matchers, + source, + &mut destination_routes, + ); + for (state_index, state) in states + .iter_mut() + .enumerate() + .filter(|(_, state)| state.writable) + { + if source_position <= state.cursor.position { + continue; + } + let destination = &destination_locations[state_index]; + let first_new = destination + .row_indices + .partition_point(|&index| new.items[index].id as u64 <= state.cursor.position); + let next_cursor = StoredCursor { + position: source_position, + mtime, + file_id: 0, + }; + let offset = ParseOffset { + byte_offset: next_cursor.position, + mtime: next_cursor.mtime, + file_id: 0, + }; + let batches = build_batches_with_locations( + state.destination.db, + &new.items, + &path_str, + state.destination.project_root, + source, + &destination.by_row_id, + Some(&destination.row_indices[first_new..]), + ) + .await; + if batches.is_empty() { + // Cursor-only transactions across every registered project + // dominate cold catch-up. Defer them to one final write; a + // crash before that point merely causes an idempotent rescan. + state.cursor = next_cursor; + state.cursor_pending = true; + continue; + } + if !state + .destination + .db + .upsert_transcript_projection_batches(&batches, &cursor_path, offset) + .await + { + state.writable = false; + continue; + } + stats.messages_upserted = stats.messages_upserted.saturating_add( + batches + .iter() + .map(|batch| batch.messages.len() as u64) + .sum::(), + ); + for batch in &batches { + state.sessions_seen.insert(batch.session.session_id.clone()); + } + state.cursor = next_cursor; + state.cursor_pending = false; + } + read_cursor.position = source_position; + if row_count < CHUNK_ROWS { + break; + } + } + for state in states + .iter() + .filter(|state| state.writable && state.cursor_pending) + { + state + .destination + .db + .advance_parse_offset( + &cursor_path, + ParseOffset { + byte_offset: state.cursor.position, + mtime: state.cursor.mtime, + file_id: state.cursor.file_id, + }, + ) + .await; + } + stats.sessions_upserted = states + .iter() + .map(|state| state.sessions_seen.len() as u64) + .sum(); + Ok(stats) +} + async fn try_ingest_user_state_db( db: &GlobalDb, source: &HermesProfileSource, @@ -444,7 +668,7 @@ async fn try_ingest_user_state_db( }; let batches = build_user_batches(db, &new.items, &path_str, source, registered_roots).await; if batches.is_empty() { - db.set_parse_offset(&cursor_path, offset).await; + db.advance_parse_offset(&cursor_path, offset).await; } else { let message_count = batches .iter() @@ -553,34 +777,66 @@ async fn build_batches( state_db_path: &str, project_root: &Path, source: &HermesProfileSource, +) -> Vec { + let turn_locations = turn_project_locations(rows, project_root, source); + build_batches_with_locations( + db, + rows, + state_db_path, + project_root, + source, + &turn_locations, + None, + ) + .await +} + +async fn build_batches_with_locations( + db: &GlobalDb, + rows: &[HermesRow], + state_db_path: &str, + project_root: &Path, + source: &HermesProfileSource, + turn_locations: &HashMap, + row_indices: Option<&[usize]>, ) -> Vec { let mut order = Vec::new(); let mut by_session: HashMap = HashMap::new(); - let turn_locations = turn_project_locations(rows, project_root, source); - for row in rows { - if row.role == "session_meta" || row.role.is_empty() { - continue; - } - if row.active == 0 { - // Rewound/undone turns are soft-deleted in Hermes; surfacing - // them as live history would misrepresent the conversation. - continue; - } - let Some(location) = turn_locations.get(&row.id) else { - continue; - }; - let Some(message) = message_from_row(row, state_db_path, source, &location) else { - continue; + { + let mut add_row = |row: &HermesRow| { + if row.role == "session_meta" || row.role.is_empty() { + return; + } + if row.active == 0 { + // Rewound/undone turns are soft-deleted in Hermes; surfacing + // them as live history would misrepresent the conversation. + return; + } + let Some(location) = turn_locations.get(&row.id) else { + return; + }; + let Some(message) = message_from_row(row, state_db_path, source, &location) else { + return; + }; + let batch = by_session.entry(row.session_id.clone()).or_insert_with(|| { + order.push(row.session_id.clone()); + TranscriptBatch { + session: session_from_row(row, state_db_path, project_root, source, &location), + messages: Vec::new(), + } + }); + batch.messages.push(message); }; - let batch = by_session.entry(row.session_id.clone()).or_insert_with(|| { - order.push(row.session_id.clone()); - TranscriptBatch { - session: session_from_row(row, state_db_path, project_root, source, &location), - messages: Vec::new(), + if let Some(row_indices) = row_indices { + for &index in row_indices { + add_row(&rows[index]); } - }); - batch.messages.push(message); + } else { + for row in rows { + add_row(row); + } + } } let mut batches = Vec::with_capacity(order.len()); @@ -750,6 +1006,144 @@ fn turn_project_locations( locations } +struct DestinationTurnLocations { + by_row_id: HashMap, + row_indices: Vec, +} + +fn turn_project_locations_for_destinations( + rows: &[HermesRow], + destination_matchers: &[ProjectRootMatcher], + source: &HermesProfileSource, + destination_routes: &mut HashMap>, +) -> Vec { + let mut by_session: HashMap<&str, Vec<&HermesRow>> = HashMap::new(); + let row_indices = rows + .iter() + .enumerate() + .map(|(index, row)| (row.id, index)) + .collect::>(); + for row in rows { + by_session.entry(&row.session_id).or_default().push(row); + } + let mut locations = (0..destination_matchers.len()) + .map(|_| DestinationTurnLocations { + by_row_id: HashMap::new(), + row_indices: Vec::new(), + }) + .collect::>(); + for session_rows in by_session.into_values() { + let fallback_candidates = if let Some(pin) = source.legacy_project_pin.as_ref() { + vec![(pin.clone(), "profile_pin")] + } else { + let mut seen = BTreeSet::new(); + session_rows + .iter() + .filter_map(|row| { + let cwd = PathBuf::from(row.session_cwd.as_deref()?.trim()); + (cwd.is_absolute() && seen.insert(cwd.clone())).then_some((cwd, "session_cwd")) + }) + .collect::>() + }; + let mut fallbacks = vec![None; destination_matchers.len()]; + for (cwd, provenance) in fallback_candidates { + for destination_index in + matching_destinations(&cwd, destination_matchers, destination_routes) + { + fallbacks[destination_index].get_or_insert_with(|| HermesSessionLocation { + cwd: cwd.clone(), + provenance, + }); + } + } + let mut turn = Vec::new(); + for row in session_rows { + if row.role == "user" && !turn.is_empty() { + assign_turn_locations_for_destinations( + &turn, + destination_matchers, + &fallbacks, + &row_indices, + &mut locations, + destination_routes, + ); + turn.clear(); + } + turn.push(row); + } + assign_turn_locations_for_destinations( + &turn, + destination_matchers, + &fallbacks, + &row_indices, + &mut locations, + destination_routes, + ); + } + for destination in &mut locations { + destination.row_indices.sort_unstable(); + } + locations +} + +fn assign_turn_locations_for_destinations( + rows: &[&HermesRow], + destination_matchers: &[ProjectRootMatcher], + fallbacks: &[Option], + row_indices: &HashMap, + locations: &mut [DestinationTurnLocations], + destination_routes: &mut HashMap>, +) { + let explicit_paths = rows + .iter() + .rev() + .flat_map(|row| structured_tool_project_paths(row)) + .collect::>(); + let mut selected = vec![None; destination_matchers.len()]; + if explicit_paths.is_empty() { + selected.clone_from_slice(fallbacks); + } else { + for path in explicit_paths { + for destination_index in + matching_destinations(&path, destination_matchers, destination_routes) + { + selected[destination_index].get_or_insert_with(|| HermesSessionLocation { + cwd: path.clone(), + provenance: "tool_project_path", + }); + } + } + } + for (location, destination) in selected.into_iter().zip(locations) { + let Some(location) = location else { + continue; + }; + for row in rows { + destination.by_row_id.insert(row.id, location.clone()); + if let Some(&index) = row_indices.get(&row.id) { + destination.row_indices.push(index); + } + } + } +} + +fn matching_destinations( + path: &Path, + destination_matchers: &[ProjectRootMatcher], + destination_routes: &mut HashMap>, +) -> Vec { + if let Some(indices) = destination_routes.get(path) { + return indices.clone(); + } + let indices = destination_matchers + .iter() + .enumerate() + .filter_map(|(index, matcher)| matcher.contains(path).then_some(index)) + .collect::>(); + destination_routes.insert(path.to_path_buf(), indices.clone()); + indices +} + fn assign_turn_location( rows: &[&HermesRow], project_root: &Path, diff --git a/src/sessions/mod.rs b/src/sessions/mod.rs index 6639f8528..8b0465010 100644 --- a/src/sessions/mod.rs +++ b/src/sessions/mod.rs @@ -127,9 +127,25 @@ pub async fn ingest_user_cursor_sessions() -> TranscriptIngestStats { } pub async fn ingest_user_global_sources() -> TranscriptIngestStats { - let codex = ingest_user_codex_sessions(None).await; - let cursor = ingest_user_cursor_sessions().await; - let stats = codex.merge(cursor); + ingest_user_global_sources_for_provider(None).await +} + +fn provider_selected(scope: Option, candidate: SessionProvider) -> bool { + scope.is_none() || scope == Some(candidate) +} + +/// Keeps the profile-level session store current without touching providers +/// outside an explicitly requested message-search scope. +pub async fn ingest_user_global_sources_for_provider( + provider: Option, +) -> TranscriptIngestStats { + let mut stats = TranscriptIngestStats::default(); + if provider_selected(provider, SessionProvider::Codex) { + stats = stats.merge(ingest_user_codex_sessions(None).await); + } + if provider_selected(provider, SessionProvider::Cursor) { + stats = stats.merge(ingest_user_cursor_sessions().await); + } let Ok(profile_root) = crate::storage::default_profile_root() else { return stats; }; @@ -139,24 +155,37 @@ pub async fn ingest_user_global_sources() -> TranscriptIngestStats { let Some(roots) = try_registered_project_roots().await else { return stats; }; - let hermes = hermes::ingest_user_sessions(&db, &roots).await; - let mut stats = stats.merge(hermes); - let claude = claude::ingest_user_sessions(&db, &profile_root, None, roots.clone()).await; - stats = stats.merge(claude); + if provider_selected(provider, SessionProvider::Hermes) { + stats = stats.merge(hermes::ingest_user_sessions(&db, &roots).await); + } + if provider_selected(provider, SessionProvider::Claude) { + stats = stats + .merge(claude::ingest_user_sessions(&db, &profile_root, None, roots.clone()).await); + } let mut sources: Vec> = Vec::new(); - if let Some(source) = vibe::VibeSource::new() { + if provider_selected(provider, SessionProvider::Vibe) + && let Some(source) = vibe::VibeSource::new() + { sources.push(Box::new(source.for_user_scope(roots.clone()))); } - if let Some(source) = cline_like::ClineLikeSource::cline() { + if provider_selected(provider, SessionProvider::Cline) + && let Some(source) = cline_like::ClineLikeSource::cline() + { sources.push(Box::new(source.for_user_scope(roots.clone()))); } - if let Some(source) = cline_like::ClineLikeSource::roo_code() { + if provider_selected(provider, SessionProvider::RooCode) + && let Some(source) = cline_like::ClineLikeSource::roo_code() + { sources.push(Box::new(source.for_user_scope(roots.clone()))); } - if let Some(source) = cline_like::ClineLikeSource::kilo() { + if provider_selected(provider, SessionProvider::Kilo) + && let Some(source) = cline_like::ClineLikeSource::kilo() + { sources.push(Box::new(source.for_user_scope(roots.clone()))); } - if let Some(source) = kiro::KiroSource::new() { + if provider_selected(provider, SessionProvider::Kiro) + && let Some(source) = kiro::KiroSource::new() + { sources.push(Box::new(source.for_user_scope(roots))); } for source in sources { @@ -164,7 +193,10 @@ pub async fn ingest_user_global_sources() -> TranscriptIngestStats { stats = stats.merge(source_stats); } if stats.messages_upserted > 0 { - crate::hooks::schedule_user_session_review("all", None); + crate::hooks::schedule_user_session_review( + provider.map_or("all", SessionProvider::id), + None, + ); } stats } @@ -203,9 +235,19 @@ pub async fn ingest_global_sources_for_provider( project_root: &Path, provider: Option, ) -> TranscriptIngestStats { - // Keep the profile-level projectless history current alongside every - // project sweep. Its independent DB cursors make repeated calls no-ops. - let _ = ingest_user_global_sources().await; + let _ = ingest_user_global_sources_for_provider(provider).await; + ingest_project_sources_for_provider(db, project_root, provider, true).await +} + +/// Project-store half of catch-up. Cross-project search runs user ingestion +/// once, then calls this per destination; Hermes can be excluded because its +/// dedicated multi-destination driver scans each source database only once. +pub(crate) async fn ingest_project_sources_for_provider( + db: &GlobalDb, + project_root: &Path, + provider: Option, + include_hermes: bool, +) -> TranscriptIngestStats { let mut sources: Vec> = Vec::new(); match provider { None => { @@ -256,13 +298,21 @@ pub async fn ingest_global_sources_for_provider( } else { stats }; - let stats = if provider.is_none() || provider == Some(SessionProvider::Hermes) { - // Hermes stores many sessions in one SQLite file per profile, so it - // plugs in beside the file-based sources rather than `TranscriptSource`. - stats.merge(hermes::ingest_for_project(db, project_root).await) - } else { - stats - }; + let stats = + if include_hermes && (provider.is_none() || provider == Some(SessionProvider::Hermes)) { + // Hermes stores many sessions in one SQLite file per profile, so it + // plugs in beside the file-based sources rather than `TranscriptSource`. + stats.merge(hermes::ingest_for_project(db, project_root).await) + } else { + stats + }; + finalize_project_ingest(db, project_root).await; + stats +} + +/// Refreshes derived session data after a caller performs its own optimized +/// transcript ingest (for example, one shared Hermes source sweep). +pub(crate) async fn finalize_project_ingest(db: &GlobalDb, project_root: &Path) { // Now that messages have landed, attribute any commits that fell inside a // recorded session span. Fail-open: a git or DB hiccup never blocks ingest. attribute_commits_after_ingest(db).await; @@ -271,7 +321,6 @@ pub async fn ingest_global_sources_for_provider( // a workflow-ingest hiccup only logs at debug, never blocks session ingest. // Runs live in their own tables, so they do not affect `stats`. let _ = workflow_ingest::ingest_workflow_runs(db, project_root).await; - stats } /// Runs the bounded commit-attribution sweep against the correlation store. @@ -513,6 +562,28 @@ impl SessionMessageType { mod git_scan_tests { use super::*; + #[test] + fn provider_scoped_user_catch_up_excludes_unrelated_providers() { + assert!(provider_selected( + Some(SessionProvider::Hermes), + SessionProvider::Hermes + )); + for unrelated in [ + SessionProvider::Codex, + SessionProvider::Cursor, + SessionProvider::Claude, + SessionProvider::Vibe, + SessionProvider::Cline, + SessionProvider::RooCode, + SessionProvider::Kilo, + SessionProvider::Kiro, + ] { + assert!(!provider_selected(Some(SessionProvider::Hermes), unrelated)); + } + assert!(provider_selected(None, SessionProvider::Codex)); + assert!(provider_selected(None, SessionProvider::Hermes)); + } + #[test] fn parse_git_log_commits_reads_sha_and_time_skipping_malformed() { let stdout = concat!( diff --git a/src/tracedecay.rs b/src/tracedecay.rs index dc3213a68..968d6ce8c 100644 --- a/src/tracedecay.rs +++ b/src/tracedecay.rs @@ -41,6 +41,7 @@ pub struct TraceDecay { config: TraceDecayConfig, project_root: PathBuf, store_layout: StoreLayout, + active_graph_layout: ActiveGraphLayout, open_options: TraceDecayOpenOptions, registry: LanguageRegistry, /// The active git branch (None if detached HEAD or not a git repo). @@ -52,6 +53,12 @@ pub struct TraceDecay { read_only: bool, } +#[derive(Debug, Clone)] +struct ActiveGraphLayout { + dirty_path: PathBuf, + sync_lock_path: PathBuf, +} + #[derive(Debug, Clone, Default)] pub struct TraceDecayOpenOptions { pub profile_root: Option, diff --git a/src/tracedecay/indexing.rs b/src/tracedecay/indexing.rs index ef119173c..8d46dcfc4 100644 --- a/src/tracedecay/indexing.rs +++ b/src/tracedecay/indexing.rs @@ -13,7 +13,6 @@ use crate::resolution::ReferenceResolver; use crate::sync; use crate::types::*; -use super::locking::{clear_dirty_sentinel_at, try_acquire_sync_lock_at, write_dirty_sentinel_at}; use super::{IndexResult, SyncResult, TraceDecay, current_timestamp}; /// Convert any backslash in a *relative* project-root-relative path to a @@ -238,8 +237,8 @@ impl TraceDecay { "project root is not a directory" ); self.ensure_branch_writable("full index")?; - let _lock = try_acquire_sync_lock_at(&self.store_layout.sync_lock_path)?; - write_dirty_sentinel_at(&self.store_layout.dirty_path); + let _lock = self.try_acquire_active_sync_lock()?; + self.write_active_dirty_sentinels(); let start = Instant::now(); // 1. Clear existing data and enter bulk-load mode @@ -378,7 +377,8 @@ impl TraceDecay { result.duration_ms > 0 || result.file_count == 0, "non-empty index completed in zero milliseconds" ); - clear_dirty_sentinel_at(&self.store_layout.dirty_path); + self.db.checkpoint().await?; + self.clear_active_dirty_sentinels(); Ok(result) } @@ -419,9 +419,10 @@ impl TraceDecay { self.ensure_branch_writable("sync files")?; - let Ok(lock) = try_acquire_sync_lock_at(&self.store_layout.sync_lock_path) else { + let Ok(lock) = self.try_acquire_active_sync_lock() else { return Ok(true); }; + self.write_active_dirty_sentinels(); let result = self.sync_single_files(&stale_files).await; drop(lock); @@ -456,7 +457,7 @@ impl TraceDecay { self.ensure_branch_writable("sync files")?; - let lock = if let Ok(lock) = try_acquire_sync_lock_at(&self.store_layout.sync_lock_path) { + let lock = if let Ok(lock) = self.try_acquire_active_sync_lock() { lock } else { // Peer is syncing. Wait for them to release the lock so the @@ -471,7 +472,7 @@ impl TraceDecay { return Ok(()); } tokio::time::sleep(Duration::from_millis(50)).await; - if let Ok(lock) = try_acquire_sync_lock_at(&self.store_layout.sync_lock_path) { + if let Ok(lock) = self.try_acquire_active_sync_lock() { // Peer released. If they covered our files, the DB is // fresh and we're done; otherwise sync ourselves. let still_stale = self.check_file_staleness(&stale_files).await; @@ -483,6 +484,7 @@ impl TraceDecay { } } }; + self.write_active_dirty_sentinels(); let _ = self.sync_single_files(&stale_files).await; drop(lock); @@ -594,7 +596,8 @@ impl TraceDecay { ) .await?; - clear_dirty_sentinel_at(&self.store_layout.dirty_path); + self.db.checkpoint().await?; + self.clear_active_dirty_sentinels(); Ok(()) } @@ -778,8 +781,8 @@ impl TraceDecay { "sync: project root is not a directory" ); self.ensure_branch_writable("sync")?; - let _lock = try_acquire_sync_lock_at(&self.store_layout.sync_lock_path)?; - write_dirty_sentinel_at(&self.store_layout.dirty_path); + let _lock = self.try_acquire_active_sync_lock()?; + self.write_active_dirty_sentinels(); let start = Instant::now(); on_progress(0, 0, "scanning files"); @@ -1043,7 +1046,8 @@ impl TraceDecay { .set_metadata("last_sync_duration_ms", &duration_ms.to_string()) .await?; - clear_dirty_sentinel_at(&self.store_layout.dirty_path); + self.db.checkpoint().await?; + self.clear_active_dirty_sentinels(); Ok(SyncResult { files_added: new_files.len(), files_modified: stale.len(), diff --git a/src/tracedecay/lifecycle.rs b/src/tracedecay/lifecycle.rs index de1a0402d..2cd905fac 100644 --- a/src/tracedecay/lifecycle.rs +++ b/src/tracedecay/lifecycle.rs @@ -38,6 +38,7 @@ impl TraceDecay { save_config_to_path(&store_layout.config_path, &config)?; let (db, _migrated) = Database::initialize(&store_layout.graph_db_path).await?; + let active_graph_layout = active_graph_layout(&store_layout.graph_db_path); if store_layout.storage_mode == storage::StorageMode::ProfileSharded { storage::write_store_manifest(&store_layout)?; } @@ -57,6 +58,7 @@ impl TraceDecay { config, project_root: project_root.to_path_buf(), store_layout, + active_graph_layout, open_options, registry: LanguageRegistry::new(), active_branch, @@ -311,6 +313,11 @@ impl TraceDecay { active_branch.as_deref(), ); + // Sync state belongs to the concrete graph DB, not the repository-wide + // store root. Different tracked branches have independent databases + // and must never clear or inherit one another's dirty marker or lock. + let active_graph_layout = active_graph_layout(&db_path); + if !db_path.exists() { return Err(TraceDecayError::Config { message: format!( @@ -322,7 +329,8 @@ impl TraceDecay { // If the dirty sentinel exists, a previous sync/index was interrupted. // Check integrity and rebuild if necessary. - let crashed = has_dirty_sentinel_at(&store_layout.dirty_path); + let crashed = has_dirty_sentinel_at(&active_graph_layout.dirty_path) + || has_dirty_sentinel_at(&store_layout.dirty_path); if crashed { eprintln!( "[tracedecay] previous operation was interrupted — checking database integrity…" @@ -346,7 +354,10 @@ impl TraceDecay { // quick integrity check. if crashed { match db.quick_check().await { - Ok(true) => clear_dirty_sentinel_at(&store_layout.dirty_path), + Ok(true) => { + clear_dirty_sentinel_at(&active_graph_layout.dirty_path); + clear_dirty_sentinel_at(&store_layout.dirty_path); + } Ok(false) => { print_corruption_warning(&db_path); return Err(recovery_required_error( @@ -366,6 +377,7 @@ impl TraceDecay { config, project_root: project_root.to_path_buf(), store_layout, + active_graph_layout, open_options, registry: LanguageRegistry::new(), active_branch, @@ -411,6 +423,7 @@ impl TraceDecay { &store_layout.data_root, active_branch.as_deref(), ); + let active_graph_layout = active_graph_layout(&db_path); if !db_path.exists() { return Err(TraceDecayError::Config { @@ -427,6 +440,7 @@ impl TraceDecay { config, project_root: project_root.to_path_buf(), store_layout, + active_graph_layout, open_options, registry: LanguageRegistry::new(), active_branch, @@ -643,6 +657,7 @@ impl TraceDecay { .ok_or_else(|| TraceDecayError::Config { message: format!("branch '{branch_name}' is not tracked"), })?; + let active_graph_layout = active_graph_layout(&db_path); if !db_path.exists() { return Err(TraceDecayError::Config { @@ -659,6 +674,7 @@ impl TraceDecay { config, project_root: project_root.to_path_buf(), store_layout, + active_graph_layout, open_options, registry: LanguageRegistry::new(), active_branch: Some(branch_name.to_string()), @@ -955,6 +971,19 @@ impl TraceDecay { } } +fn graph_sidecar_path(db_path: &Path, suffix: &str) -> PathBuf { + let mut file_name = db_path.file_name().unwrap_or_default().to_os_string(); + file_name.push(suffix); + db_path.with_file_name(file_name) +} + +fn active_graph_layout(db_path: &Path) -> super::ActiveGraphLayout { + super::ActiveGraphLayout { + dirty_path: graph_sidecar_path(db_path, ".dirty"), + sync_lock_path: graph_sidecar_path(db_path, ".sync.lock"), + } +} + #[derive(Debug)] struct StoreIdentityInventory { project_id: String, diff --git a/src/tracedecay/locking.rs b/src/tracedecay/locking.rs index 1b980c46e..49c747fca 100644 --- a/src/tracedecay/locking.rs +++ b/src/tracedecay/locking.rs @@ -47,6 +47,41 @@ pub struct SyncLockGuard { path: PathBuf, } +pub(super) struct ActiveSyncLockGuard { + _active: SyncLockGuard, + _legacy: Option, +} + +impl super::TraceDecay { + pub(super) fn try_acquire_active_sync_lock(&self) -> Result { + let active = try_acquire_sync_lock_at(&self.active_graph_layout.sync_lock_path)?; + let legacy = if self.active_graph_layout.sync_lock_path == self.store_layout.sync_lock_path + { + None + } else { + Some(try_acquire_sync_lock_at(&self.store_layout.sync_lock_path)?) + }; + Ok(ActiveSyncLockGuard { + _active: active, + _legacy: legacy, + }) + } + + pub(super) fn write_active_dirty_sentinels(&self) { + write_dirty_sentinel_at(&self.active_graph_layout.dirty_path); + if self.active_graph_layout.dirty_path != self.store_layout.dirty_path { + write_dirty_sentinel_at(&self.store_layout.dirty_path); + } + } + + pub(super) fn clear_active_dirty_sentinels(&self) { + clear_dirty_sentinel_at(&self.active_graph_layout.dirty_path); + if self.active_graph_layout.dirty_path != self.store_layout.dirty_path { + clear_dirty_sentinel_at(&self.store_layout.dirty_path); + } + } +} + impl Drop for SyncLockGuard { fn drop(&mut self) { let _ = std::fs::remove_file(&self.path); diff --git a/tests/storage_suite/branch_db_safety_test.rs b/tests/storage_suite/branch_db_safety_test.rs index ac290183c..3fb58d3ca 100644 --- a/tests/storage_suite/branch_db_safety_test.rs +++ b/tests/storage_suite/branch_db_safety_test.rs @@ -74,10 +74,55 @@ async fn open_untracked_project() -> (IsolatedEnv, PathBuf, TraceDecay) { assert_eq!(feature.active_branch(), Some("feature/untracked")); assert_eq!(feature.serving_branch(), Some("feature/untracked")); assert!(!feature.is_fallback()); + let layout = feature.store_layout(); + assert_eq!(layout.dirty_path, layout.data_root.join("dirty")); + assert_eq!(layout.sync_lock_path, layout.data_root.join("sync.lock")); + assert_ne!(feature.db_path(), layout.graph_db_path); (env, project, feature) } +#[tokio::test] +// Regression: init and reopen must use the same graph-scoped lock. +async fn init_index_uses_graph_specific_sync_lock() { + let (_env, project) = IsolatedEnv::acquire().await; + fs::create_dir_all(project.join("src")).unwrap(); + fs::write(project.join("src/lib.rs"), "pub fn indexed() {}\n").unwrap(); + + let cg = TraceDecay::init(&project).await.unwrap(); + let lock_path = PathBuf::from(format!("{}.sync.lock", cg.db_path().display())); + fs::write(&lock_path, std::process::id().to_string()).unwrap(); + + let err = match cg.index_all().await { + Ok(_) => panic!("active graph lock must block init indexing"), + Err(err) => err, + }; + assert!( + err.to_string() + .contains("another sync is already in progress") + ); +} + +#[tokio::test] +async fn open_honors_and_clears_legacy_dirty_sentinel() { + let (_env, project) = IsolatedEnv::acquire().await; + fs::create_dir_all(project.join("src")).unwrap(); + fs::write(project.join("src/lib.rs"), "pub fn indexed() {}\n").unwrap(); + + let cg = TraceDecay::init(&project).await.unwrap(); + cg.index_all().await.unwrap(); + let legacy_dirty = cg.store_layout().data_root.join("dirty"); + fs::write(&legacy_dirty, "interrupted legacy writer").unwrap(); + drop(cg); + + let reopened = TraceDecay::open(&project).await.unwrap(); + assert!( + !legacy_dirty.exists(), + "successful recovery must clear legacy sentinel" + ); + drop(reopened); +} + async fn open_detached_fallback_project() -> (IsolatedEnv, PathBuf, TraceDecay) { let (env, project) = IsolatedEnv::acquire().await; diff --git a/tests/transcript_ingest_suite/hermes.rs b/tests/transcript_ingest_suite/hermes.rs index 7f4b0c714..481d4a501 100644 --- a/tests/transcript_ingest_suite/hermes.rs +++ b/tests/transcript_ingest_suite/hermes.rs @@ -8,16 +8,48 @@ use std::path::{Path, PathBuf}; use serde_json::json; use tempfile::TempDir; -use tracedecay::global_db::GlobalDb; +use tracedecay::global_db::{GlobalDb, ParseOffset}; use tracedecay::sessions::SessionRecord; use tracedecay::sessions::cursor::open_project_session_db; -use tracedecay::sessions::hermes::{ingest_for_project, ingest_homes, ingest_user_homes}; +use tracedecay::sessions::hermes::{ + ProjectIngestDestination, ingest_for_project, ingest_homes, ingest_homes_for_projects, + ingest_user_homes, +}; use tracedecay::sessions::lcm::LcmPreflightRequest; use crate::support::{assert_metadata_path_eq, create_git_repo_with_linked_worktree}; const SESSION_ID: &str = "20260101_000000_abc123"; +#[tokio::test] +async fn hermes_row_cursor_cannot_regress_during_overlapping_sweeps() { + let tmp = TempDir::new().unwrap(); + let db = GlobalDb::open_at(&tmp.path().join("sessions.db")) + .await + .unwrap(); + let cursor = "state.db#turn-project-v2"; + db.advance_parse_offset( + cursor, + ParseOffset { + byte_offset: 200, + mtime: 20, + file_id: 0, + }, + ) + .await; + db.advance_parse_offset( + cursor, + ParseOffset { + byte_offset: 100, + mtime: 10, + file_id: 0, + }, + ) + .await; + + assert_eq!(db.get_parse_offset(cursor).await.unwrap().byte_offset, 200); +} + #[tokio::test] // Intentional: this test changes HOME/USERPROFILE/HERMES_HOME while storage // discovery is running, so it must share the profile-environment lock used by @@ -462,6 +494,126 @@ async fn hermes_ingest_is_incremental_and_idempotent() { assert_eq!(session.ended_at, Some(1_780_629_340)); } +#[tokio::test] +async fn hermes_shared_sweep_routes_one_source_to_multiple_project_stores() { + let tmp = TempDir::new().unwrap(); + let (hermes_home, first_project) = setup(&tmp); + let second_project = tmp.path().join("second-project"); + crate::support::init_project_at(&second_project); + let state_db = write_hermes_profile(&hermes_home, "test", None).await; + let conn = open_state_db(&state_db).await; + conn.execute( + "UPDATE sessions SET cwd = ?1 WHERE id = ?2", + libsql::params![first_project.to_string_lossy().as_ref(), SESSION_ID], + ) + .await + .unwrap(); + let second_session = "20260101_000100_def456"; + conn.execute( + "INSERT INTO sessions (id, source, model, started_at, cwd, title) + VALUES (?1, 'telegram', 'gpt-5.5', 1780629500.0, ?2, 'Second project')", + libsql::params![second_session, second_project.to_string_lossy().as_ref()], + ) + .await + .unwrap(); + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) + VALUES (?1, 'user', 'Route this message to the second project', 1780629501.0)", + libsql::params![second_session], + ) + .await + .unwrap(); + + let first_db = open_project_session_db(&first_project).await.unwrap(); + let second_db = open_project_session_db(&second_project).await.unwrap(); + let destinations = [ + ProjectIngestDestination { + db: &first_db, + project_root: &first_project, + }, + ProjectIngestDestination { + db: &second_db, + project_root: &second_project, + }, + ]; + let stats = ingest_homes_for_projects(std::slice::from_ref(&hermes_home), &destinations).await; + + assert_eq!(stats.messages_upserted, 5); + assert!(first_db.get_session("hermes", SESSION_ID).await.is_some()); + assert!( + first_db + .get_session("hermes", second_session) + .await + .is_none() + ); + assert!(second_db.get_session("hermes", SESSION_ID).await.is_none()); + assert!( + second_db + .get_session("hermes", second_session) + .await + .is_some() + ); + assert_eq!( + ingest_homes_for_projects(std::slice::from_ref(&hermes_home), &destinations) + .await + .messages_upserted, + 0 + ); +} + +#[tokio::test] +#[ignore = "manual cold-history benchmark; requires TRACEDECAY_HERMES_BENCH_HOME and TRACEDECAY_HERMES_BENCH_PROJECT"] +async fn hermes_shared_sweep_cold_history_completes_under_sixty_seconds() { + let hermes_home = PathBuf::from( + std::env::var("TRACEDECAY_HERMES_BENCH_HOME").expect("Hermes home for benchmark"), + ); + let project_root = PathBuf::from( + std::env::var("TRACEDECAY_HERMES_BENCH_PROJECT").expect("project root for benchmark"), + ); + let output_root = PathBuf::from( + std::env::var("TRACEDECAY_HERMES_BENCH_OUTPUT").expect("fast output root for benchmark"), + ); + std::fs::create_dir_all(&output_root).unwrap(); + let temp = tempfile::Builder::new() + .prefix("hermes-cold-catchup-") + .tempdir_in(output_root) + .unwrap(); + let mut project_roots = vec![project_root]; + for index in 1..31 { + let root = temp.path().join(format!("project-{index}")); + crate::support::init_project_at(&root); + project_roots.push(root); + } + let mut dbs = Vec::with_capacity(project_roots.len()); + for index in 0..project_roots.len() { + dbs.push( + GlobalDb::open_at_without_structured_backfill( + &temp.path().join(format!("sessions-{index}.db")), + ) + .await + .unwrap(), + ); + } + let destinations = dbs + .iter() + .zip(&project_roots) + .map(|(db, project_root)| ProjectIngestDestination { db, project_root }) + .collect::>(); + + let started = std::time::Instant::now(); + let stats = ingest_homes_for_projects(std::slice::from_ref(&hermes_home), &destinations).await; + let elapsed = started.elapsed(); + + assert!( + stats.messages_upserted > 0, + "benchmark must ingest real history" + ); + assert!( + elapsed < std::time::Duration::from_secs(60), + "cold Hermes catch-up took {elapsed:?}" + ); +} + #[tokio::test] async fn hermes_profile_pinned_elsewhere_is_not_ingested() { let tmp = TempDir::new().unwrap();