diff --git a/crates/tracedecay-code-index-runtime/src/git_watch.rs b/crates/tracedecay-code-index-runtime/src/git_watch.rs index 265a8d482c..414df92ff6 100644 --- a/crates/tracedecay-code-index-runtime/src/git_watch.rs +++ b/crates/tracedecay-code-index-runtime/src/git_watch.rs @@ -612,6 +612,9 @@ async fn repository_task(inner: Arc, state: Arc) { /// state — it only records *what kind of path changed* so the debounce drain /// can resolve the actual git state once, after quiescence. fn classify_and_mark(state: &Arc, event: ¬ify::Event) { + if state.is_retired() { + return; + } let event_roots = state.event_roots(&event.paths); state.clear_retry(); // Cheap synchronous classification into the dirty set. We use `try_lock` to @@ -656,6 +659,9 @@ fn is_notify_capacity_error(error: ¬ify::Error) -> bool { } fn mark_notify_failure(state: &WatchState, error: ¬ify::Error) { + if state.is_retired() { + return; + } let status = if is_notify_capacity_error(error) { ProjectWatchStatus::NotifyCapacity } else { @@ -676,7 +682,7 @@ fn mark_notify_failure(state: &WatchState, error: ¬ify::Error) { /// Converts any callback event that could not record detailed path evidence /// into one conservative reconciliation plan. async fn materialize_pending_reconciliation(state: &WatchState) { - if !state.reconciliation_pending.load(Ordering::Acquire) { + if state.is_retired() || !state.reconciliation_pending.load(Ordering::Acquire) { return; } let mut dirty = state.dirty.lock().await; diff --git a/crates/tracedecay-code-index-runtime/src/git_watch/state.rs b/crates/tracedecay-code-index-runtime/src/git_watch/state.rs index 504976b974..c076cc0b80 100644 --- a/crates/tracedecay-code-index-runtime/src/git_watch/state.rs +++ b/crates/tracedecay-code-index-runtime/src/git_watch/state.rs @@ -347,6 +347,13 @@ impl WatchState { self.signal_retirement(); } + pub fn is_retired(&self) -> bool { + self.ownership + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .retired + } + fn signal_retirement(&self) { self.retirement.cancel(); self.wake.notify_waiters(); diff --git a/crates/tracedecay-dashboard-api/src/analytics_api.rs b/crates/tracedecay-dashboard-api/src/analytics_api.rs index 0371b58f47..b908744725 100644 --- a/crates/tracedecay-dashboard-api/src/analytics_api.rs +++ b/crates/tracedecay-dashboard-api/src/analytics_api.rs @@ -5,6 +5,7 @@ //! falls back to the legacy `dashboard_hint_events` table when present. use std::collections::BTreeMap; +use std::path::Path; use axum::extract::State; use axum::response::Json; @@ -652,7 +653,7 @@ fn build_subagent_tree(rows: Vec) -> Vec, - project_key: &str, + project_root: &Path, ) -> Result { let Some(db) = db else { return Ok(AnalyticsSubagentTreePayloadV1 { @@ -671,6 +672,8 @@ async fn subagent_tree_reading( }; let connection = db.read_connection(); + let canonical = RegisteredGlobalDb::canonical_project_key(project_root); + let opened = project_root.to_string_lossy().into_owned(); let rows = query_rows( &connection, "SELECT provider, @@ -685,14 +688,16 @@ async fn subagent_tree_reading( COALESCE(parent_tool_use_id, '') AS parent_tool_use_id FROM sessions -- Either column may carry the project: `project_key` is a provider's - -- own label and `project_path` the canonical root. Matching both is + -- own label and `project_path` the stored root. Matching both is -- the convention every scoped session read in `registered_sessions` - -- already uses, and matching only one silently empties the tree for - -- whichever provider labels its sessions the other way. - WHERE (project_key = ?1 OR project_path = ?1) + -- already uses. The opened spelling and the canonical OS identity + -- (`/var` vs `/private/var`) are one project; matching only the + -- canonical key silently empties the tree for rows stored under the + -- alias the host wrote. + WHERE (project_key IN (?1, ?2) OR project_path IN (?1, ?2)) ORDER BY COALESCE(started_at, 0), provider, session_id - LIMIT ?2", - params![project_key, SUBAGENT_TREE_SESSION_CEILING], + LIMIT ?3", + params![canonical, opened, SUBAGENT_TREE_SESSION_CEILING], ) .await .map_err(|error| format!("analytics subagent tree query failed: {error}"))?; @@ -751,8 +756,12 @@ pub async fn subagent_tree( ) -> Json>> { hotpath::future!( async move { - let project_key = RegisteredGlobalDb::canonical_project_key(&state.project_root); - match subagent_tree_reading(&state.host_io, state.lcm_db.as_deref(), &project_key).await + match subagent_tree_reading( + &state.host_io, + state.lcm_db.as_deref(), + &state.project_root, + ) + .await { Ok(payload) if !payload.available => Json(DashboardEnvelopeV1::unavailable( scope_from_state(&state), diff --git a/crates/tracedecay-runtime-core/src/worktree.rs b/crates/tracedecay-runtime-core/src/worktree.rs index 7a2503edf1..17bc79ef74 100644 --- a/crates/tracedecay-runtime-core/src/worktree.rs +++ b/crates/tracedecay-runtime-core/src/worktree.rs @@ -289,6 +289,8 @@ fn git_command() -> std::process::Command { command.env_remove("GIT_DIR"); command.env_remove("GIT_WORK_TREE"); command.env_remove("GIT_COMMON_DIR"); + command.env_remove("GIT_INDEX_FILE"); + command.env_remove("GIT_OBJECT_DIRECTORY"); command } @@ -300,12 +302,17 @@ mod tests { use tempfile::tempdir; fn run_git(cwd: &Path, args: &[&str]) { - let status = git_command() + let output = git_command() .args(args) .current_dir(cwd) - .status() + .output() .expect("git not on PATH — required for worktree tests"); - assert!(status.success(), "git {args:?} failed in {}", cwd.display()); + assert!( + output.status.success(), + "git {args:?} failed in {}: {}", + cwd.display(), + String::from_utf8_lossy(&output.stderr) + ); } #[test] diff --git a/crates/tracedecay-sessions/src/runtime/shared.rs b/crates/tracedecay-sessions/src/runtime/shared.rs index 350820d55c..5f28769344 100644 --- a/crates/tracedecay-sessions/src/runtime/shared.rs +++ b/crates/tracedecay-sessions/src/runtime/shared.rs @@ -202,6 +202,24 @@ pub fn path_identity_key(path: &str) -> String { .unwrap_or_else(|| path.to_owned()) } +/// Stored form of a live filesystem project path. +/// +/// [`path_identity_key`] folds Windows display syntax only; it leaves Unix +/// symlink aliases byte-exact. Session rows and dashboard reads must agree on +/// one OS identity (`/var` vs `/private/var`, `/tmp/link` vs `/tmp/real`) or a +/// scoped session query silently returns zero rows. +#[must_use] +pub fn durable_project_path_key(path: &str) -> String { + // Relative values include profile-scoped sentinels and opaque project IDs; + // their identity must not depend on directories in the daemon working directory. + if !Path::new(path).is_absolute() { + return path_identity_key(path); + } + let canonical = + tracedecay_runtime_core::path_safety::canonicalize_path_or_existing_parent(Path::new(path)); + path_identity_key(&canonical.to_string_lossy()) +} + fn canonical_drive_path(path: &str) -> Option { let head = path.as_bytes(); if head.len() >= 3 @@ -1108,6 +1126,55 @@ mod tests { assert_eq!(path_identity_key(r"\\server"), r"\\server"); } + #[test] + fn durable_project_path_preserves_non_path_identity() { + const CHILD: &str = "TD_SESSION_PATH_SENTINEL_TEST"; + if std::env::var_os(CHILD).is_some() { + for sentinel in ["user", "unknown", "opaque-project-id"] { + assert!(Path::new(sentinel).is_dir()); + assert_eq!(super::durable_project_path_key(sentinel), sentinel); + } + return; + } + let temp = tempfile::tempdir().unwrap(); + for sentinel in ["user", "unknown", "opaque-project-id"] { + std::fs::create_dir(temp.path().join(sentinel)).unwrap(); + } + // Isolate cwd in a child process so parallel tests retain their own paths. + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "durable_project_path_preserves_non_path_identity", + "--nocapture", + ]) + .env(CHILD, "1") + .current_dir(temp.path()) + .status() + .unwrap(); + assert!(status.success()); + } + + #[cfg(unix)] + #[test] + fn durable_project_path_collapses_a_symlink_alias() { + let temp = tempfile::TempDir::new().expect("temporary root"); + let real = temp.path().join("real"); + let link = temp.path().join("link"); + std::fs::create_dir_all(&real).expect("real project root"); + std::os::unix::fs::symlink(&real, &link).expect("directory alias"); + + let via_link = super::durable_project_path_key(link.to_str().expect("utf-8 link")); + let via_real = super::durable_project_path_key(real.to_str().expect("utf-8 real")); + assert_eq!( + via_link, via_real, + "one directory reached through a symlink alias is one stored project path" + ); + assert_ne!( + link.to_string_lossy().as_ref(), + via_link.as_str(), + "fixture must keep the caller spelling distinct from the stored identity" + ); + } + fn unknown_then_resolved_identity(path: &Path) -> GitRepositoryIdentityOutcome { if MATCHER_CACHE_RESOLVER_CALLS.fetch_add(1, Ordering::SeqCst) == 0 { GitRepositoryIdentityOutcome::Unknown(GitDiscoveryUnknown::DeadlineExceeded) diff --git a/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs b/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs index 24dbc02b8e..6cfd4e1b77 100644 --- a/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs +++ b/crates/tracedecay-sessions/src/runtime/store_access/sessions.rs @@ -15,7 +15,7 @@ use tracedecay_lcm::retrieval_content::{ }; use super::super::registered_db::{SessionRegisteredDb, SessionStoreAccess}; -use super::super::shared::path_identity_key; +use super::super::shared::{durable_project_path_key, path_identity_key}; use super::search::{ SESSION_MESSAGE_SEARCH_MAX_FETCH, downrank_inventory_messages, interleave_workflow_search_results, session_fts_query, @@ -43,8 +43,8 @@ pub(crate) const EXISTING_SESSION_MESSAGE_IDS_SQL: &str = "SELECT messages.messa /// Appends the project-scope predicate. /// /// `project_key` is an opaque authority and stays byte-exact. `project_path` -/// is written through `path_identity_key`, so the same selector can use its -/// exact spelling for the key and its canonical path spelling for the path. +/// may retain its opened spelling in observation projection or its canonical +/// OS identity in transcript persistence. Scoped reads accept both spellings. fn push_project_identity_predicate( sql: &mut String, query_params: &mut Vec, @@ -52,11 +52,13 @@ fn push_project_identity_predicate( ) { query_params.push(Value::Text(project_selector.to_owned())); let key_parameter = query_params.len(); - query_params.push(Value::Text(path_identity_key(project_selector))); + query_params.push(Value::Text(durable_project_path_key(project_selector))); let path_parameter = query_params.len(); + query_params.push(Value::Text(path_identity_key(project_selector))); + let opened_parameter = query_params.len(); let _ = write!( sql, - " AND (s.project_key = ?{key_parameter} OR s.project_path = ?{path_parameter})" + " AND (s.project_key = ?{key_parameter} OR s.project_path IN (?{path_parameter}, ?{opened_parameter}))" ); } @@ -1204,3 +1206,47 @@ fn row_to_workflow_message( metadata_json: Some(JsonValue::Object(metadata).to_string()), }) } + +#[cfg(all(test, unix))] +mod identity_tests { + use super::{Value, durable_project_path_key, push_project_identity_predicate}; + + #[test] + fn project_identity_reads_opened_and_canonical_paths_without_aliasing_keys() { + let temp = tempfile::tempdir().unwrap(); + let real = temp.path().join("real"); + let alias = temp.path().join("alias"); + std::fs::create_dir(&real).unwrap(); + std::os::unix::fs::symlink(&real, &alias).unwrap(); + let opened = alias.to_str().unwrap(); + let canonical = durable_project_path_key(opened); + assert_ne!(opened, canonical); + let db = rusqlite::Connection::open_in_memory().unwrap(); + db.execute_batch("CREATE TABLE sessions (project_key TEXT, project_path TEXT);") + .unwrap(); + for (key, path) in [ + ("typed-project-a", opened), + ("typed-project-a", canonical.as_str()), + (opened, "user"), + (canonical.as_str(), "user"), + ("unrelated", "unknown"), + ] { + db.execute("INSERT INTO sessions VALUES (?1, ?2)", [key, path]) + .unwrap(); + } + let mut sql = "SELECT count(*) FROM sessions s WHERE 1 = 1".to_owned(); + let mut params = Vec::new(); + push_project_identity_predicate(&mut sql, &mut params, opened); + let params = params.iter().map(|value| match value { + Value::Text(text) => text.as_str(), + _ => panic!("project identity parameters must be text"), + }); + let count: i64 = db + .query_row(&sql, rusqlite::params_from_iter(params), |row| row.get(0)) + .unwrap(); + assert_eq!( + count, 3, + "both path forms and only the byte-exact opaque key match" + ); + } +} diff --git a/crates/tracedecay-sessions/src/runtime/store_access/transcript.rs b/crates/tracedecay-sessions/src/runtime/store_access/transcript.rs index 07693ef3ed..c996ba26de 100644 --- a/crates/tracedecay-sessions/src/runtime/store_access/transcript.rs +++ b/crates/tracedecay-sessions/src/runtime/store_access/transcript.rs @@ -9,7 +9,7 @@ use super::super::git_correlation::{ CommitSessionRecord, SpanObservation, enqueue_git_evidence_publication, }; use super::super::registered_db::{SessionRegisteredDb, SessionStoreAccess, SessionWriteTxn}; -use super::super::shared::path_identity_key; +use super::super::shared::{durable_project_path_key, path_identity_key}; use super::codex_goal_reconciliation::find_preceding_codex_goal_response; use super::types::{TranscriptBatch, TranscriptPersistenceError}; @@ -311,7 +311,7 @@ impl SessionStoreAccess<'_, D> { session.provider.clone(), session.session_id.clone(), session.project_key.clone(), - path_identity_key(&session.project_path), + durable_project_path_key(&session.project_path), session.title.clone(), session.started_at, session.ended_at, diff --git a/crates/tracedecay-store-runtime/src/store_locator_resolver.rs b/crates/tracedecay-store-runtime/src/store_locator_resolver.rs index 4a4013f8a5..9b1092108e 100644 --- a/crates/tracedecay-store-runtime/src/store_locator_resolver.rs +++ b/crates/tracedecay-store-runtime/src/store_locator_resolver.rs @@ -29,6 +29,7 @@ use hotpath::rw_locks::{ use sha2::{Digest, Sha256}; use tracedecay_domain::canonical_text::sha256_hex; use tracedecay_runtime_core::db::DatabaseAuthority; +use tracedecay_runtime_core::path_safety::canonicalize_path_or_existing_parent; use tracedecay_runtime_core::shard_runtime::registry::{ ResolvedStoreLocator, StoreRuntimeKey, StoreRuntimeOpenMode, StoreRuntimeRegistryFailure, StoreRuntimeRegistryFuture, StoreRuntimeResolver, @@ -92,7 +93,13 @@ pub struct LocalProjectEnrollmentAuthorityV1 { impl LocalProjectEnrollmentAuthorityV1 { pub fn new(project_id: ProjectId, enrollment_roots: impl IntoIterator) -> Self { - let mut enrollment_roots = enrollment_roots.into_iter().collect::>(); + // macOS `/var` and `/private/var` are one directory. Two registrations + // of the same project that differ only by that alias are one authority, + // not DuplicateProjectAuthority. + let mut enrollment_roots = enrollment_roots + .into_iter() + .map(|root| canonicalize_path_or_existing_parent(&root)) + .collect::>(); enrollment_roots.sort(); enrollment_roots.dedup(); Self { @@ -2509,4 +2516,34 @@ mod tests { }) ); } + + /// macOS `/var` versus `/private/var`, reproduced with an explicit alias. + #[cfg(unix)] + #[test] + fn a_symlinked_enrollment_root_is_the_same_project_authority() { + let temp = tempfile::TempDir::new().expect("temporary root"); + let real = temp.path().join("real"); + let link = temp.path().join("link"); + fs::create_dir_all(&real).expect("real enrollment root"); + std::os::unix::fs::symlink(&real, &link).expect("directory alias"); + let project_id = id::("project.enrollment-alias"); + + let via_link = LocalProjectEnrollmentAuthorityV1::new(project_id.clone(), [link]); + let via_real = LocalProjectEnrollmentAuthorityV1::new(project_id.clone(), [real]); + assert_eq!( + via_link, via_real, + "one project reached through a symlink alias is one authority" + ); + + let resolver = LocalStoreRuntimeResolverV1::new(LocalProfileStoreAuthorityV1::new( + id::("brain.enrollment-alias"), + id::("profile.enrollment-alias"), + temp.path().join("profile"), + )) + .with_project_authority(via_link) + .expect("first spelling") + .with_project_authority(via_real) + .expect("alias spelling must not be DuplicateProjectAuthority"); + let _ = resolver; + } } diff --git a/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes.rs b/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes.rs index 4a71491bfe..6ef8a57c7d 100644 --- a/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes.rs +++ b/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes.rs @@ -205,6 +205,20 @@ async fn replay_projectless_hermes_receipts( for seq in retained_leases.into_iter().rev() { replay.defer(seq).await?; } + if target_outcome.is_none() + && let Some(seq) = target_seq + { + // The concurrent profile worker may have already committed this + // seq. `HostAdmissionRuntime::commit` returns Ok(0) when + // `seq <= committed_through` without requiring a lease; that is + // the spool watermark, not an inferred ExactDuplicate. Any other + // commit result is the broker's typed failure (lost / never + // committed). `accepted_for_replay` stays only for a full drain. + target_outcome = Some(match replay.commit(seq).await { + Ok(_) => HostAdmissionOutcome::replay_completed(true, false), + Err(outcome) => outcome, + }); + } Ok(terminal_outcome .or(target_outcome) .or(retained_outcome) diff --git a/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes/tests.rs b/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes/tests.rs index aa01a83842..50686d5a6c 100644 --- a/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes/tests.rs +++ b/crates/tracedecay/src/mcp/tools/handlers/hook_runtime/hermes/tests.rs @@ -208,6 +208,52 @@ async fn malformed_profile_payload_is_quarantined_across_reopen() { assert_eq!(recovered.quarantine_count().await, 1); } +#[tokio::test] +async fn inline_replay_reports_worker_committed_target_and_rejects_never_committed_seq() { + let temp = tempfile::TempDir::new().unwrap(); + let profile_root = temp.path().join("tracedecay-profile"); + std::fs::create_dir_all(&profile_root).unwrap(); + let fixture = HostAdmissionTestRuntimeV1::profile(&profile_root) + .await + .unwrap(); + let broker = fixture + .host_admission_broker_for_test(HostAdmissionScope::Profile) + .unwrap(); + let payload = valid_hermes_terminal_receipt_payload("session-race", "wm-race-1"); + let admitted = broker.admit("hermes:race-source", &payload).await.unwrap(); + + // The daemon worker drains with no target seq. Winning the replay lock + // commits `admitted.seq` and empties the spool before the inline path runs. + let worker_outcome = replay_projectless_hermes_host_admission(&broker, &profile_root).await; + assert!( + matches!( + worker_outcome.status, + HostAdmissionStatus::Committed | HostAdmissionStatus::AcceptedForReplay + ), + "worker drain must settle the admitted receipt, got {worker_outcome:?}" + ); + assert_eq!(broker.pending_count().await, 0); + + let raced = replay_projectless_hermes_receipts(&broker, &profile_root, Some(admitted.seq)) + .await + .expect("inline replay must finish with a typed disposition"); + assert_eq!( + raced.status, + HostAdmissionStatus::Committed, + "a worker-committed target seq must be reported as settled, got {raced:?}" + ); + + let missing_seq = admitted.seq.saturating_add(1); + let missing = replay_projectless_hermes_receipts(&broker, &profile_root, Some(missing_seq)) + .await + .expect("absent target must finish with a typed disposition"); + assert_eq!( + missing, + tracedecay_sessions::admission::HostAdmissionOutcome::spool_ack_conflict(), + "a seq the worker never committed must stay a typed failure, got {missing:?}" + ); +} + #[tokio::test] async fn unsupported_profile_payload_version_is_retained_without_apply() { let temp = tempfile::TempDir::new().unwrap();