Merge remote-tracking branch 'origin/fix/user-message-search-storage-scope' into fix/user-message-search-storage-scope

Author: ScriptedAlchemy

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,

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<ToolResult> {
     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)
     }))

src/sessions/mod.rs

@@ -54,6 +54,15 @@ pub async fn registered_project_roots() -> Vec<PathBuf> {
 /// is projectless.
 pub async fn try_registered_project_roots() -> Option<Vec<PathBuf>> {
     let global = GlobalDb::open().await?;
+    registered_project_roots_from(&global).await
+}
+
+async fn try_registered_project_roots_at(profile_root: &Path) -> Option<Vec<PathBuf>> {
+    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<Vec<PathBuf>> {
     let mut roots = global
         .list_project_paths()
         .await
@@ -72,27 +81,42 @@ pub async fn ingest_user_codex_sessions(session_id: Option<String>) -> 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<String>,
+    registered_roots: Vec<PathBuf>,
+) -> 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<PathBuf>,
+) -> 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(&registered_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<SessionProvider>, candidate: SessionProvider)
 /// outside an explicitly requested message-search scope.
 pub async fn ingest_user_global_sources_for_provider(
     provider: Option<SessionProvider>,
+) -> 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<SessionProvider>,
+) -> 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<SessionProvider>,
+    roots: Vec<PathBuf>,
 ) -> 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<Box<dyn TranscriptSource>> = 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 {
@@ -307,20 +349,7 @@ pub async fn ingest_global_sources_for_provider(
     project_root: &Path,
     provider: Option<SessionProvider>,
 ) -> TranscriptIngestStats {
-    ingest_global_sources_for_provider_inner(db, project_root, provider, true).await
-}
-
-async fn ingest_global_sources_for_provider_inner(
-    db: &GlobalDb,
-    project_root: &Path,
-    provider: Option<SessionProvider>,
-    include_user_scope: bool,
-) -> TranscriptIngestStats {
-    // Keep the profile-level projectless history current alongside every
-    // project sweep while respecting an explicit provider scope.
-    if include_user_scope {
-        let _ = ingest_user_global_sources_for_provider(provider).await;
-    }
+    let _ = ingest_user_global_sources_for_provider(provider).await;
     ingest_project_sources_for_provider(db, project_root, provider, true).await
 }
 
@@ -415,7 +444,7 @@ pub(crate) async fn ingest_global_sources_for_startup(
     project_root: &Path,
 ) -> TranscriptIngestStats {
     let user = ingest_user_global_sources_for_startup().await;
-    user.merge(ingest_global_sources_for_provider_inner(db, project_root, None, false).await)
+    user.merge(ingest_project_sources_for_provider(db, project_root, None, true).await)
 }
 
 /// Runs the bounded commit-attribution sweep against the correlation store.

tests/mcp_suite/mcp_handler_test.rs

@@ -12470,7 +12470,7 @@ async fn user_message_search_cli_bridge_accepts_storage_scope() {
             "tracedecay_message_search",
             "--json",
             "--args",
-            r#"{"storage_scope":"user","provider":"codex","query":"apricot","catch_up":false,"format":"json"}"#,
+            r#"{"storage_scope":"user","provider":"codex","query":"apricot","format":"json"}"#,
         ])
         .output()
         .unwrap();
@@ -12483,6 +12483,7 @@ async fn user_message_search_cli_bridge_accepts_storage_scope() {
     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");
 }