fix: retry uncertain session discovery

Author: HashemKhalifa

.changeset/session-worktree-discovery.md

@@ -0,0 +1,5 @@
+---
+"tracedecay": patch
+---
+
+Bound session worktree discovery so Git timeouts remain retryable without re-entering expensive repository scans or advancing transcript cursors.

src/daemon/git_watch.rs

@@ -466,13 +466,52 @@ async fn supervise_project(inner: Arc<GitWatcherInner>, state: Arc<WatchState>)
 /// One project's event loop: build the notify watcher over git metadata, then
 /// debounce raw events into coalesced syncs. On watcher construction/death,
 /// fall back to a 5-minute mtime poll for THIS project only.
+enum IdentityDiscoveryDisposition {
+    Watch(crate::worktree::GitRepoIdentity),
+    Degraded,
+    Retry,
+}
+
+fn identity_discovery_disposition(
+    outcome: crate::worktree::GitRepoIdentityOutcome,
+) -> IdentityDiscoveryDisposition {
+    match outcome {
+        crate::worktree::GitRepoIdentityOutcome::Resolved(identity) => {
+            IdentityDiscoveryDisposition::Watch(identity)
+        }
+        crate::worktree::GitRepoIdentityOutcome::NotFound => IdentityDiscoveryDisposition::Degraded,
+        crate::worktree::GitRepoIdentityOutcome::Unknown => IdentityDiscoveryDisposition::Retry,
+    }
+}
+
 async fn project_task(inner: Arc<GitWatcherInner>, state: Arc<WatchState>) {
-    let Some(identity) = crate::worktree::git_repo_identity(&state.project_root) else {
-        // Not a resolvable git repo (yet). Degrade to polling so a later `git
-        // init` / clone is still eventually covered.
-        state.health.set_degraded(true);
-        degraded_poll_loop(&inner, &state, None).await;
-        return;
+    let mut discovery_backoff = Duration::from_millis(500);
+    let identity = loop {
+        match identity_discovery_disposition(crate::worktree::git_repo_identity_outcome(
+            &state.project_root,
+        )) {
+            IdentityDiscoveryDisposition::Watch(identity) => break identity,
+            IdentityDiscoveryDisposition::Degraded => {
+                // Definitively not a git repo (yet). Degrade to polling so a
+                // later `git init` / clone is still eventually covered.
+                state.health.set_degraded(true);
+                degraded_poll_loop(&inner, &state, None).await;
+                return;
+            }
+            IdentityDiscoveryDisposition::Retry => {
+                state.health.set_degraded(true);
+                state.health.beat();
+                log_daemon_event(
+                    "git_watch_discovery_retry",
+                    &[
+                        ("project", state.project_root.display().to_string()),
+                        ("backoff_ms", discovery_backoff.as_millis().to_string()),
+                    ],
+                );
+                tokio::time::sleep(discovery_backoff).await;
+                discovery_backoff = (discovery_backoff * 2).min(RESTART_BACKOFF_MAX);
+            }
+        }
     };
     let common_dir = identity.common_dir;
 

src/daemon/git_watch/tests.rs

@@ -102,6 +102,18 @@ fn heartbeat_staleness() {
     assert!(old.heartbeat_stale());
 }
 
+#[test]
+fn timed_out_identity_discovery_retries_instead_of_degrading_forever() {
+    assert!(matches!(
+        identity_discovery_disposition(crate::worktree::GitRepoIdentityOutcome::Unknown),
+        IdentityDiscoveryDisposition::Retry
+    ));
+    assert!(matches!(
+        identity_discovery_disposition(crate::worktree::GitRepoIdentityOutcome::NotFound),
+        IdentityDiscoveryDisposition::Degraded
+    ));
+}
+
 /// The shared coordinator must not start a second store-writing lifetime while
 /// the first one is held. Paused Tokio time plus Notify/oneshot handshakes make
 /// this a scheduling-state assertion rather than a wall-clock sleep.

src/sessions/hermes.rs

@@ -37,9 +37,9 @@ use serde_json::{Map, Value};
 use crate::agents::hermes::read_config_pinned_project_root;
 use crate::global_db::{GlobalDb, ParseOffset, TranscriptBatch};
 use crate::sessions::shared::{
-    NewRows, ProjectRootMatcher, StoredCursor, TranscriptIngestStats, TranscriptLocation,
-    TranscriptLocationMetadataKeys, append_location_metadata, content_storage_text_and_tools,
-    path_belongs_to_project, preview_title, title_from_messages,
+    NewRows, ProjectMembership, ProjectRootMatcher, StoredCursor, TranscriptIngestStats,
+    TranscriptLocation, TranscriptLocationMetadataKeys, append_location_metadata,
+    content_storage_text_and_tools, path_belongs_to_project, preview_title, title_from_messages,
 };
 use crate::sessions::{SessionMessageRecord, SessionRecord};
 
@@ -539,7 +539,13 @@ async fn try_ingest_state_db_for_projects(
             &destination_matchers,
             source,
             &mut destination_routes,
-        );
+        )
+        .map_err(|_| {
+            format!(
+                "could not classify Hermes rows from '{}' because project membership is unknown",
+                state_db.display()
+            )
+        })?;
         for (state_index, state) in states
             .iter_mut()
             .enumerate()
@@ -980,12 +986,17 @@ struct DestinationTurnLocations {
     row_indices: Vec<usize>,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum DestinationRoutingError {
+    UnknownMembership,
+}
+
 fn turn_project_locations_for_destinations(
     rows: &[HermesRow],
     destination_matchers: &[ProjectRootMatcher],
     source: &HermesProfileSource,
     destination_routes: &mut HashMap<PathBuf, Vec<usize>>,
-) -> Vec<DestinationTurnLocations> {
+) -> Result<Vec<DestinationTurnLocations>, DestinationRoutingError> {
     let mut by_session: HashMap<&str, Vec<&HermesRow>> = HashMap::new();
     let row_indices = rows
         .iter()
@@ -1017,7 +1028,7 @@ fn turn_project_locations_for_destinations(
         let mut fallbacks = vec![None; destination_matchers.len()];
         for (cwd, provenance) in fallback_candidates {
             for destination_index in
-                matching_destinations(&cwd, destination_matchers, destination_routes)
+                matching_destinations(&cwd, destination_matchers, destination_routes)?
             {
                 fallbacks[destination_index].get_or_insert_with(|| HermesSessionLocation {
                     cwd: cwd.clone(),
@@ -1035,7 +1046,7 @@ fn turn_project_locations_for_destinations(
                     &row_indices,
                     &mut locations,
                     destination_routes,
-                );
+                )?;
                 turn.clear();
             }
             turn.push(row);
@@ -1047,12 +1058,12 @@ fn turn_project_locations_for_destinations(
             &row_indices,
             &mut locations,
             destination_routes,
-        );
+        )?;
     }
     for destination in &mut locations {
         destination.row_indices.sort_unstable();
     }
-    locations
+    Ok(locations)
 }
 
 fn assign_turn_locations_for_destinations(
@@ -1062,7 +1073,7 @@ fn assign_turn_locations_for_destinations(
     row_indices: &HashMap<i64, usize>,
     locations: &mut [DestinationTurnLocations],
     destination_routes: &mut HashMap<PathBuf, Vec<usize>>,
-) {
+) -> Result<(), DestinationRoutingError> {
     let explicit_paths = rows
         .iter()
         .rev()
@@ -1074,7 +1085,7 @@ fn assign_turn_locations_for_destinations(
     } else {
         for path in explicit_paths {
             for destination_index in
-                matching_destinations(&path, destination_matchers, destination_routes)
+                matching_destinations(&path, destination_matchers, destination_routes)?
             {
                 selected[destination_index].get_or_insert_with(|| HermesSessionLocation {
                     cwd: path.clone(),
@@ -1094,23 +1105,29 @@ fn assign_turn_locations_for_destinations(
             }
         }
     }
+    Ok(())
 }
 
 fn matching_destinations(
     path: &Path,
     destination_matchers: &[ProjectRootMatcher],
     destination_routes: &mut HashMap<PathBuf, Vec<usize>>,
-) -> Vec<usize> {
+) -> Result<Vec<usize>, DestinationRoutingError> {
     if let Some(indices) = destination_routes.get(path) {
-        return indices.clone();
+        return Ok(indices.clone());
+    }
+    let mut indices = Vec::new();
+    for (index, matcher) in destination_matchers.iter().enumerate() {
+        match matcher.contains_status(path) {
+            ProjectMembership::Match => indices.push(index),
+            ProjectMembership::NoMatch => {}
+            ProjectMembership::Unknown => {
+                return Err(DestinationRoutingError::UnknownMembership);
+            }
+        }
     }
-    let indices = destination_matchers
-        .iter()
-        .enumerate()
-        .filter_map(|(index, matcher)| matcher.contains(path).then_some(index))
-        .collect::<Vec<_>>();
     destination_routes.insert(path.to_path_buf(), indices.clone());
-    indices
+    Ok(indices)
 }
 
 fn assign_turn_location(
@@ -1422,3 +1439,92 @@ fn file_mtime_secs(path: &Path) -> u64 {
         .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
         .map_or(0, |duration| duration.as_secs())
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use super::*;
+
+    static IDENTITY_ATTEMPTS: AtomicUsize = AtomicUsize::new(0);
+
+    fn retrying_identity(path: &Path) -> crate::worktree::GitRepoIdentityOutcome {
+        let root = path
+            .ancestors()
+            .find(|ancestor| ancestor.file_name().is_some_and(|name| name == "repo"))
+            .unwrap_or(path);
+        if IDENTITY_ATTEMPTS.fetch_add(1, Ordering::SeqCst) == 0 {
+            return crate::worktree::GitRepoIdentityOutcome::Unknown;
+        }
+        crate::worktree::GitRepoIdentityOutcome::Resolved(crate::worktree::GitRepoIdentity {
+            worktree_root: root.to_path_buf(),
+            common_dir: root.join(".git"),
+        })
+    }
+
+    fn row_with_cwd(cwd: &Path) -> HermesRow {
+        HermesRow {
+            id: 1,
+            session_id: "session".to_string(),
+            role: "user".to_string(),
+            content: Some("retry me".to_string()),
+            tool_name: None,
+            tool_calls: None,
+            timestamp: None,
+            session_title: None,
+            session_model: None,
+            parent_session_id: None,
+            session_started_at: None,
+            session_ended_at: None,
+            session_source: None,
+            session_cwd: Some(cwd.to_string_lossy().to_string()),
+            session_input_tokens: None,
+            session_output_tokens: None,
+            session_cache_read_tokens: None,
+            session_cache_write_tokens: None,
+            session_reasoning_tokens: None,
+            active: 1,
+        }
+    }
+
+    #[test]
+    fn unknown_destination_route_retries_without_advancing_cursor() {
+        IDENTITY_ATTEMPTS.store(0, Ordering::SeqCst);
+        let temp = tempfile::TempDir::new().expect("temp dir");
+        let project_root = temp.path().join("repo");
+        let cwd = project_root.join("packages/app");
+        std::fs::create_dir_all(&cwd).expect("cwd");
+        let rows = vec![row_with_cwd(&cwd)];
+        let source = HermesProfileSource {
+            state_db: temp.path().join("state.db"),
+            profile: None,
+            legacy_project_pin: None,
+        };
+        let previous = StoredCursor::default();
+        let mut persisted = previous;
+        let mut routes = HashMap::new();
+
+        let first_matchers = vec![ProjectRootMatcher::new_with_identity_resolver(
+            &project_root,
+            retrying_identity,
+        )];
+        assert!(
+            turn_project_locations_for_destinations(&rows, &first_matchers, &source, &mut routes,)
+                .is_err()
+        );
+        assert_eq!(persisted, previous);
+        assert!(routes.is_empty(), "unknown routes must not be cached");
+
+        let retry_matchers = vec![ProjectRootMatcher::new_with_identity_resolver(
+            &project_root,
+            retrying_identity,
+        )];
+        let locations =
+            turn_project_locations_for_destinations(&rows, &retry_matchers, &source, &mut routes)
+                .expect("the same source rows should route after identity recovers");
+        assert_eq!(locations[0].row_indices, vec![0]);
+        persisted.position = rows[0].id as u64;
+        assert!(persisted.position > previous.position);
+        assert_eq!(IDENTITY_ATTEMPTS.load(Ordering::SeqCst), 3);
+    }
+}