Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/user-message-search-scope.md
Original file line number Diff line number Diff line change
@@ -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.
16 changes: 13 additions & 3 deletions .config/nextest.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand All @@ -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)::/))'
Expand Down
6 changes: 6 additions & 0 deletions src/agents/codex.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
5 changes: 5 additions & 0 deletions src/agents/codex/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
])
Expand Down Expand Up @@ -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();
Expand All @@ -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(),
])
Expand Down
3 changes: 3 additions & 0 deletions src/automation/jobs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down
3 changes: 3 additions & 0 deletions src/cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
64 changes: 53 additions & 11 deletions src/commands.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
}
Expand Down
23 changes: 20 additions & 3 deletions src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)]
Expand All @@ -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,
}
}
}
Expand All @@ -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"),
}
}
}
}
Expand Down
77 changes: 58 additions & 19 deletions src/daemon.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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",
&[
Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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());
Expand All @@ -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() {
Expand All @@ -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;
Expand Down
Loading
Loading