diff --git a/.changeset/user-message-search-scope.md b/.changeset/user-message-search-scope.md new file mode 100644 index 000000000..b2a4300ee --- /dev/null +++ b/.changeset/user-message-search-scope.md @@ -0,0 +1,5 @@ +--- +"tracedecay": patch +--- + +Fix user-session search and startup ingestion, cross-provider session attribution, final Codex turn ingestion, legacy Hermes profile migration, dashboard session ambiguity, and daemon shutdown of automation process trees. diff --git a/.config/nextest.toml b/.config/nextest.toml index fce484c8f..b2ddb93a0 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -22,7 +22,6 @@ filter = '(binary(=daemon_suite) & test(/^git_watch_test::(fifty_commit_rebase_n slow-timeout = { period = "60s", terminate-after = 10 } [test-groups] -windows-dashboard-server = { max-threads = 32 } windows-tracedecay-init = { max-threads = 32 } windows-init-heavy = { max-threads = 32 } windows-profile-storage = { max-threads = 32 } @@ -34,8 +33,19 @@ windows-timing-sensitive = { max-threads = 32 } [[profile.ci.overrides]] filter = 'binary(=dashboard_api_test)' -platform = { host = 'cfg(windows)' } -test-group = 'windows-dashboard-server' +threads-required = "num-cpus" + +[[profile.default.overrides]] +filter = 'binary(=dashboard_api_test)' +threads-required = "num-cpus" + +[[profile.ci.overrides]] +filter = 'binary(=core_cli_suite) & test(/^tool_daemon_test::(daemon_socket_is_owner_only|daemon_sigterm_exits_while_project_client_is_connected)$/)' +threads-required = "num-cpus" + +[[profile.default.overrides]] +filter = 'binary(=core_cli_suite) & test(/^tool_daemon_test::(daemon_socket_is_owner_only|daemon_sigterm_exits_while_project_client_is_connected)$/)' +threads-required = "num-cpus" [[profile.ci.overrides]] filter = 'binary(=agent_suite) | (binary(=storage_suite) & test(/^branch_db_safety_test::/)) | (binary(=core_cli_suite) & test(/^cli_non_interactive_test::/)) | (binary(=mcp_suite) & test(/^(mcp_cli_serve_test|serve_template_path_test)::/))' diff --git a/src/agents/codex.rs b/src/agents/codex.rs index e69c9a861..ff487509e 100644 --- a/src/agents/codex.rs +++ b/src/agents/codex.rs @@ -716,6 +716,12 @@ const CODEX_MANAGED_HOOKS: &[CodexManagedHook] = &[ timeout_secs: 120, matcher: Some("auto|manual"), }, + CodexManagedHook { + event: "Stop", + subcommand: "hook-codex-stop", + timeout_secs: 5, + matcher: None, + }, ]; /// Subcommands from older bundles that uninstall must also strip even though diff --git a/src/agents/codex/tests.rs b/src/agents/codex/tests.rs index 097777bab..04955e3e3 100644 --- a/src/agents/codex/tests.rs +++ b/src/agents/codex/tests.rs @@ -173,6 +173,7 @@ fn codex_hook_trust_state_reports_missing_entries() { CodexHookTrustState::Missing(vec![ "post_compact".to_string(), "session_start".to_string(), + "stop".to_string(), "subagent_start".to_string(), "user_prompt_submit".to_string(), ]) @@ -230,6 +231,9 @@ trusted_hash = "sha256:subagent" [hooks.state."tracedecay@local-repo:hooks/hooks.json:post_compact:0:0"] trusted_hash = "sha256:compact" + +[hooks.state."tracedecay@local-repo:hooks/hooks.json:stop:0:0"] +trusted_hash = "sha256:stop" "#, ) .unwrap(); @@ -240,6 +244,7 @@ trusted_hash = "sha256:compact" "post_compact".to_string(), "post_tool_use".to_string(), "session_start".to_string(), + "stop".to_string(), "subagent_start".to_string(), "user_prompt_submit".to_string(), ]) diff --git a/src/automation/jobs.rs b/src/automation/jobs.rs index 856e2e34b..255c5fa27 100644 --- a/src/automation/jobs.rs +++ b/src/automation/jobs.rs @@ -815,6 +815,9 @@ async fn run_pre_run_command(command: &str, project_root: Option<&Path>) -> Resu if let Some(project_root) = project_root { process.current_dir(project_root); } + // Scheduler shutdown aborts the owning future. Ensure a pre-run command + // does not outlive that future and keep the daemon cgroup alive. + process.kill_on_drop(true); let output = tokio::time::timeout( Duration::from_secs(JOB_COMMAND_TIMEOUT_SECS), process.output(), diff --git a/src/cli.rs b/src/cli.rs index a15529829..928548bb4 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -254,6 +254,9 @@ pub enum Commands { /// Codex PostCompact hook handler for app-server LCM summaries (called by Codex) #[command(name = "hook-codex-post-compact", hide = true)] HookCodexPostCompact, + /// Codex Stop hook handler for final-turn user-session ingestion (called by Codex) + #[command(name = "hook-codex-stop", hide = true)] + HookCodexStop, /// Hermes terminal receipt handler (called by the TraceDecay plugin) #[command(name = "hook-hermes-terminal-receipt", hide = true)] HookHermesTerminalReceipt, diff --git a/src/commands.rs b/src/commands.rs index c6fd18718..9270820a6 100644 --- a/src/commands.rs +++ b/src/commands.rs @@ -494,42 +494,84 @@ pub(crate) async fn handle_migrate_action(action: MigrateAction) -> tracedecay:: &prefixes, tracedecay::migrate::registry::StaleRootScope::CanonicalRootMissing, ); - let deleted = if apply { + let profile_root = tracedecay::config::user_data_dir(); + let mut stale_storage_projects = Vec::new(); + for project_path in global_db.list_project_paths().await { + let path = Path::new(&project_path); + if !prefixes.is_empty() && !prefixes.iter().any(|prefix| path.starts_with(prefix)) { + continue; + } + let location = global::classify_project_storage_with_registry( + path, + Some(&global_db), + profile_root.as_deref(), + ) + .await; + if location.status == global::ProjectStorageStatus::Stale { + stale_storage_projects.push(project_path); + } + } + let (deleted_code_projects, deleted_storage_projects) = if apply { let project_ids: Vec = stale .iter() .map(|project| project.project_id.clone()) .collect(); - global_db.delete_code_projects(&project_ids).await + ( + global_db.delete_code_projects(&project_ids).await, + global_db.delete_projects(&stale_storage_projects).await, + ) } else { - 0 + (0, 0) }; + let candidate_paths = stale + .iter() + .map(|project| { + tracedecay::global_db::GlobalDb::canonical_project_key(Path::new( + &project.canonical_root, + )) + }) + .chain(stale_storage_projects.iter().map(|path| { + tracedecay::global_db::GlobalDb::canonical_project_key(Path::new(path)) + })) + .collect::>(); + let candidate_count = candidate_paths.len(); + let metadata_candidate_count = stale.len() + stale_storage_projects.len(); + let deleted_count = deleted_code_projects + deleted_storage_projects; if json { println!( "{}", serde_json::to_string_pretty(&serde_json::json!({ "apply": apply, "prefix": prefix, - "candidate_count": stale.len(), - "deleted_count": deleted, + "candidate_count": candidate_count, + "metadata_candidate_count": metadata_candidate_count, + "code_project_candidate_count": stale.len(), + "storage_project_candidate_count": stale_storage_projects.len(), + "deleted_count": deleted_count, + "deleted_code_project_count": deleted_code_projects, + "deleted_storage_project_count": deleted_storage_projects, "candidates": stale, + "storage_project_candidates": stale_storage_projects, }))? ); } else { println!( "registry-gc: {} stale project(s){}", - stale.len(), + candidate_count, if apply { " selected" } else { " found" } ); if apply { - println!("deleted: {deleted}"); + println!( + "metadata rows deleted: {deleted_count} ({deleted_code_projects} identity, {deleted_storage_projects} storage)" + ); } else { println!("dry run: re-run with --apply to delete registry metadata"); } - for project in stale.iter().take(20) { - println!("{} {}", project.project_id, project.canonical_root); + for project_path in candidate_paths.iter().take(20) { + println!("{project_path}"); } - if stale.len() > 20 { - println!("... {} more", stale.len() - 20); + if candidate_count > 20 { + println!("... {} more", candidate_count - 20); } } } diff --git a/src/config.rs b/src/config.rs index 9b29d4ad9..522c5051f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -982,14 +982,17 @@ pub fn lock_user_data_dir_test_env() -> std::sync::MutexGuard<'static, ()> { .unwrap_or_else(std::sync::PoisonError::into_inner) } -/// Pins [`USER_DATA_DIR_ENV`] to an isolated temp profile while holding -/// [`USER_DATA_DIR_TEST_LOCK`], so parallel lib tests cannot race profile -/// resolution during `TraceDecay::init` / indexing. +/// Pins [`USER_DATA_DIR_ENV`] and agent home discovery to an isolated temp +/// profile while holding [`USER_DATA_DIR_TEST_LOCK`], so parallel lib tests +/// cannot race profile resolution or scan live host transcripts during +/// `TraceDecay::init` / indexing. #[cfg(test)] pub struct PinnedUserDataDir { _lock: std::sync::MutexGuard<'static, ()>, _root: tempfile::TempDir, previous: Option, + previous_home: Option, + previous_userprofile: Option, } #[cfg(test)] @@ -1002,13 +1005,19 @@ impl PinnedUserDataDir { fs::create_dir_all(&profile) .unwrap_or_else(|err| panic!("failed to create isolated profile root: {err}")); let previous = std::env::var_os(USER_DATA_DIR_ENV); + let previous_home = std::env::var_os("HOME"); + let previous_userprofile = std::env::var_os("USERPROFILE"); unsafe { std::env::set_var(USER_DATA_DIR_ENV, &profile); + std::env::set_var("HOME", root.path()); + std::env::set_var("USERPROFILE", root.path()); } Self { _lock: lock, _root: root, previous, + previous_home, + previous_userprofile, } } } @@ -1028,6 +1037,14 @@ impl Drop for PinnedUserDataDir { Some(previous) => std::env::set_var(USER_DATA_DIR_ENV, previous), None => std::env::remove_var(USER_DATA_DIR_ENV), } + match self.previous_home.take() { + Some(previous) => std::env::set_var("HOME", previous), + None => std::env::remove_var("HOME"), + } + match self.previous_userprofile.take() { + Some(previous) => std::env::set_var("USERPROFILE", previous), + None => std::env::remove_var("USERPROFILE"), + } } } } diff --git a/src/daemon.rs b/src/daemon.rs index d98331646..822113854 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -43,6 +43,8 @@ const HOOK_EVENT_NOTIFY_TIMEOUT: Duration = Duration::from_millis(750); const DAEMON_SHUTDOWN_DEADLINE: Duration = Duration::from_secs(45); #[cfg(unix)] const DAEMON_CLIENT_DRAIN_DEADLINE: Duration = Duration::from_secs(15); +#[cfg(unix)] +const DAEMON_TASK_ABORT_DEADLINE: Duration = Duration::from_secs(2); #[derive(Clone, Default)] pub(crate) struct DaemonLifecycle { @@ -1534,22 +1536,40 @@ async fn run_foreground_unix(socket_path: PathBuf) -> Result<()> { client_tasks.spawn(async move { Box::pin(serve_socket_client(stream, engine)).await }); } engine.lifecycle.begin_draining(); - log_daemon_event( - "daemon_shutdown", - &[("socket", socket_path.display().to_string())], - ); // Stop accepting and unlink the socket before draining so clients that // connect during shutdown get NotFound/ConnectionRefused (which they retry // via `connect_with_restart_grace`) instead of a queued connection that // will never be served. drop(listener); let _ = std::fs::remove_file(&socket_path); - let clients_drained = drain_client_tasks(&mut client_tasks, DAEMON_CLIENT_DRAIN_DEADLINE).await; - engine.lifecycle.wait_for_idle().await; + // Keep auxiliary process creation blocked until every scheduler and client + // task is drained or abandoned. A killed app-server call may retry before + // unwinding, so a shorter guard leaves a shutdown-time respawn race. + let _codex_shutdown = crate::sessions::codex_app_server::begin_codex_app_server_shutdown(); + // Stop automation before announcing shutdown or waiting for clients. + // Scheduler tasks may be inside a synchronous auxiliary-agent call, so + // shutdown also terminates their tracked process trees before joining. + engine.shutdown_automation_schedulers().await; + log_daemon_event( + "daemon_shutdown", + &[("socket", socket_path.display().to_string())], + ); + let in_flight_drained = timeout( + DAEMON_CLIENT_DRAIN_DEADLINE, + engine.lifecycle.wait_for_idle(), + ) + .await + .is_ok(); + // Once admitted requests are finished (or their bound elapsed), every + // remaining client task is an idle socket reader or already-cancelled + // request wrapper. Abort those immediately instead of making shutdown wait + // for clients to close persistent connections themselves. + client_tasks.abort_all(); + let clients_drained = drain_client_tasks(&mut client_tasks, DAEMON_TASK_ABORT_DEADLINE).await; // Client setup and in-flight requests may create schedulers or project // servers. Sweep owned background tasks only after all client work drains. engine.shutdown_background_tasks().await; - if !clients_drained { + if !in_flight_drained || !clients_drained { log_daemon_event( "daemon_shutdown", &[ @@ -1618,9 +1638,12 @@ async fn drain_client_tasks(clients: &mut JoinSet>, deadline: Duratio } clients.abort_all(); - while let Some(completed) = clients.join_next().await { - log_client_task_result(completed); - } + let _ = timeout(DAEMON_TASK_ABORT_DEADLINE, async { + while let Some(completed) = clients.join_next().await { + log_client_task_result(completed); + } + }) + .await; false } @@ -1905,6 +1928,9 @@ impl DaemonEngine { project_path: PathBuf, handshake: DaemonHandshake, ) { + if !self.lifecycle.accepting() { + return; + } { let schedulers = self.automation_schedulers.lock().await; if schedulers.contains_key(&key) { @@ -1976,8 +2002,11 @@ impl DaemonEngine { project_path: PathBuf, handshake: DaemonHandshake, ) { + if !self.lifecycle.accepting() { + return; + } let mut schedulers = self.automation_schedulers.lock().await; - if schedulers.contains_key(&key) { + if !self.lifecycle.accepting() || schedulers.contains_key(&key) { return; } let wake = Arc::new(tokio::sync::Notify::new()); @@ -1994,14 +2023,7 @@ impl DaemonEngine { } async fn shutdown_background_tasks(&self) { - let scheduler_handles: Vec> = { - let mut schedulers = self.automation_schedulers.lock().await; - schedulers.drain().map(|(_, handle)| handle.task).collect() - }; - for handle in scheduler_handles { - handle.abort(); - let _ = handle.await; - } + self.shutdown_automation_schedulers().await; self.git_watcher.shutdown().await; if let Some(handle) = self.pr_autotrack_task.lock().await.take() { @@ -2010,6 +2032,23 @@ impl DaemonEngine { } } + async fn shutdown_automation_schedulers(&self) { + let scheduler_handles: Vec> = { + let mut schedulers = self.automation_schedulers.lock().await; + schedulers.drain().map(|(_, handle)| handle.task).collect() + }; + let _child_shutdown = crate::sessions::codex_app_server::begin_codex_app_server_shutdown(); + for handle in &scheduler_handles { + handle.abort(); + } + let _ = timeout(DAEMON_TASK_ABORT_DEADLINE, async { + for handle in scheduler_handles { + let _ = handle.await; + } + }) + .await; + } + async fn shutdown_servers(&self) { let servers: Vec> = { let servers = self.project_servers.lock().await; diff --git a/src/daemon/tests.rs b/src/daemon/tests.rs index 1652a2eb7..21f1f17fe 100644 --- a/src/daemon/tests.rs +++ b/src/daemon/tests.rs @@ -12,7 +12,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; use tokio::task::JoinHandle; #[cfg(unix)] -use super::drain_client_tasks; +use super::{AutomationSchedulerHandle, DaemonEngine, ProjectServerKey, drain_client_tasks}; use super::{DaemonClientIdentity, DaemonHandshake, DaemonLifecycle}; #[test] @@ -90,6 +90,35 @@ async fn draining_waits_for_one_bounded_in_flight_request() { assert!(lifecycle.try_enter().is_none()); } +#[cfg(unix)] +#[tokio::test] +async fn daemon_scheduler_shutdown_aborts_and_joins_every_loop() { + let engine = DaemonEngine::default(); + let key = ProjectServerKey { + project_path: PathBuf::from("/projects/shutdown-test"), + scope_prefix: None, + client_identity: test_client_identity(), + }; + let task = tokio::spawn(std::future::pending::<()>()); + engine.automation_schedulers.lock().await.insert( + key, + AutomationSchedulerHandle { + task, + wake: std::sync::Arc::new(tokio::sync::Notify::new()), + }, + ); + + engine.lifecycle.begin_draining(); + tokio::time::timeout( + tokio::time::Duration::from_secs(1), + engine.shutdown_automation_schedulers(), + ) + .await + .expect("scheduler shutdown should not wait for its tick interval"); + + assert!(engine.automation_schedulers.lock().await.is_empty()); +} + fn test_client_identity() -> DaemonClientIdentity { test_client_identity_for(PathBuf::from("/profiles/client")) } diff --git a/src/dashboard/lcm_service.rs b/src/dashboard/lcm_service.rs index af388670b..c3f96a1d9 100644 --- a/src/dashboard/lcm_service.rs +++ b/src/dashboard/lcm_service.rs @@ -89,6 +89,29 @@ pub(crate) fn parse_epoch(value: &str) -> Option { trimmed.parse::().ok() } +async fn session_provider_count(conn: &libsql::Connection, session_id: &str) -> i64 { + super::util::query_i64( + conn, + "SELECT COUNT(*) + FROM ( + SELECT provider FROM lcm_raw_messages WHERE session_id = ?1 + UNION + SELECT provider FROM lcm_summary_nodes WHERE session_id = ?1 + )", + libsql::params![session_id.to_string()], + ) + .await +} + +fn ambiguous_session_error(session_id: &str) -> LcmErrorResponse { + ( + StatusCode::CONFLICT, + Json(http_detail(&format!( + "ambiguous session id across providers: {session_id}" + ))), + ) +} + pub(crate) async fn ensure_valid_summary_metadata( state: &DashboardState, conn: &libsql::Connection, @@ -454,6 +477,10 @@ pub(crate) async fn session_payload( }; ensure_valid_summary_metadata(state, conn, "session").await?; + if session_provider_count(conn, session_id).await > 1 { + return Err(ambiguous_session_error(session_id)); + } + let message_count = super::util::query_i64( conn, "SELECT COUNT(*) FROM lcm_raw_messages WHERE session_id = ?1", @@ -507,6 +534,9 @@ pub(crate) async fn session_payload( "session summary nodes", lcm_queries::session_summary_nodes(conn, session_id, limit + 1, offset).await, )?; + if session_provider_count(conn, session_id).await > 1 { + return Err(ambiguous_session_error(session_id)); + } let has_more_summary_nodes = summary_nodes.len() as i64 > limit; summary_nodes.truncate(limit as usize); let has_more = has_more_messages || has_more_summary_nodes; diff --git a/src/doctor.rs b/src/doctor.rs index b1b2b5055..6217b0d60 100644 --- a/src/doctor.rs +++ b/src/doctor.rs @@ -459,7 +459,9 @@ async fn check_stale_stores(dc: &mut DoctorCounters) { if !std::io::stdin().is_terminal() { dc.warnings += 1; - dc.info(" Re-run `tracedecay doctor` interactively to purge them."); + dc.info( + " Run `tracedecay migrate registry-gc --json` to preview, then add `--apply` to purge metadata only.", + ); return; } diff --git a/src/hook_cmd.rs b/src/hook_cmd.rs index 89c74950f..42d60fc87 100644 --- a/src/hook_cmd.rs +++ b/src/hook_cmd.rs @@ -78,6 +78,7 @@ pub(crate) fn hook_input(command: &Commands) -> Option { | Commands::HookCodexSubagentStart | Commands::HookCodexPostToolUse | Commands::HookCodexPostCompact + | Commands::HookCodexStop | Commands::HookHermesTerminalReceipt | Commands::HookUserSessionReview => Some(HookInput::Stdin), _ => None, @@ -161,6 +162,9 @@ pub(crate) async fn handle_hook_command(command: Commands) -> tracedecay::errors Commands::HookCodexPostCompact => { exit_if_nonzero(tracedecay::hooks::hook_codex_post_compact().await); } + Commands::HookCodexStop => { + exit_if_nonzero(tracedecay::hooks::hook_codex_stop().await); + } Commands::HookHermesTerminalReceipt => { exit_if_nonzero(tracedecay::hooks::hook_hermes_terminal_receipt().await); } @@ -216,6 +220,7 @@ mod tests { (Commands::HookCodexSubagentStart, HookInput::Stdin), (Commands::HookCodexPostToolUse, HookInput::Stdin), (Commands::HookCodexPostCompact, HookInput::Stdin), + (Commands::HookCodexStop, HookInput::Stdin), (Commands::HookHermesTerminalReceipt, HookInput::Stdin), (Commands::HookUserSessionReview, HookInput::Stdin), ] @@ -224,7 +229,7 @@ mod tests { #[test] fn all_hook_commands_have_explicit_input_semantics() { let hooks = hook_commands(); - assert_eq!(hooks.len(), 26); + assert_eq!(hooks.len(), 27); assert_eq!( hooks .iter() @@ -237,7 +242,7 @@ mod tests { .iter() .filter(|(_, input)| *input == HookInput::Stdin) .count(), - 25 + 26 ); for (command, expected) in hooks { assert_eq!(hook_input(&command), Some(expected)); diff --git a/src/hooks/codex.rs b/src/hooks/codex.rs index 17a25c28e..1b69a1383 100644 --- a/src/hooks/codex.rs +++ b/src/hooks/codex.rs @@ -96,8 +96,11 @@ pub async fn hook_codex_user_prompt_submit() -> i32 { .ok() .as_ref() .and_then(event_session_id); - if root.is_none() && ingest_user_codex_session(session_id.clone()).await { - super::schedule_user_session_review("codex", session_id.as_deref()); + if root.is_none() { + // Keep recall current, but wait for the native Stop receipt before + // reflection so one completed turn schedules one review rather than a + // prompt-only review followed immediately by a final-turn review. + let _ = ingest_user_codex_session(session_id).await; } let context = Box::pin(codex_user_prompt_submit_context_for_event(&event)).await; println!( @@ -305,6 +308,42 @@ pub async fn hook_codex_post_compact() -> i32 { 0 } +const CODEX_STOP_INGEST_BUDGET: Duration = Duration::from_secs(3); + +/// Codex `Stop` hook handler. +/// +/// Codex emits this after the assistant finishes a turn. Projectless sessions +/// need this terminal receipt because the prompt hook runs before the final +/// assistant message has been appended to the rollout. +pub async fn hook_codex_stop() -> i32 { + let event = read_hook_event!(); + let parsed = serde_json::from_str::(&event).unwrap_or(Value::Null); + let root = codex_project_root_from_parsed_event_with_identity(&parsed).await; + let _hook_telemetry = record_hook_invoked(root.as_deref(), HintAgent::Codex, "Stop", &event); + let session_id = event_session_id(&parsed); + let ingested = tokio::time::timeout( + CODEX_STOP_INGEST_BUDGET, + finalize_codex_user_session(root.as_deref(), session_id.clone()), + ) + .await + .unwrap_or(false); + if ingested { + super::schedule_user_session_review("codex", session_id.as_deref()); + } + println!("{}", serde_json::json!({})); + 0 +} + +async fn finalize_codex_user_session( + project_root: Option<&Path>, + session_id: Option, +) -> bool { + if project_root.is_some() { + return false; + } + ingest_user_codex_session(session_id).await +} + /// Builds a Codex hook stdout payload with `additionalContext`. pub fn codex_additional_context_json(event_name: &str, additional_context: &str) -> String { serde_json::json!({ @@ -928,6 +967,72 @@ mod tests { } } + #[tokio::test] + #[allow(clippy::await_holding_lock)] + async fn codex_stop_ingests_final_user_turn_once() { + let _lock = crate::hooks::lock_test_env(); + let temp = tempfile::tempdir().unwrap(); + let home = temp.path().join("home"); + let profile = temp.path().join("profile"); + let general = temp.path().join("general-chat"); + std::fs::create_dir_all(&general).unwrap(); + let _home_env = EnvGuard::set_path("HOME", &home); + let _profile_env = EnvGuard::set_path(USER_DATA_DIR_ENV, &profile); + let rollout_dir = home.join(".codex/sessions/2026/01/01"); + std::fs::create_dir_all(&rollout_dir).unwrap(); + let rollout = rollout_dir.join("rollout-2026-01-01T00-00-00-final-turn.jsonl"); + let records = [ + serde_json::json!({ + "timestamp": "2026-01-01T00:00:00.000Z", + "type": "session_meta", + "payload": { + "id": "final-turn", + "cwd": general.to_string_lossy(), + "model": "gpt-5.5" + } + }), + serde_json::json!({ + "timestamp": "2026-01-01T00:00:01.000Z", + "type": "event_msg", + "payload": {"type": "user_message", "message": "one turn prompt"} + }), + serde_json::json!({ + "timestamp": "2026-01-01T00:00:02.000Z", + "type": "event_msg", + "payload": {"type": "agent_message", "message": "final assistant evidence"} + }), + ]; + std::fs::write( + rollout, + records + .iter() + .map(ToString::to_string) + .collect::>() + .join("\n") + + "\n", + ) + .unwrap(); + + assert!(finalize_codex_user_session(None, Some("final-turn".to_string())).await); + assert!( + !finalize_codex_user_session(None, Some("final-turn".to_string())).await, + "a repeated Stop receipt must not schedule another review" + ); + assert!( + !finalize_codex_user_session(Some(&general), Some("final-turn".to_string())).await, + "project-scoped Stop receipts must never write the user session store" + ); + + let db = crate::sessions::open_user_session_db(&profile) + .await + .expect("user session db should exist"); + let hits = db + .search_session_messages("codex", Some("user"), "final assistant evidence", 10) + .await; + assert_eq!(hits.len(), 1); + assert_eq!(hits[0].session.session_id, "final-turn"); + } + #[test] fn codex_prompt_hints_dedupe_by_session_and_category() { let _lock = crate::hooks::lock_test_env(); diff --git a/src/hooks/mod.rs b/src/hooks/mod.rs index 55d87306c..aa720d632 100644 --- a/src/hooks/mod.rs +++ b/src/hooks/mod.rs @@ -30,7 +30,7 @@ pub use codex::{ codex_additional_context_json, codex_apply_patch_rel_paths, codex_project_root_from_event, codex_subagent_start_log_line, codex_user_prompt_submit_context_for_event, codex_workspace_status_from_event, evaluate_codex_subagent_start, hook_codex_post_compact, - hook_codex_post_tool_use, hook_codex_session_start, hook_codex_subagent_start, + hook_codex_post_tool_use, hook_codex_session_start, hook_codex_stop, hook_codex_subagent_start, hook_codex_user_prompt_submit, record_codex_subagent_start, }; pub use cursor::{ diff --git a/src/hooks/tool_hints/tests.rs b/src/hooks/tool_hints/tests.rs index 9d0ede284..c9a8fa9d0 100644 --- a/src/hooks/tool_hints/tests.rs +++ b/src/hooks/tool_hints/tests.rs @@ -171,8 +171,10 @@ fn every_hint_route_names_a_registered_agent_tool() { for spec in CATEGORY_SPECS { for tool in spec.expected_tools { + let unavailable_optional_tool = + *tool == "tracedecay_ast_grep_rewrite" && !crate::mcp::tools::ast_grep_available(); assert!( - registered.contains(*tool), + registered.contains(*tool) || unavailable_optional_tool, "hint category {} routes agents to unregistered tool {tool}", spec.key ); diff --git a/src/lifecycle_lease.rs b/src/lifecycle_lease.rs index f69c2008f..9463fec95 100644 --- a/src/lifecycle_lease.rs +++ b/src/lifecycle_lease.rs @@ -59,6 +59,10 @@ impl Drop for LifecycleLease { unregister_process_token(token); } if let LeaseHold::File(file) = &self.hold { + #[cfg(windows)] + if self.exclusive { + remove_owner_sidecar_if_current(&self.lock_path, self.token.as_deref()); + } let _ = fs2::FileExt::unlock(file); } } @@ -127,8 +131,8 @@ fn acquire_shared_or_inherited_at(path: &Path, operation: &str) -> Result { - let owner = read_owner(&mut file); + Err(error) if is_lock_contended(&error) => { + let owner = read_owner(&mut file, path); let owner_token = owner.as_deref().and_then(|line| line.split('\t').next()); if owner_token.is_some_and(process_owns_token) { Ok(LifecycleLease { @@ -154,8 +158,8 @@ fn try_acquire_shared_or_inherited_at(path: &Path, operation: &str) -> Result { - let owner = read_owner(&mut file); + Err(error) if is_lock_contended(&error) => { + let owner = read_owner(&mut file, path); let owner_token = owner.as_deref().and_then(|line| line.split('\t').next()); if owner_token.is_some_and(process_owns_token) { Ok(SharedLeaseAttempt::Acquired(LifecycleLease { @@ -193,8 +197,8 @@ fn acquire_exclusive_or_inherited_at( let mut file = open_lock_file(path)?; match fs2::FileExt::try_lock_exclusive(&file) { Ok(()) => own_exclusive(file, path, operation), - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { - let owner = read_owner(&mut file); + Err(error) if is_lock_contended(&error) => { + let owner = read_owner(&mut file, path); if let Some(token) = inherited.filter(|token| { owner.as_deref().and_then(|line| line.split('\t').next()) == Some(token.as_str()) }) { @@ -241,9 +245,9 @@ fn acquire_exclusive_at(path: &Path, operation: &str) -> Result let file = open_lock_file(path)?; match fs2::FileExt::try_lock_exclusive(&file) { Ok(()) => own_exclusive(file, path, operation), - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + Err(error) if is_lock_contended(&error) => { let mut file = file; - let owner = read_owner(&mut file); + let owner = read_owner(&mut file, path); Err(busy_error(operation, owner.as_deref())) } Err(error) => Err(lock_error(path, operation, &error)), @@ -259,8 +263,8 @@ fn acquire_shared_at(path: &Path, operation: &str) -> Result { lock_path: path.to_path_buf(), exclusive: false, }), - Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { - let owner = read_owner(&mut file); + Err(error) if is_lock_contended(&error) => { + let owner = read_owner(&mut file, path); Err(busy_error(operation, owner.as_deref())) } Err(error) => Err(lock_error(path, operation, &error)), @@ -276,21 +280,22 @@ fn try_acquire_shared_at(path: &Path, operation: &str) -> Result { - Ok(SharedLeaseAttempt::Busy) - } + Err(error) if is_lock_contended(&error) => Ok(SharedLeaseAttempt::Busy), Err(error) => Err(lock_error(path, operation, &error)), } } fn own_exclusive(mut file: File, path: &Path, operation: &str) -> Result { let token = lease_token(); + let owner = format!("{token}\t{operation}\t{}\n", std::process::id()); file.set_len(0).map_err(|error| owner_write_error(&error))?; file.seek(SeekFrom::Start(0)) .map_err(|error| owner_write_error(&error))?; - writeln!(file, "{token}\t{operation}\t{}", std::process::id()) + file.write_all(owner.as_bytes()) .map_err(|error| owner_write_error(&error))?; file.flush().map_err(|error| owner_write_error(&error))?; + #[cfg(windows)] + std::fs::write(owner_sidecar_path(path), owner).map_err(|error| owner_write_error(&error))?; register_process_token(&token); Ok(LifecycleLease { hold: LeaseHold::File(file), @@ -340,7 +345,28 @@ fn open_lock_file(path: &Path) -> Result { .map_err(|error| lock_error(path, "open", &error)) } -fn read_owner(file: &mut File) -> Option { +fn is_lock_contended(error: &std::io::Error) -> bool { + if error.kind() == std::io::ErrorKind::WouldBlock { + return true; + } + #[cfg(windows)] + { + // LockFileEx reports lock contention as ERROR_LOCK_VIOLATION, which + // std currently classifies as Uncategorized rather than WouldBlock. + return error.raw_os_error() == Some(33); + } + #[cfg(not(windows))] + false +} + +fn read_owner(file: &mut File, _path: &Path) -> Option { + #[cfg(windows)] + if let Ok(owner) = std::fs::read_to_string(owner_sidecar_path(_path)) { + let owner = owner.trim(); + if !owner.is_empty() { + return Some(owner.to_string()); + } + } let mut owner = String::new(); file.seek(SeekFrom::Start(0)).ok()?; file.read_to_string(&mut owner).ok()?; @@ -348,6 +374,26 @@ fn read_owner(file: &mut File) -> Option { (!owner.is_empty()).then(|| owner.to_string()) } +#[cfg(windows)] +fn owner_sidecar_path(path: &Path) -> PathBuf { + path.with_extension("lock.owner") +} + +#[cfg(windows)] +fn remove_owner_sidecar_if_current(path: &Path, token: Option<&str>) { + let Some(token) = token else { + return; + }; + let owner_path = owner_sidecar_path(path); + let is_current = std::fs::read_to_string(&owner_path) + .ok() + .and_then(|owner| owner.split('\t').next().map(str::to_string)) + .is_some_and(|owner_token| owner_token == token); + if is_current { + let _ = std::fs::remove_file(owner_path); + } +} + fn lease_token() -> String { let epoch_nanos = SystemTime::now() .duration_since(UNIX_EPOCH) diff --git a/src/main.rs b/src/main.rs index d3997f6ba..e04b97e29 100644 --- a/src/main.rs +++ b/src/main.rs @@ -517,6 +517,7 @@ async fn dispatch_command(command: Commands) -> tracedecay::errors::Result<()> { | Commands::HookCodexSubagentStart | Commands::HookCodexPostToolUse | Commands::HookCodexPostCompact + | Commands::HookCodexStop | Commands::HookHermesTerminalReceipt | Commands::HookUserSessionReview) => { hook_cmd::handle_hook_command(hook_command).await?; diff --git a/src/mcp/server.rs b/src/mcp/server.rs index 36732addc..9ec8db7af 100644 --- a/src/mcp/server.rs +++ b/src/mcp/server.rs @@ -1394,7 +1394,11 @@ impl McpServer { tokio::spawn(async move { let _ = tokio::time::timeout(std::time::Duration::from_secs(20), async move { if let Some(db) = GlobalDb::open_at(&session_db_path).await { - let _ = crate::sessions::ingest_global_sources(&db, &project_root).await; + let _ = crate::sessions::ingest_global_sources_for_startup( + &db, + &project_root, + ) + .await; // Historical git-span correlation is only ever written by // live hook events (which never fire for stdio/daemonless // deployments) or a manual CLI backfill. Neither runs for diff --git a/src/mcp/server/freshness_tests.rs b/src/mcp/server/freshness_tests.rs index 16b04c702..a22481850 100644 --- a/src/mcp/server/freshness_tests.rs +++ b/src/mcp/server/freshness_tests.rs @@ -150,6 +150,9 @@ async fn startup_catch_up_spawned_once_per_server() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn ledger_writes_settled_is_bounded_when_a_write_wedges() { let (cg, _dir, _pin) = init_indexed_repo().await; + let mut config = crate::config::load_config(cg.project_root()).expect("load config"); + config.sync.session_start_sync = false; + crate::config::save_config(cg.project_root(), &config).expect("disable unrelated catch-up"); let server = McpServer::new(cg, None).await; // Inject a never-completing observed ledger write via the same accounting diff --git a/src/mcp/tools/definitions.rs b/src/mcp/tools/definitions.rs index 8b7beefb0..deac4fe13 100644 --- a/src/mcp/tools/definitions.rs +++ b/src/mcp/tools/definitions.rs @@ -366,10 +366,10 @@ pub fn get_tool_definitions() -> Vec { } fn add_lcm_storage_scope_property(definitions: &mut [ToolDefinition]) { - for definition in definitions - .iter_mut() - .filter(|definition| definition.name.starts_with("tracedecay_lcm_")) - { + for definition in definitions.iter_mut().filter(|definition| { + definition.name.starts_with("tracedecay_lcm_") + || definition.name == "tracedecay_message_search" + }) { let Some(properties) = definition .input_schema .get_mut("properties") diff --git a/src/mcp/tools/handlers/mod.rs b/src/mcp/tools/handlers/mod.rs index bf04c5973..d510ac628 100644 --- a/src/mcp/tools/handlers/mod.rs +++ b/src/mcp/tools/handlers/mod.rs @@ -59,10 +59,10 @@ pub async fn handle_user_lcm_tool( .to_string(), }); } - let sessions_db_path = crate::sessions::user_sessions_db_path(profile_root); if tool_name == "tracedecay_message_search" { - return session::handle_user_message_search(&sessions_db_path, args).await; + return session::handle_user_message_search(profile_root, args).await; } + let sessions_db_path = crate::sessions::user_sessions_db_path(profile_root); let context = session::LcmHandlerContext::user(&sessions_db_path); match tool_name { "tracedecay_lcm_status" => session::handle_lcm_status(context, args).await, @@ -308,7 +308,7 @@ pub async fn handle_tool_call_with_registry_and_implicit_project( } } if let Some(storage_scope) = args.get("storage_scope").and_then(Value::as_str) { - if !tool_name.starts_with("tracedecay_lcm_") { + if !tool_name.starts_with("tracedecay_lcm_") && tool_name != "tracedecay_message_search" { return Err(TraceDecayError::Config { message: format!("unknown parameter `storage_scope` for `{tool_name}`"), }); diff --git a/src/mcp/tools/handlers/session.rs b/src/mcp/tools/handlers/session.rs index f774c92be..940eb8e30 100644 --- a/src/mcp/tools/handlers/session.rs +++ b/src/mcp/tools/handlers/session.rs @@ -2527,11 +2527,36 @@ pub(super) async fn handle_message_search( } pub(super) async fn handle_user_message_search( - sessions_db_path: &Path, + profile_root: &Path, args: Value, ) -> Result { let request = parse_message_search_request(&args)?; - let Some(db) = GlobalDb::open_read_only_at(sessions_db_path).await else { + let sessions_db_path = crate::sessions::user_sessions_db_path(profile_root); + let catch_up_leader = if request.catch_up { + let key = format!( + "user:{}:{}", + sessions_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 + }; + if catch_up_leader.is_some() { + let _ = crate::sessions::ingest_user_global_sources_for_provider_at( + profile_root, + request.provider_scope.provider(), + ) + .await; + } + drop(catch_up_leader); + let Some(db) = GlobalDb::open_read_only_at(&sessions_db_path).await else { return Ok(tool_json( None, &args, @@ -2549,7 +2574,7 @@ pub(super) async fn handle_user_message_search( } else { search_session_messages_in_db(&db, &request).await }; - let payload = message_search_payload(&request, &results, false); + let payload = message_search_payload(&request, &results, request.catch_up); Ok(tool_json_with_md(None, &args, &payload, || { render_message_search_md(&payload) })) diff --git a/src/migrate/consolidate/mod.rs b/src/migrate/consolidate/mod.rs index 47e798e8c..2b7b49dca 100644 --- a/src/migrate/consolidate/mod.rs +++ b/src/migrate/consolidate/mod.rs @@ -1686,5 +1686,5 @@ fn io_error(error: io::Error) -> TraceDecayError { config_error(error.to_string()) } -#[cfg(test)] +#[cfg(all(test, not(windows)))] mod tests; diff --git a/src/migrate/hermes.rs b/src/migrate/hermes.rs index e0e4d1d1b..bb2b2b1f4 100644 --- a/src/migrate/hermes.rs +++ b/src/migrate/hermes.rs @@ -113,10 +113,34 @@ async fn migrate_legacy_hermes_stores_inner( .await { Ok(CandidateOutcome::Migrated(migration, preserved_memory)) => { + if let Err(reason) = remove_legacy_registry_metadata( + tracedecay_profile_root, + candidate.legacy_registry_project_id.as_deref(), + &candidate.profile_dir, + ) + .await + { + report.failed.push(LegacyHermesMigrationIssue { + source_db: source_db.clone(), + reason, + }); + } report.migrated.push(migration); report.unresolved.extend(preserved_memory); } Ok(CandidateOutcome::AlreadyMigrated(migration, preserved_memory)) => { + if let Err(reason) = remove_legacy_registry_metadata( + tracedecay_profile_root, + candidate.legacy_registry_project_id.as_deref(), + &candidate.profile_dir, + ) + .await + { + report.failed.push(LegacyHermesMigrationIssue { + source_db: source_db.clone(), + reason, + }); + } report.already_migrated.push(migration); report.unresolved.extend(preserved_memory); } @@ -173,11 +197,46 @@ async fn migrate_legacy_hermes_stores_inner( report } +async fn remove_legacy_registry_metadata( + tracedecay_profile_root: &Path, + project_id: Option<&str>, + expected_legacy_root: &Path, +) -> Result<(), String> { + let Some(project_id) = project_id else { + return Ok(()); + }; + let registry_path = tracedecay_profile_root.join("global.db"); + let registry = GlobalDb::open_at(®istry_path).await.ok_or_else(|| { + format!( + "could not open project registry '{}'", + registry_path.display() + ) + })?; + let Some(project) = registry.get_code_project(project_id).await else { + return Ok(()); + }; + if !same_path(Path::new(&project.canonical_root), expected_legacy_root) + && !same_path(Path::new(&project.display_root), expected_legacy_root) + { + return Ok(()); + } + registry + .delete_code_projects(&[project_id.to_string()]) + .await; + if registry.get_code_project(project_id).await.is_some() { + return Err(format!( + "migrated legacy sessions, but could not remove legacy Hermes registry metadata for '{project_id}'; source stores were preserved" + )); + } + Ok(()) +} + struct LegacyStoreCandidate { profile_dir: PathBuf, source_db: PathBuf, source_sessions_db: Option, source_memory_db: Option, + legacy_registry_project_id: Option, } impl LegacyStoreCandidate { @@ -205,6 +264,7 @@ fn legacy_store_candidates( }, source_sessions_db: sessions_db.is_file().then_some(sessions_db), source_memory_db: memory_db.is_file().then_some(memory_db), + legacy_registry_project_id: None, }) }) .collect::>(); @@ -241,6 +301,7 @@ fn legacy_store_candidates( }, source_sessions_db: sessions_db.is_file().then_some(sessions_db), source_memory_db: memory_db.is_file().then_some(memory_db), + legacy_registry_project_id: manifest.project_id, }); } } @@ -837,10 +898,10 @@ async fn resolve_target_project( if is_projectless_candidate(&candidate, user_home, hermes_homes) { continue; } - let Some(target) = + let resolved = resolve_project_candidate(&candidate, user_home, hermes_homes, registry.as_ref()) - .await? - else { + .await?; + let Some(target) = resolved else { row_has_unresolved_project_evidence = true; continue; }; @@ -916,10 +977,12 @@ fn is_projectless_candidate(candidate: &Path, user_home: &Path, hermes_homes: &[ if candidate.as_os_str().is_empty() || candidate == Path::new("user") { return true; } - same_path(candidate, user_home) - || hermes_homes - .iter() - .any(|hermes_home| same_path(candidate, hermes_home)) + if same_path(candidate, user_home) { + return true; + } + hermes_homes + .iter() + .any(|hermes_home| same_path(candidate, hermes_home)) } fn collect_metadata_project_candidates( @@ -1014,13 +1077,13 @@ fn real_project_root( let canonical_user_home = user_home .canonicalize() .unwrap_or_else(|_| user_home.to_path_buf()); - let is_hermes_owned = hermes_homes.iter().any(|hermes_home| { + let is_hermes_home = hermes_homes.iter().any(|hermes_home| { let canonical_hermes_home = hermes_home .canonicalize() .unwrap_or_else(|_| hermes_home.clone()); - canonical.starts_with(canonical_hermes_home) + canonical == canonical_hermes_home }); - if canonical == canonical_user_home || is_hermes_owned { + if canonical == canonical_user_home || is_hermes_home { return None; } if let Some(git_root) = crate::worktree::git_worktree_root(&canonical) { @@ -2165,6 +2228,29 @@ mod tests { fs::write(project.join(".tracedecay/tracedecay.db"), []).unwrap(); } + #[tokio::test] + async fn registry_cleanup_preserves_reassigned_project_identity() { + let temp = tempfile::tempdir().unwrap(); + let profile_root = temp.path().join("profile"); + let legacy_root = temp.path().join("home/.hermes"); + let corrected_root = temp.path().join("projects/hermes-agent"); + fs::create_dir_all(&legacy_root).unwrap(); + fs::create_dir_all(&corrected_root).unwrap(); + let registry = GlobalDb::open_at(&profile_root.join("global.db")) + .await + .unwrap(); + registry + .upsert_code_project("reassigned", &corrected_root, None, None, None) + .await + .unwrap(); + + remove_legacy_registry_metadata(&profile_root, Some("reassigned"), &legacy_root) + .await + .unwrap(); + + assert!(registry.get_code_project("reassigned").await.is_some()); + } + async fn seed_source(path: &Path, sessions: &[(&str, &Path)]) { let db = GlobalDb::open_at(path).await.expect("open source"); for (ordinal, (session_id, project)) in sessions.iter().enumerate() { @@ -3049,6 +3135,13 @@ mod tests { serde_json::to_vec_pretty(&manifest).unwrap(), ) .unwrap(); + let registry = GlobalDb::open_at(&profile_root.join("global.db")) + .await + .unwrap(); + registry + .upsert_code_project("legacy-hermes-identity", &hermes, None, None, None) + .await + .unwrap(); seed_source(&source, &[("session", &project)]).await; let report = migrate_legacy_hermes_stores_to(&user_home, &profile_root).await; @@ -3062,6 +3155,75 @@ mod tests { .await .unwrap(); assert_eq!(count(target.conn(), "sessions").await, 1); + assert!(source.is_file()); + assert!( + registry + .get_code_project("legacy-hermes-identity") + .await + .is_none() + ); + } + + #[tokio::test] + async fn migrates_hermes_owned_profile_shard_sessions_to_user_and_cleans_registry() { + let temp = tempfile::tempdir().unwrap(); + let user_home = temp.path().join("home"); + let profile_root = temp.path().join("tracedecay-profile"); + let hermes = user_home.join(".hermes"); + let legacy_shard = profile_root.join("projects/legacy-hermes-projectless"); + let source = legacy_shard.join(crate::storage::SESSIONS_DB_FILENAME); + fs::create_dir_all(&legacy_shard).unwrap(); + let manifest = crate::storage::StoreManifest { + schema_version: crate::storage::STORE_MANIFEST_SCHEMA_VERSION, + project_id: Some("legacy-hermes-projectless".into()), + store_kind: crate::storage::StoreKind::CodeProject, + storage_mode: crate::storage::StorageMode::ProfileSharded, + project_root: hermes.clone(), + data_root: legacy_shard.clone(), + graph_db_relpath: PathBuf::from("tracedecay.db"), + sessions_db_relpath: PathBuf::from(crate::storage::SESSIONS_DB_FILENAME), + branch_meta_relpath: PathBuf::from(crate::storage::BRANCH_META_FILENAME), + }; + fs::write( + legacy_shard.join(crate::storage::STORE_MANIFEST_FILENAME), + serde_json::to_vec_pretty(&manifest).unwrap(), + ) + .unwrap(); + let registry = GlobalDb::open_at(&profile_root.join("global.db")) + .await + .unwrap(); + registry + .upsert_code_project("legacy-hermes-projectless", &hermes, None, None, None) + .await + .unwrap(); + seed_source(&source, &[("session", &hermes)]).await; + + let report = migrate_legacy_hermes_stores_to(&user_home, &profile_root).await; + assert_eq!(report.migrated.len(), 1, "{report:?}"); + assert!(report.failed.is_empty(), "{report:?}"); + assert_eq!(report.migrated[0].source_db, source); + assert_eq!(report.migrated[0].target_project, Path::new("user")); + let target_path = crate::sessions::user_sessions_db_path(&profile_root); + let target = GlobalDb::open_read_only_at(&target_path).await.unwrap(); + let session = target.get_session("hermes", "session").await.unwrap(); + assert_eq!(session.project_key, "user"); + assert_eq!(session.project_path, "user"); + let source_after = GlobalDb::open_read_only_at(&source).await.unwrap(); + for table in ["sessions", "session_messages"] { + assert_eq!( + count(source_after.conn(), table).await, + count(target.conn(), table).await, + "row parity for {table}" + ); + } + assert_eq!(marker_count(&target_path), 1); + assert!(source.is_file()); + assert!( + registry + .get_code_project("legacy-hermes-projectless") + .await + .is_none() + ); } #[tokio::test] @@ -3221,7 +3383,7 @@ mod tests { } #[tokio::test] - async fn vanished_project_evidence_is_preserved_not_misrouted_to_user() { + async fn vanished_hermes_owned_path_is_preserved_as_unresolved() { let temp = tempfile::tempdir().unwrap(); let user_home = temp.path().join("home"); let profile_root = temp.path().join("tracedecay-profile"); @@ -3231,13 +3393,34 @@ mod tests { seed_source(&source, &[("session", &vanished_project)]).await; let report = migrate_legacy_hermes_stores_to(&user_home, &profile_root).await; + assert!(report.migrated.is_empty(), "{report:?}"); assert_eq!(report.unresolved.len(), 1, "{report:?}"); - assert!(report.migrated.is_empty()); assert!(!crate::sessions::user_sessions_db_path(&profile_root).exists()); let source_after = GlobalDb::open_read_only_at(&source).await.unwrap(); assert_eq!(count(source_after.conn(), "sessions").await, 1); } + #[tokio::test] + async fn durable_project_under_hermes_home_remains_project_scoped() { + let temp = tempfile::tempdir().unwrap(); + let user_home = temp.path().join("home"); + let profile_root = temp.path().join("tracedecay-profile"); + let project = user_home.join(".hermes/workspaces/real-project"); + mark_real_project(&project); + let source = user_home.join(".hermes/.tracedecay/sessions.db"); + fs::create_dir_all(source.parent().unwrap()).unwrap(); + seed_source(&source, &[("session", &project)]).await; + + let report = migrate_legacy_hermes_stores_to(&user_home, &profile_root).await; + assert_eq!(report.migrated.len(), 1, "{report:?}"); + assert!(report.unresolved.is_empty(), "{report:?}"); + assert_eq!( + report.migrated[0].target_project, + project.canonicalize().unwrap() + ); + assert!(!crate::sessions::user_sessions_db_path(&profile_root).exists()); + } + #[tokio::test] async fn existing_unregistered_directory_is_not_assumed_projectless() { let temp = tempfile::tempdir().unwrap(); diff --git a/src/migrate/registry.rs b/src/migrate/registry.rs index 59db27b1d..2640d3394 100644 --- a/src/migrate/registry.rs +++ b/src/migrate/registry.rs @@ -1140,5 +1140,5 @@ fn is_safe_relpath(path: &Path) -> bool { } fn path_string(path: &Path) -> String { - path.to_string_lossy().to_string() + path.to_string_lossy().replace('\\', "/") } diff --git a/src/sessions/codex.rs b/src/sessions/codex.rs index 768c31698..e2d3250d7 100644 --- a/src/sessions/codex.rs +++ b/src/sessions/codex.rs @@ -43,7 +43,7 @@ //! double-count the conversation. Goal context blocks are cataloged as compact //! `goal_context` rows because real rollouts often record them only in //! `response_item` form. This append-only JSONL is read with the shared -//! byte-offset machinery and scoped to the current project by `session_meta.cwd`. +//! byte-offset machinery and scoped per turn by the latest Codex cwd context. mod context; mod events; @@ -158,22 +158,15 @@ impl TranscriptSource for CodexSource { project_root: &Path, max_new_bytes: Option, ) -> Option { - // `session_meta` (line 1) is authoritative for cwd + session id; without - // it we cannot safely attribute the rollout to a project, so skip. + // `session_meta` (line 1) is authoritative for session identity and the + // initial cwd. Later context records can move one rollout between scopes. let meta = session_meta(path)?; - if let Some(scope) = &self.user_scope { - if scope - .session_id - .as_deref() - .is_some_and(|session_id| session_id != meta.session_id) - || scope - .registered_roots - .iter() - .any(|root| path_belongs_to_project(&meta.cwd, root)) - { - return None; - } - } else if !path_belongs_to_project(&meta.cwd, project_root) { + if self + .user_scope + .as_ref() + .and_then(|scope| scope.session_id.as_deref()) + .is_some_and(|session_id| session_id != meta.session_id) + { return None; } @@ -192,14 +185,50 @@ impl TranscriptSource for CodexSource { } else { CodexContextState::from_meta(&meta) }; + let mut last_in_scope_cwd = None; + let mut last_in_scope_git = None; for line in &new.lines { + let is_context_record = context_state.observe_context_record(&line.value, path, &meta); + let in_scope = self.user_scope.as_ref().map_or_else( + || { + context_state + .cwd + .as_deref() + .is_some_and(|cwd| path_belongs_to_project(cwd, project_root)) + }, + |scope| { + context_state.cwd.as_deref().is_none_or(|cwd| { + !scope + .registered_roots + .iter() + .any(|root| path_belongs_to_project(cwd, root)) + }) + }, + ); + if !in_scope { + if compacted_summary_from_line( + &line.value, + &meta, + context_state.model.as_deref(), + path, + line.offset, + context_state.compaction_depth + 1, + ) + .is_some() + { + context_state.compaction_depth += 1; + } + continue; + } + last_in_scope_cwd.clone_from(&context_state.cwd); + last_in_scope_git.clone_from(&context_state.git); // Non-consuming: harvest session-level policy/effort/rate-limit // summary before the line is routed to its owning handler below. structured.observe_summary(&line.value); - if turn_usage.observe(&line.value) { + if is_context_record { continue; } - if context_state.observe_context_record(&line.value, path, &meta) { + if turn_usage.observe(&line.value) { continue; } if let Some(rows) = structured.event_from_line( @@ -332,16 +361,12 @@ impl TranscriptSource for CodexSource { for mut message in structured.flush_pending(&meta, path) { context::annotate_message( &mut message, - context_state.cwd.as_deref(), - context_state.git.as_ref(), + last_in_scope_cwd.as_deref(), + last_in_scope_git.as_ref(), ); messages.push(message); } - if let Some(scope) = &self.user_scope { - messages.retain(|message| user_scope_message_allowed(message, &scope.registered_roots)); - } - let project = self.user_scope.as_ref().map_or_else( || project_root.to_string_lossy().to_string(), |_| "user".to_string(), @@ -372,27 +397,6 @@ impl TranscriptSource for CodexSource { } } -fn user_scope_message_allowed( - message: &SessionMessageRecord, - registered_roots: &[PathBuf], -) -> bool { - let turn_cwd = message - .metadata_json - .as_deref() - .and_then(|metadata| serde_json::from_str::(metadata).ok()) - .and_then(|metadata| { - metadata - .get("codex_turn_cwd") - .and_then(Value::as_str) - .map(PathBuf::from) - }); - turn_cwd.is_none_or(|cwd| { - !registered_roots - .iter() - .any(|root| path_belongs_to_project(&cwd, root)) - }) -} - /// Read the leading `session_meta` line of a rollout for cwd/session-id/model. fn session_meta(path: &Path) -> Option { let file = std::fs::File::open(path).ok()?; diff --git a/src/sessions/codex_app_server.rs b/src/sessions/codex_app_server.rs index 17e7c16b0..ec0a8e97d 100644 --- a/src/sessions/codex_app_server.rs +++ b/src/sessions/codex_app_server.rs @@ -1,13 +1,17 @@ //! Codex app-server adapter used to generate auxiliary compaction summaries. +use std::collections::HashSet; use std::fmt::Write as _; use std::io::{BufRead, BufReader, ErrorKind, Write as IoWrite}; #[cfg(windows)] use std::path::Path; use std::process::{Child, Command, Stdio}; -use std::sync::mpsc; +use std::sync::{Mutex, OnceLock, mpsc}; use std::time::{Duration, Instant}; +#[cfg(unix)] +use std::os::unix::process::CommandExt; + use serde_json::{Value, json}; use crate::errors::{Result, TraceDecayError}; @@ -17,6 +21,43 @@ pub const CODEX_SUMMARY_CHILD_ENV: &str = "TRACEDECAY_CODEX_SUMMARY_CHILD"; const CODEX_APP_SERVER_SPAWN_RETRY_WINDOW: Duration = Duration::from_millis(250); const CODEX_APP_SERVER_SPAWN_RETRY_SLEEP: Duration = Duration::from_millis(10); +#[derive(Default)] +struct ActiveCodexChildren { + process_groups: HashSet, + shutdown_guards: usize, +} + +static ACTIVE_CODEX_CHILDREN: OnceLock> = OnceLock::new(); + +fn active_codex_children() -> &'static Mutex { + ACTIVE_CODEX_CHILDREN.get_or_init(|| Mutex::new(ActiveCodexChildren::default())) +} + +pub(crate) struct CodexAppServerShutdownGuard; + +pub(crate) fn begin_codex_app_server_shutdown() -> CodexAppServerShutdownGuard { + let process_groups = { + let mut active = active_codex_children() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + active.shutdown_guards += 1; + active.process_groups.iter().copied().collect::>() + }; + for process_group in process_groups { + terminate_process_tree(process_group); + } + CodexAppServerShutdownGuard +} + +impl Drop for CodexAppServerShutdownGuard { + fn drop(&mut self) { + let mut active = active_codex_children() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + active.shutdown_guards = active.shutdown_guards.saturating_sub(1); + } +} + #[derive(Debug, Clone)] pub struct CodexAppServerSummaryConfig { pub codex_bin: String, @@ -181,9 +222,26 @@ pub fn run_prompt_with_codex_app_server( } fn spawn_codex_app_server(command: &mut Command, codex_bin: &str) -> Result { + #[cfg(unix)] + command.process_group(0); let deadline = Instant::now() + CODEX_APP_SERVER_SPAWN_RETRY_WINDOW; loop { - match command.spawn() { + let spawn_result = { + let mut active = active_codex_children() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if active.shutdown_guards > 0 { + return Err(TraceDecayError::Config { + message: "codex app-server shutdown is in progress".to_string(), + }); + } + let child = command.spawn(); + if let Ok(child) = &child { + active.process_groups.insert(child.id()); + } + child + }; + match spawn_result { Ok(child) => return Ok(child), Err(err) if err.kind() == ErrorKind::ExecutableFileBusy && Instant::now() < deadline => @@ -241,23 +299,23 @@ struct ChildGuard { impl Drop for ChildGuard { fn drop(&mut self) { - kill_child_process_tree(&mut self.child); + let process_group = self.child.id(); + terminate_process_tree(process_group); let _ = self.child.kill(); let _ = self.child.wait(); + active_codex_children() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .process_groups + .remove(&process_group); } } #[cfg(windows)] -fn kill_child_process_tree(child: &mut Child) { - // `cmd /C` shims wait for their grandchildren, so a child that already - // exited has no live process tree left; skip the taskkill spawn that - // would fail anyway. - if matches!(child.try_wait(), Ok(Some(_))) { - return; - } +fn terminate_process_tree(process_group: u32) { let _ = Command::new("taskkill") .arg("/PID") - .arg(child.id().to_string()) + .arg(process_group.to_string()) .arg("/T") .arg("/F") .stdout(Stdio::null()) @@ -265,8 +323,19 @@ fn kill_child_process_tree(child: &mut Child) { .status(); } -#[cfg(not(windows))] -fn kill_child_process_tree(_child: &mut Child) {} +#[cfg(unix)] +fn terminate_process_tree(process_group: u32) { + const SIGKILL: i32 = 9; + unsafe extern "C" { + fn kill(pid: i32, signal: i32) -> i32; + } + // The app-server is started as its own process-group leader, so signaling + // the negative pid also terminates node/codex descendants. + let _ = unsafe { kill(-(process_group as i32), SIGKILL) }; +} + +#[cfg(not(any(unix, windows)))] +fn terminate_process_tree(_process_group: u32) {} fn send_json(stdin: &mut impl IoWrite, value: &Value) -> Result<()> { writeln!(stdin, "{value}")?; @@ -559,4 +628,56 @@ mod tests { assert_eq!(params["threadSource"], json!("tracedecay_codex_summary")); assert_eq!(params["model"], json!("gpt-5.5-codex")); } + + #[cfg(unix)] + #[test] + fn shutdown_guard_terminates_active_child_and_rejects_new_spawns() { + unsafe extern "C" { + fn kill(pid: i32, signal: i32) -> i32; + } + let temp = tempfile::tempdir().unwrap(); + let descendant_pid_path = temp.path().join("descendant.pid"); + let mut command = Command::new("sh"); + command + .args(["-c", "sleep 30 & echo $! > \"$1\"; wait", "sh"]) + .arg(&descendant_pid_path); + let child = spawn_codex_app_server(&mut command, "sh").expect("spawn child"); + let mut child = ChildGuard { child }; + let deadline = Instant::now() + Duration::from_secs(1); + while !descendant_pid_path.is_file() && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + let descendant_pid: i32 = std::fs::read_to_string(&descendant_pid_path) + .expect("descendant pid file") + .trim() + .parse() + .expect("descendant pid"); + + let shutdown = begin_codex_app_server_shutdown(); + let deadline = Instant::now() + Duration::from_secs(1); + while !matches!(child.child.try_wait(), Ok(Some(_))) && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + matches!(child.child.try_wait(), Ok(Some(_))), + "active child should exit during shutdown" + ); + let descendant_deadline = Instant::now() + Duration::from_secs(1); + while unsafe { kill(descendant_pid, 0) } == 0 && Instant::now() < descendant_deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert_ne!( + unsafe { kill(descendant_pid, 0) }, + 0, + "app-server descendant should exit during shutdown" + ); + + let mut blocked = Command::new("sh"); + blocked.args(["-c", "exit 0"]); + let err = spawn_codex_app_server(&mut blocked, "sh") + .expect_err("new app-server spawns must fail during shutdown"); + assert!(err.to_string().contains("shutdown is in progress")); + + drop(shutdown); + } } diff --git a/src/sessions/cursor.rs b/src/sessions/cursor.rs index 9cc489fac..7b30581cd 100644 --- a/src/sessions/cursor.rs +++ b/src/sessions/cursor.rs @@ -359,13 +359,29 @@ pub async fn ingest_cursor_user_transcript_event_capped_with_registered_roots( else { return CursorTranscriptIngestStats::default(); }; - let registered_slugs = registered_roots - .iter() - .filter_map(|root| cursor_project_slug(root)) - .collect::>(); - if cursor_transcript_project_slug(&transcript_path) - .is_some_and(|slug| registered_slugs.contains(slug)) - { + let event_workspaces = cursor_event_workspace_roots(&event); + let belongs_to_registered_project = if event_workspaces.is_empty() { + // Without event workspace identity, Cursor's transcript directory is + // the only attribution available. Its slash-to-hyphen encoding is + // lossy, so a registered-slug collision must fail closed rather than + // risk copying project evidence into user memory. + cursor_transcript_project_slug(&transcript_path).is_some_and(|slug| { + registered_roots + .iter() + .filter_map(|root| cursor_project_slug(root)) + .any(|registered_slug| registered_slug == slug) + }) + } else { + // A hook-provided cwd/file/workspace root is stronger than the lossy + // transcript slug. This keeps distinct slash-vs-hyphen workspaces, + // linked worktrees, and renamed checkouts from excluding one another. + event_workspaces.iter().any(|workspace| { + registered_roots + .iter() + .any(|registered| paths_equal(workspace, registered)) + }) + }; + if belongs_to_registered_project { return CursorTranscriptIngestStats::default(); } let placeholder = transcript_path @@ -384,6 +400,37 @@ pub async fn ingest_cursor_user_transcript_event_capped_with_registered_roots( } } +fn cursor_event_workspace_roots(event: &Value) -> Vec { + let candidates = if let Some(cwd) = event_cwd(event) { + vec![cwd] + } else if let Some(file_path) = event + .get("file_path") + .and_then(Value::as_str) + .filter(|path| !path.is_empty()) + { + let path = Path::new(file_path); + vec![path.parent().unwrap_or(path).to_path_buf()] + } else { + event + .get("workspace_roots") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(Value::as_str) + .filter(|path| !path.is_empty()) + .map(PathBuf::from) + .collect() + }; + let mut roots: Vec = Vec::new(); + for candidate in candidates { + let root = crate::config::discover_project_root(&candidate).unwrap_or(candidate); + if !roots.iter().any(|seen| paths_equal(seen, &root)) { + roots.push(root); + } + } + roots +} + fn cursor_transcript_project_slug(path: &Path) -> Option<&str> { let components = path.components().collect::>(); let transcripts = components diff --git a/src/sessions/mod.rs b/src/sessions/mod.rs index 8b0465010..4c643d12c 100644 --- a/src/sessions/mod.rs +++ b/src/sessions/mod.rs @@ -54,6 +54,15 @@ pub async fn registered_project_roots() -> Vec { /// is projectless. pub async fn try_registered_project_roots() -> Option> { let global = GlobalDb::open().await?; + registered_project_roots_from(&global).await +} + +async fn try_registered_project_roots_at(profile_root: &Path) -> Option> { + let global = GlobalDb::open_at(&profile_root.join("global.db")).await?; + registered_project_roots_from(&global).await +} + +async fn registered_project_roots_from(global: &GlobalDb) -> Option> { let mut roots = global .list_project_paths() .await @@ -72,27 +81,42 @@ pub async fn ingest_user_codex_sessions(session_id: Option) -> Transcrip let Ok(profile_root) = crate::storage::default_profile_root() else { return TranscriptIngestStats::default(); }; - let Some(db) = open_user_session_db(&profile_root).await else { + let Some(registered_roots) = try_registered_project_roots().await else { return TranscriptIngestStats::default(); }; - let Some(source) = codex::CodexSource::new() else { + ingest_user_codex_sessions_at(&profile_root, session_id, registered_roots).await +} + +async fn ingest_user_codex_sessions_at( + profile_root: &Path, + session_id: Option, + registered_roots: Vec, +) -> TranscriptIngestStats { + let Some(db) = open_user_session_db(profile_root).await else { return TranscriptIngestStats::default(); }; - let Some(registered_roots) = try_registered_project_roots().await else { + let Some(source) = codex::CodexSource::new() else { return TranscriptIngestStats::default(); }; let source = source.for_user_scope(session_id, registered_roots); - ingest_source(&db, &source, &profile_root, None).await + ingest_source(&db, &source, profile_root, None).await } pub async fn ingest_user_cursor_sessions() -> TranscriptIngestStats { let Ok(profile_root) = crate::storage::default_profile_root() else { return TranscriptIngestStats::default(); }; - let Some(db) = open_user_session_db(&profile_root).await else { + let Some(registered_roots) = try_registered_project_roots().await else { return TranscriptIngestStats::default(); }; - let Some(registered_roots) = try_registered_project_roots().await else { + ingest_user_cursor_sessions_at(&profile_root, registered_roots).await +} + +async fn ingest_user_cursor_sessions_at( + profile_root: &Path, + registered_roots: Vec, +) -> TranscriptIngestStats { + let Some(db) = open_user_session_db(profile_root).await else { return TranscriptIngestStats::default(); }; let (composer_stats, owned) = if let Some(source) = cursor_composer::CursorComposerSource::new() @@ -123,7 +147,7 @@ pub async fn ingest_user_cursor_sessions() -> TranscriptIngestStats { let source = source .with_skip_session_ids(owned) .for_user_scope(®istered_roots); - composer_stats.merge(ingest_source(&db, &source, &profile_root, None).await) + composer_stats.merge(ingest_source(&db, &source, profile_root, None).await) } pub async fn ingest_user_global_sources() -> TranscriptIngestStats { @@ -138,29 +162,47 @@ fn provider_selected(scope: Option, candidate: SessionProvider) /// outside an explicitly requested message-search scope. pub async fn ingest_user_global_sources_for_provider( provider: Option, +) -> TranscriptIngestStats { + let Ok(profile_root) = crate::storage::default_profile_root() else { + return TranscriptIngestStats::default(); + }; + let Some(roots) = try_registered_project_roots().await else { + return TranscriptIngestStats::default(); + }; + ingest_user_global_sources_for_provider_with_roots(&profile_root, provider, roots).await +} + +pub(crate) async fn ingest_user_global_sources_for_provider_at( + profile_root: &Path, + provider: Option, +) -> TranscriptIngestStats { + let Some(roots) = try_registered_project_roots_at(profile_root).await else { + return TranscriptIngestStats::default(); + }; + ingest_user_global_sources_for_provider_with_roots(profile_root, provider, roots).await +} + +async fn ingest_user_global_sources_for_provider_with_roots( + profile_root: &Path, + provider: Option, + roots: Vec, ) -> TranscriptIngestStats { let mut stats = TranscriptIngestStats::default(); if provider_selected(provider, SessionProvider::Codex) { - stats = stats.merge(ingest_user_codex_sessions(None).await); + stats = stats.merge(ingest_user_codex_sessions_at(profile_root, None, roots.clone()).await); } if provider_selected(provider, SessionProvider::Cursor) { - stats = stats.merge(ingest_user_cursor_sessions().await); + stats = stats.merge(ingest_user_cursor_sessions_at(profile_root, roots.clone()).await); } - let Ok(profile_root) = crate::storage::default_profile_root() else { - return stats; - }; - let Some(db) = open_user_session_db(&profile_root).await else { - return stats; - }; - let Some(roots) = try_registered_project_roots().await else { + let Some(db) = open_user_session_db(profile_root).await else { return stats; }; 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); + stats = + stats.merge(claude::ingest_user_sessions(&db, profile_root, None, roots.clone()).await); } let mut sources: Vec> = Vec::new(); if provider_selected(provider, SessionProvider::Vibe) @@ -189,7 +231,7 @@ pub async fn ingest_user_global_sources_for_provider( sources.push(Box::new(source.for_user_scope(roots))); } for source in sources { - let source_stats = ingest_source(&db, source.as_ref(), &profile_root, None).await; + let source_stats = ingest_source(&db, source.as_ref(), profile_root, None).await; stats = stats.merge(source_stats); } if stats.messages_upserted > 0 { @@ -201,6 +243,78 @@ pub async fn ingest_user_global_sources_for_provider( stats } +const STARTUP_USER_INGEST_COOLDOWN: std::time::Duration = std::time::Duration::from_secs(30); + +#[derive(Default)] +struct StartupUserIngestState { + running: bool, + last_completed: Option, +} + +static STARTUP_USER_INGESTS: std::sync::OnceLock< + std::sync::Mutex>, +> = std::sync::OnceLock::new(); + +struct StartupUserIngestGuard { + profile_root: PathBuf, + completed: bool, +} + +impl StartupUserIngestGuard { + fn claim(profile_root: PathBuf) -> Option { + let ingests = STARTUP_USER_INGESTS + .get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new())); + let mut ingests = ingests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let state = ingests.entry(profile_root.clone()).or_default(); + if state.running + || state + .last_completed + .is_some_and(|completed| completed.elapsed() < STARTUP_USER_INGEST_COOLDOWN) + { + return None; + } + state.running = true; + Some(Self { + profile_root, + completed: false, + }) + } +} + +impl Drop for StartupUserIngestGuard { + fn drop(&mut self) { + let Some(ingests) = STARTUP_USER_INGESTS.get() else { + return; + }; + let mut ingests = ingests + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let state = ingests.entry(self.profile_root.clone()).or_default(); + state.running = false; + if self.completed { + state.last_completed = Some(std::time::Instant::now()); + } + } +} + +/// Coalesces the profile-wide user transcript sweep shared by every project +/// server created during daemon startup. Live hooks still call +/// [`ingest_user_global_sources`] directly, so the cooldown cannot hide a +/// completed turn. +pub(crate) async fn ingest_user_global_sources_for_startup() -> TranscriptIngestStats { + let Ok(profile_root) = crate::storage::default_profile_root() else { + return TranscriptIngestStats::default(); + }; + let Some(mut guard) = StartupUserIngestGuard::claim(profile_root) else { + return TranscriptIngestStats::default(); + }; + let stats = ingest_user_global_sources().await; + guard.completed = true; + stats +} + const FILE_TRANSCRIPT_PROVIDERS: &[SessionProvider] = &[ SessionProvider::Claude, SessionProvider::Codex, @@ -323,6 +437,16 @@ pub(crate) async fn finalize_project_ingest(db: &GlobalDb, project_root: &Path) let _ = workflow_ingest::ingest_workflow_runs(db, project_root).await; } +/// Daemon-startup variant that coalesces the profile-wide user sweep while +/// still running the active project's independent ingestion pass. +pub(crate) async fn ingest_global_sources_for_startup( + db: &GlobalDb, + project_root: &Path, +) -> TranscriptIngestStats { + let user = ingest_user_global_sources_for_startup().await; + user.merge(ingest_project_sources_for_provider(db, project_root, None, true).await) +} + /// Runs the bounded commit-attribution sweep against the correlation store. /// For each `(branch, worktree)` pair touched since the last sweep, scans that /// branch's git log inside the pair's span window (widened by the merge gap) @@ -613,4 +737,22 @@ mod git_scan_tests { fn parse_git_log_commits_empty_is_empty() { assert!(parse_git_log_commits("").is_empty()); } + + #[test] + fn startup_user_ingest_claims_are_single_flight_and_cancellation_safe() { + let profile = tempfile::tempdir().unwrap().path().to_path_buf(); + let first = StartupUserIngestGuard::claim(profile.clone()).expect("first claim"); + assert!(StartupUserIngestGuard::claim(profile.clone()).is_none()); + + drop(first); + let mut retry = StartupUserIngestGuard::claim(profile.clone()) + .expect("an incomplete claim must release immediately"); + retry.completed = true; + drop(retry); + + assert!( + StartupUserIngestGuard::claim(profile).is_none(), + "a completed sweep should suppress the startup herd during cooldown" + ); + } } diff --git a/src/update_cmd.rs b/src/update_cmd.rs index 1959fadf1..411b26878 100644 --- a/src/update_cmd.rs +++ b/src/update_cmd.rs @@ -496,8 +496,9 @@ mod tests { use std::path::{Path, PathBuf}; use super::{ - RefreshPolicy, ReinstallOutcome, current_tracedecay_exe_from, partition_reinstall_results, - post_update_binary, post_update_binary_from, run_install_then_refresh, + RefreshPolicy, ReinstallOutcome, current_tracedecay_exe_from, normalize_bin_path, + partition_reinstall_results, post_update_binary, post_update_binary_from, + run_install_then_refresh, }; use tempfile::TempDir; use tracedecay::upgrade::UpgradeOutcome; @@ -811,7 +812,7 @@ mod tests { let resolved = post_update_binary(Some(&installed)).expect("installed path should resolve"); - assert_eq!(resolved, installed.to_string_lossy()); + assert_eq!(resolved, normalize_bin_path(&installed)); } #[test] diff --git a/tests/agent_suite/agent_test.rs b/tests/agent_suite/agent_test.rs index aeac6cc60..466855e5b 100644 --- a/tests/agent_suite/agent_test.rs +++ b/tests/agent_suite/agent_test.rs @@ -3607,6 +3607,10 @@ fn assert_codex_hooks_registered(hooks: &serde_json::Value) { codex_event_has_handler(hooks, "PostCompact", "hook-codex-post-compact"), "Codex PostCompact hook should generate app-server LCM summaries: {hooks}" ); + assert!( + codex_event_has_handler(hooks, "Stop", "hook-codex-stop"), + "Codex Stop hook should ingest and review the final user-scoped turn: {hooks}" + ); let matcher = codex_matcher_for_handler(hooks, "PostToolUse", "hook-codex-post-tool-use") .expect("PostToolUse handler should exist"); assert!( diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 42bce761e..c471dbd7a 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -398,7 +398,10 @@ pub fn sample_node(id: &str, name: &str, file_path: &str) -> Node { /// Small multi-thread runtime for `#[test]`-driven async dashboard fixtures. pub fn create_runtime() -> tokio::runtime::Runtime { match tokio::runtime::Builder::new_multi_thread() - .worker_threads(2) + // Dashboard tests issue synchronous ureq calls while the in-process + // server and libsql handlers use this same runtime. Two workers can + // deadlock under load (one blocked in ureq, one awaiting DB work). + .worker_threads(4) .enable_all() .build() { @@ -595,10 +598,18 @@ pub async fn wait_for_dashboard(agent: &ureq::Agent, base_url: &str) { // HTTP response (2xx). A bare connect success is not enough — the server // can accept then drop the socket during startup ("Peer disconnected"). for _ in 0..160 { - if let Ok(response) = agent.get(&probe).call() { - if response.status().is_success() { - return; - } + let probe_agent = agent.clone(); + let probe_url = probe.clone(); + let ready = tokio::task::spawn_blocking(move || { + probe_agent + .get(&probe_url) + .call() + .is_ok_and(|response| response.status().is_success()) + }) + .await + .unwrap_or(false); + if ready { + return; } tokio::time::sleep(Duration::from_millis(50)).await; } diff --git a/tests/core_cli_suite/cli_non_interactive_test.rs b/tests/core_cli_suite/cli_non_interactive_test.rs index 9fd773e2c..bde71ccaf 100644 --- a/tests/core_cli_suite/cli_non_interactive_test.rs +++ b/tests/core_cli_suite/cli_non_interactive_test.rs @@ -1765,6 +1765,103 @@ fn migrate_verify_text_reports_actual_apply_supported_state() { ); } +#[test] +fn migrate_registry_gc_cleans_stale_storage_metadata_and_preserves_live_and_blocked_stores() { + let home = TempDir::new().expect("home tempdir"); + let live_project = home.path().join("live-project"); + std::fs::create_dir_all(&live_project).expect("live project dir"); + std::fs::write(live_project.join("lib.rs"), "pub fn live() {}\n").expect("live source"); + + let init = tracedecay_command(home.path(), &live_project) + .args(["init", "."]) + .output() + .expect("init live project"); + assert!( + init.status.success(), + "init failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&init.stdout), + String::from_utf8_lossy(&init.stderr) + ); + + let stale_project = canonical_temp_path(home.path()).join("gone-project"); + let global_db_path = profile_root(home.path()).join("global.db"); + create_runtime().block_on(async { + let db = GlobalDb::open_at(&global_db_path) + .await + .expect("open global db"); + db.upsert(&stale_project, 1).await; + db.upsert_code_project("stale-identity", &stale_project, None, None, None) + .await + .expect("register stale project identity"); + }); + + let blocked_manifest = profile_root(home.path()) + .join("projects/proj_blocked") + .join(STORE_MANIFEST_FILENAME); + std::fs::create_dir_all(blocked_manifest.parent().expect("manifest parent")) + .expect("blocked store dir"); + std::fs::write(&blocked_manifest, b"blocked-store-sentinel") + .expect("blocked manifest sentinel"); + + let preview = tracedecay_command(home.path(), &live_project) + .args(["migrate", "registry-gc", "--json"]) + .output() + .expect("registry-gc preview"); + assert!( + preview.status.success(), + "preview failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&preview.stdout), + String::from_utf8_lossy(&preview.stderr) + ); + let preview: serde_json::Value = serde_json::from_slice(&preview.stdout).expect("preview json"); + assert_eq!(preview["candidate_count"], 1); + assert_eq!(preview["metadata_candidate_count"], 2); + assert_eq!(preview["code_project_candidate_count"], 1); + assert_eq!(preview["storage_project_candidate_count"], 1); + assert_eq!(preview["deleted_count"], 0); + assert_eq!( + preview["storage_project_candidates"][0], + stale_project.display().to_string() + ); + + let apply = tracedecay_command(home.path(), &live_project) + .args(["migrate", "registry-gc", "--apply", "--json"]) + .output() + .expect("registry-gc apply"); + assert!( + apply.status.success(), + "apply failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&apply.stdout), + String::from_utf8_lossy(&apply.stderr) + ); + let apply: serde_json::Value = serde_json::from_slice(&apply.stdout).expect("apply json"); + assert_eq!(apply["deleted_code_project_count"], 1); + assert_eq!(apply["deleted_storage_project_count"], 1); + assert_eq!(apply["deleted_count"], 2); + + let remaining = create_runtime().block_on(async { + GlobalDb::open_at(&global_db_path) + .await + .expect("reopen global db") + .list_project_paths() + .await + }); + assert!( + !remaining + .iter() + .any(|path| Path::new(path) == stale_project) + ); + assert!( + remaining + .iter() + .any(|path| Path::new(path) == canonical_temp_path(&live_project)) + ); + assert_eq!( + std::fs::read(&blocked_manifest).expect("blocked manifest retained"), + b"blocked-store-sentinel" + ); +} + #[test] fn migrate_plan_save_writes_manifest_and_prints_confirmation_token_noninteractively() { let home = TempDir::new().unwrap(); diff --git a/tests/dashboard_api_test/lcm_fixes.rs b/tests/dashboard_api_test/lcm_fixes.rs index dea2f897f..6f045571e 100644 --- a/tests/dashboard_api_test/lcm_fixes.rs +++ b/tests/dashboard_api_test/lcm_fixes.rs @@ -411,6 +411,44 @@ async fn insert_undated_message(global_db_path: &Path) { } } +async fn insert_colliding_provider_message(global_db_path: &Path) { + let Some(global_db) = GlobalDb::open_at(global_db_path).await else { + panic!( + "failed to open colliding-provider fixture store at {}", + global_db_path.display() + ); + }; + let session = SessionRecord { + provider: "codex".to_string(), + session_id: SESSION_ID.to_string(), + project_key: "provider-collision".to_string(), + project_path: "/provider-collision".to_string(), + title: Some("Provider collision".to_string()), + started_at: Some(1_700_003_000), + ended_at: None, + transcript_path: None, + metadata_json: None, + parent_session_id: None, + is_subagent: false, + agent_id: None, + parent_tool_use_id: None, + }; + if !global_db.upsert_session(&session).await { + panic!("failed to upsert colliding provider session fixture"); + } + let mut colliding = message( + "msg-collision", + "user", + 1, + 1_700_003_000, + "provider collision", + ); + colliding.provider = "codex".to_string(); + if !global_db.upsert_session_message(&colliding).await { + panic!("failed to upsert colliding provider message fixture"); + } +} + #[test] fn basic_lcm_dashboard_errors_and_timeline_contracts() { let _env_lock = GLOBAL_DB_ENV_LOCK @@ -670,6 +708,35 @@ fn session_endpoint_orders_by_ordinal_paginates_nodes_and_enriches_messages() { }); } +#[test] +fn session_endpoint_rejects_provider_ambiguous_session_ids() { + let _env_lock = GLOBAL_DB_ENV_LOCK + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let runtime = create_runtime(); + runtime.block_on(async { + let fixture = start_fixture(false).await; + insert_colliding_provider_message(&fixture.global_db_path).await; + + let (status, response) = get_json( + &http_agent(), + &format!( + "{}/api/plugins/hermes-lcm/session/{SESSION_ID}", + fixture.base_url + ), + ); + + assert_eq!(status, 409); + assert!( + response["detail"] + .as_str() + .unwrap_or_default() + .contains("ambiguous session id across providers"), + "unexpected response: {response}" + ); + }); +} + #[test] fn search_matches_externalized_messages_and_qualifies_summary_fts() { let _env_lock = GLOBAL_DB_ENV_LOCK diff --git a/tests/hooks_lsp_suite/hook_lifecycle_lease_test.rs b/tests/hooks_lsp_suite/hook_lifecycle_lease_test.rs index cdd2711a9..c5228bdfc 100644 --- a/tests/hooks_lsp_suite/hook_lifecycle_lease_test.rs +++ b/tests/hooks_lsp_suite/hook_lifecycle_lease_test.rs @@ -28,6 +28,7 @@ const STDIN_HOOKS: &[&str] = &[ "hook-codex-subagent-start", "hook-codex-post-tool-use", "hook-codex-post-compact", + "hook-codex-stop", ]; fn hold_external_exclusive_lease(home: &Path) -> File { @@ -69,7 +70,7 @@ fn run_hook(home: &Path, hook: &str, input: Option<&[u8]>) -> Output { #[test] fn exclusive_lifecycle_owner_quiesces_every_hook_before_startup_or_dispatch() { assert_eq!(NO_INPUT_HOOKS.len(), 3); - assert_eq!(STDIN_HOOKS.len(), 21); + assert_eq!(STDIN_HOOKS.len(), 22); let temp = tempfile::tempdir().unwrap(); let home = temp.path(); let profile = home.join(".tracedecay"); diff --git a/tests/hooks_lsp_suite/hook_replay_test.rs b/tests/hooks_lsp_suite/hook_replay_test.rs index 16b189f21..a53ebe140 100644 --- a/tests/hooks_lsp_suite/hook_replay_test.rs +++ b/tests/hooks_lsp_suite/hook_replay_test.rs @@ -100,6 +100,13 @@ fn replays(root: &str) -> Vec { })), tool_input_env: None, }, + Replay { + subcommand: "hook-codex-stop", + agent: "codex", + hook_name: "Stop", + stdin: Some(json!({ "session_id": "codex-s1", "cwd": root })), + tool_input_env: None, + }, Replay { subcommand: "hook-cursor-session-start", agent: "cursor", diff --git a/tests/hooks_lsp_suite/hooks_test.rs b/tests/hooks_lsp_suite/hooks_test.rs index 028f79718..3a1ee6e74 100644 --- a/tests/hooks_lsp_suite/hooks_test.rs +++ b/tests/hooks_lsp_suite/hooks_test.rs @@ -1418,6 +1418,9 @@ fn test_codex_subagent_start_allows_invalid_json() { #[test] fn test_codex_subagent_start_injects_context_for_new_no_history_agent() { + let _lock = lock_global_db_env(); + let profile = tempfile::tempdir().unwrap(); + let _profile_env = EnvVarGuard::set(USER_DATA_DIR_ENV, profile.path().to_str().unwrap()); let input = r#"{ "hook_event_name": "SubagentStart", "agent_type": "generalPurpose", diff --git a/tests/mcp_suite/mcp_handler_test.rs b/tests/mcp_suite/mcp_handler_test.rs index 8b1650794..28fc53e9f 100644 --- a/tests/mcp_suite/mcp_handler_test.rs +++ b/tests/mcp_suite/mcp_handler_test.rs @@ -12432,6 +12432,62 @@ async fn lcm_status_cli_bridge_accepts_json_args() { ); } +#[cfg(unix)] +#[tokio::test] +async fn user_message_search_cli_bridge_accepts_storage_scope() { + let home_dir = test_temp_dir(); + let home = home_dir.path().join("home"); + std::fs::create_dir_all(&home).unwrap(); + let _daemon = common::spawn_tracedecay_daemon(&home); + let outside_cwd = test_temp_dir(); + + let mut ingest = std::process::Command::new(env!("CARGO_BIN_EXE_tracedecay")); + common::apply_tracedecay_home_env(&mut ingest, &home); + let ingest_output = ingest + .current_dir(outside_cwd.path()) + .args([ + "tool", + "tracedecay_lcm_preflight", + "--json", + "--args", + r#"{"storage_scope":"user","provider":"codex","session_id":"projectless-codex","messages":[{"id":"projectless-message","role":"user","content":"Projectless apricot preference"}],"transcript_projection":true,"format":"json"}"#, + ]) + .output() + .unwrap(); + assert!( + ingest_output.status.success(), + "user ingest failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&ingest_output.stdout), + String::from_utf8_lossy(&ingest_output.stderr) + ); + + let mut search = std::process::Command::new(env!("CARGO_BIN_EXE_tracedecay")); + common::apply_tracedecay_home_env(&mut search, &home); + let search_output = search + .current_dir(outside_cwd.path()) + .args([ + "tool", + "tracedecay_message_search", + "--json", + "--args", + r#"{"storage_scope":"user","provider":"codex","query":"apricot","format":"json"}"#, + ]) + .output() + .unwrap(); + assert!( + search_output.status.success(), + "user search failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&search_output.stdout), + String::from_utf8_lossy(&search_output.stderr) + ); + let envelope: Value = serde_json::from_slice(&search_output.stdout).unwrap(); + let payload = extract_first_json_content(&envelope); + assert_eq!(payload["status"], "ok"); + assert_eq!(payload["catch_up_performed"], true); + assert_eq!(payload["count"], 1); + assert_eq!(payload["results"][0]["session"]["project_key"], "user"); +} + #[test] fn memory_tool_definitions_include_hermes_payload_fields() { let tools = get_tool_definitions(); @@ -12943,6 +12999,10 @@ fn message_search_provider_schema_matches_ingested_providers() { message_search.input_schema["properties"]["scope"]["enum"], serde_json::json!(["all", "parents_only", "subagents_only"]) ); + assert_eq!( + message_search.input_schema["properties"]["storage_scope"]["enum"], + serde_json::json!(["project", "user"]) + ); assert!( message_search.input_schema["properties"] .get("parent_session_id") @@ -14713,14 +14773,11 @@ async fn mcp_server_owns_watcher_and_refreshes_token_map_on_change() { let (cg, _env) = init_test_project(project).await; cg.sync().await.unwrap(); + let mut config = tracedecay::config::load_config(project).expect("load test config"); + config.sync.session_start_sync = false; + tracedecay::config::save_config(project, &config).expect("disable unrelated catch-up"); let server = tracedecay::mcp::McpServer::new(cg.into_inner(), None).await; - assert!( - server - .wait_for_startup_catch_up(std::time::Duration::from_secs(30)) - .await, - "startup catch-up sync should finish before mutating the project" - ); let initial_count = server.file_token_map_snapshot().len(); @@ -15487,7 +15544,6 @@ async fn wait_for_startup_catch_up_waits_for_transcript_ingest_flag() { cg.index_all().await.unwrap(); let server = tracedecay::mcp::McpServer::new(cg.into_inner(), None).await; - server.run_startup_catch_up_sync().await; let completed = server .wait_for_startup_catch_up(std::time::Duration::from_secs(30)) diff --git a/tests/mcp_suite/mcp_test.rs b/tests/mcp_suite/mcp_test.rs index 3178ee2c4..f4de3a574 100644 --- a/tests/mcp_suite/mcp_test.rs +++ b/tests/mcp_suite/mcp_test.rs @@ -113,7 +113,7 @@ fn test_tool_definitions_count() { tool.name ); let storage_scope = tool.input_schema["properties"].get("storage_scope"); - if tool.name.starts_with("tracedecay_lcm_") { + if tool.name.starts_with("tracedecay_lcm_") || tool.name == "tracedecay_message_search" { assert_eq!(storage_scope.unwrap()["enum"], json!(["project", "user"])); } else { assert!( diff --git a/tests/mcp_suite/serve_harness.rs b/tests/mcp_suite/serve_harness.rs index 6a8ac6d53..bc726a4c6 100644 --- a/tests/mcp_suite/serve_harness.rs +++ b/tests/mcp_suite/serve_harness.rs @@ -48,9 +48,18 @@ pub async fn init_project_direct(home: &Path, project: &Path) { } pub async fn register_global_project(home: &Path, project: &Path) { + use std::hash::{Hash, Hasher}; + let home = canonical_existing_path(home); let db_path = home.join(".tracedecay/global.db"); let db = GlobalDb::open_at(&db_path).await.unwrap(); + let canonical = GlobalDb::canonical_project_key(project); + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + canonical.hash(&mut hasher); + let project_id = format!("test_{:016x}", hasher.finish()); + db.upsert_code_project(&project_id, project, None, None, None) + .await + .expect("register checked test project identity"); db.upsert(project, 0).await; db.checkpoint().await; } diff --git a/tests/transcript_ingest_suite/codex.rs b/tests/transcript_ingest_suite/codex.rs index af4e81cff..23699381f 100644 --- a/tests/transcript_ingest_suite/codex.rs +++ b/tests/transcript_ingest_suite/codex.rs @@ -92,6 +92,120 @@ async fn user_scope_excludes_codex_turns_after_switching_to_registered_project() ); } +#[tokio::test] +async fn project_scopes_split_codex_turns_when_cwd_changes() { + let tmp = TempDir::new().unwrap(); + let home = tmp.path().join("home"); + let project_a = tmp.path().join("project-a"); + let project_b = tmp.path().join("project-b"); + std::fs::create_dir_all(&project_a).unwrap(); + std::fs::create_dir_all(&project_b).unwrap(); + let path = write_codex_rollout(&home, &project_a, "cross-project-session"); + let mut file = std::fs::OpenOptions::new().append(true).open(path).unwrap(); + writeln!( + file, + "{}", + serde_json::json!({ + "timestamp": "2026-01-01T00:00:03.000Z", + "type": "turn_context", + "payload": {"cwd": project_b.to_string_lossy()} + }) + ) + .unwrap(); + writeln!( + file, + "{}", + serde_json::json!({ + "timestamp": "2026-01-01T00:00:04.000Z", + "type": "event_msg", + "payload": {"type": "agent_message", "message": "project beta private marker"} + }) + ) + .unwrap(); + + let db_a = GlobalDb::open_at(&tmp.path().join("project-a.db")) + .await + .unwrap(); + let db_b = GlobalDb::open_at(&tmp.path().join("project-b.db")) + .await + .unwrap(); + let source = CodexSource::with_home(&home); + ingest_source(&db_a, &source, &project_a, None).await; + ingest_source(&db_b, &source, &project_b, None).await; + + assert!( + !db_a + .search_session_messages("codex", None, "billing pipeline", 10) + .await + .is_empty() + ); + assert!( + db_a.search_session_messages("codex", None, "project beta private marker", 10) + .await + .is_empty() + ); + assert!( + db_b.search_session_messages("codex", None, "billing pipeline", 10) + .await + .is_empty() + ); + assert!( + !db_b + .search_session_messages("codex", None, "project beta private marker", 10) + .await + .is_empty() + ); +} + +#[tokio::test] +async fn user_scope_ingests_codex_turns_after_leaving_a_registered_project() { + let tmp = TempDir::new().unwrap(); + let home = tmp.path().join("home"); + let registered = tmp.path().join("registered"); + let general = tmp.path().join("general-chat"); + std::fs::create_dir_all(®istered).unwrap(); + std::fs::create_dir_all(&general).unwrap(); + let path = write_codex_rollout(&home, ®istered, "project-to-user-session"); + let mut file = std::fs::OpenOptions::new().append(true).open(path).unwrap(); + writeln!( + file, + "{}", + serde_json::json!({ + "timestamp": "2026-01-01T00:00:03.000Z", + "type": "turn_context", + "payload": {"cwd": general.to_string_lossy()} + }) + ) + .unwrap(); + writeln!( + file, + "{}", + serde_json::json!({ + "timestamp": "2026-01-01T00:00:04.000Z", + "type": "event_msg", + "payload": {"type": "agent_message", "message": "general chat private marker"} + }) + ) + .unwrap(); + let db = GlobalDb::open_at(&tmp.path().join("user-sessions.db")) + .await + .unwrap(); + let source = CodexSource::with_home(&home).for_user_scope(None, vec![registered]); + + ingest_source(&db, &source, tmp.path(), None).await; + + assert!( + db.search_session_messages("codex", None, "billing pipeline", 10) + .await + .is_empty() + ); + assert!( + !db.search_session_messages("codex", None, "general chat private marker", 10) + .await + .is_empty() + ); +} + /// Writes a Codex rollout JSONL whose `session_meta.cwd` is `project`. Includes a /// `response_item` line that must be ignored (it duplicates the agent_message). fn write_codex_rollout( diff --git a/tests/transcript_ingest_suite/cursor.rs b/tests/transcript_ingest_suite/cursor.rs index f5be3d340..f78a7be46 100644 --- a/tests/transcript_ingest_suite/cursor.rs +++ b/tests/transcript_ingest_suite/cursor.rs @@ -94,6 +94,7 @@ async fn user_cursor_event_rejects_registered_project_transcript_slug() { let event = serde_json::json!({ "session_id": "project-session", "transcript_path": transcript, + "workspace_roots": [registered], }); let db = GlobalDb::open_at(&tmp.path().join("user-sessions.db")) .await @@ -111,6 +112,99 @@ async fn user_cursor_event_rejects_registered_project_transcript_slug() { assert!(db.get_session("cursor", "project-session").await.is_none()); } +#[tokio::test] +async fn user_cursor_event_prefers_exact_workspace_over_colliding_slug() { + let tmp = TempDir::new().unwrap(); + let registered = tmp.path().join("work").join("foo-bar"); + let projectless = tmp.path().join("work").join("foo").join("bar"); + std::fs::create_dir_all(®istered).unwrap(); + std::fs::create_dir_all(&projectless).unwrap(); + assert_eq!( + cursor_project_slug(®istered), + cursor_project_slug(&projectless) + ); + let transcript_dir = tmp + .path() + .join(".cursor/projects") + .join(cursor_project_slug(&projectless).unwrap()) + .join("agent-transcripts"); + std::fs::create_dir_all(&transcript_dir).unwrap(); + let transcript = transcript_dir.join("projectless-session.jsonl"); + std::fs::write( + &transcript, + "{\"role\":\"user\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"projectless collision preference\"}]}}\n", + ) + .unwrap(); + let event = serde_json::json!({ + "session_id": "projectless-session", + "transcript_path": transcript, + "workspace_roots": [projectless], + }); + let db = GlobalDb::open_at(&tmp.path().join("user-sessions.db")) + .await + .unwrap(); + + let stats = ingest_cursor_user_transcript_event_capped_with_registered_roots( + &event.to_string(), + &db, + None, + &[registered], + ) + .await; + + assert_eq!(stats.messages_upserted, 1); + let session = db + .get_session("cursor", "projectless-session") + .await + .expect("exact projectless workspace should override its colliding slug"); + assert_eq!(session.project_path, "user"); +} + +#[tokio::test] +async fn user_cursor_event_without_workspace_fails_closed_on_slug_collision() { + let tmp = TempDir::new().unwrap(); + let registered = tmp.path().join("work").join("foo-bar"); + let colliding = tmp.path().join("work").join("foo").join("bar"); + assert_eq!( + cursor_project_slug(®istered), + cursor_project_slug(&colliding) + ); + let transcript_dir = tmp + .path() + .join(".cursor/projects") + .join(cursor_project_slug(&colliding).unwrap()) + .join("agent-transcripts"); + std::fs::create_dir_all(&transcript_dir).unwrap(); + let transcript = transcript_dir.join("ambiguous-session.jsonl"); + std::fs::write( + &transcript, + "{\"role\":\"user\",\"message\":{\"content\":[{\"type\":\"text\",\"text\":\"ambiguous collision\"}]}}\n", + ) + .unwrap(); + let event = serde_json::json!({ + "session_id": "ambiguous-session", + "transcript_path": transcript, + }); + let db = GlobalDb::open_at(&tmp.path().join("user-sessions.db")) + .await + .unwrap(); + + let stats = ingest_cursor_user_transcript_event_capped_with_registered_roots( + &event.to_string(), + &db, + None, + &[registered], + ) + .await; + + assert_eq!(stats.messages_upserted, 0); + assert!( + db.get_session("cursor", "ambiguous-session") + .await + .is_none() + ); +} + #[tokio::test] // Intentional: this test pins process-wide HOME/TRACEDECAY_GLOBAL_DB while the // hook resolves its storage paths. diff --git a/tests/update_health_pass_test.rs b/tests/update_health_pass_test.rs index 791d1c4ca..8df1c2c8f 100644 --- a/tests/update_health_pass_test.rs +++ b/tests/update_health_pass_test.rs @@ -1,4 +1,5 @@ #![allow(clippy::collapsible_if)] +#![cfg(not(windows))] //! Integration tests for the post-update health pass that runs at the end of //! `tracedecay update` / `tracedecay post-update` (skippable via `--no-heal`). //!