Merge pull request #448 from ScriptedAlchemy/fix/user-message-search-storage-scope

fix(sessions): harden user routing and daemon shutdown

Author: ScriptedAlchemy

.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.

.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)::/))'

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

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(),
         ])

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(),

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,

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<String> = 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::<std::collections::BTreeSet<_>>();
+            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);
                 }
             }
         }

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<OsString>,
+    previous_home: Option<OsString>,
+    previous_userprofile: Option<OsString>,
 }
 
 #[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"),
+            }
         }
     }
 }

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<Result<()>>, 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<JoinHandle<()>> = {
-            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<JoinHandle<()>> = {
+            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<Arc<crate::mcp::McpServer>> = {
             let servers = self.project_servers.lock().await;