diff --git a/crates/mcp-server/src/hook_handlers/post_tool_use.rs b/crates/mcp-server/src/hook_handlers/post_tool_use.rs index 77fd6f1..ba6d5d2 100644 --- a/crates/mcp-server/src/hook_handlers/post_tool_use.rs +++ b/crates/mcp-server/src/hook_handlers/post_tool_use.rs @@ -259,6 +259,22 @@ fn completes_session_init(normalized_tool_name: &str, tool_response: &Value) -> matches!(normalized_tool_name, "init" | "context") && tool_call_succeeded(tool_response) } +/// Whether the Git checkout at `cwd` lacks a verified link to the project the +/// server resolved for it. Folders outside Git and agents' task worktrees are +/// never linked automatically, so they need no audit. +fn needs_link_audit(cwd: &str, workspace_id: Uuid, project_id: Option) -> bool { + let Some(layout) = mcp_session::checkout_repository_layout(cwd) else { + return false; + }; + let root = layout.repository_root.to_string_lossy().to_string(); + if crate::link::is_agent_worktree(&root) { + return false; + } + !project_id.is_some_and(|project_id| { + mcp_session::auto_init::checkout_binding_matches(&root, Some(workspace_id), project_id) + }) +} + fn contextstream_post_tool_nudge( normalized_tool_name: &str, action: &str, @@ -892,6 +908,12 @@ pub async fn handle() -> Result<()> { project_id, api_session_id.as_deref(), ); + // The bridge syncs only checkouts linked on this machine. + // Recorded above, this one is now in its next audit; ask for + // that audit now rather than in up to six hours. + if needs_link_audit(&cwd, workspace_id, project_id) { + let _ = crate::watch::request_link_audit(); + } // Docs written before this init/context were promised a mirror. if hook_feature_enabled("CONTEXTSTREAM_DOC_MIRROR_ENABLED") { let _ = super::durable_capture::mirror_pending_docs(&host_session).await; @@ -1880,6 +1902,49 @@ mod tests { ); assert!(msg.is_none()); } + + #[test] + fn a_git_checkout_without_a_verified_link_asks_the_bridge_for_an_audit() { + let directory = tempdir().unwrap(); + let linked = directory.path().join("linked"); + let project_id = Uuid::new_v4(); + std::fs::create_dir_all(&linked).unwrap(); + let workspace_id = write_repo_config(&linked, project_id); + let linked = linked.to_string_lossy(); + assert!(!needs_link_audit(&linked, workspace_id, Some(project_id))); + // The server resolved another project, or none. + assert!(needs_link_audit( + &linked, + workspace_id, + Some(Uuid::new_v4()) + )); + assert!(needs_link_audit(&linked, workspace_id, None)); + + let unlinked = directory.path().join("unlinked"); + std::fs::create_dir_all(unlinked.join(".git")).unwrap(); + assert!(needs_link_audit( + &unlinked.to_string_lossy(), + workspace_id, + Some(project_id) + )); + + // Plain folders and agents' task worktrees are never linked + // automatically. + let plain = directory.path().join("plain"); + std::fs::create_dir_all(&plain).unwrap(); + assert!(!needs_link_audit( + &plain.to_string_lossy(), + workspace_id, + Some(project_id) + )); + let task = directory.path().join("app/.claude/worktrees/agent-a1"); + std::fs::create_dir_all(task.join(".git")).unwrap(); + assert!(!needs_link_audit( + &task.to_string_lossy(), + workspace_id, + Some(project_id) + )); + } } #[cfg(test)] diff --git a/crates/mcp-server/src/hook_handlers/session_start.rs b/crates/mcp-server/src/hook_handlers/session_start.rs index bc72345..8d8bf0d 100644 --- a/crates/mcp-server/src/hook_handlers/session_start.rs +++ b/crates/mcp-server/src/hook_handlers/session_start.rs @@ -53,7 +53,14 @@ pub async fn handle() -> Result<()> { let config = load_config(&cwd); // Check index status for the current project - let index_info = check_index_status(&cwd); + let index_info = match check_index_status(&cwd) { + IndexInfo::NotIndexed => checkout_sync_status(&cwd).unwrap_or(IndexInfo::NotIndexed), + known => known, + }; + if matches!(index_info, IndexInfo::NotLinked) { + // A folder an earlier session recorded is in the bridge's next audit. + let _ = crate::watch::request_link_audit(); + } // Try to fetch compact restore state and startup context from API in // parallel so SessionStart can stay under the hook timeout budget. @@ -113,6 +120,18 @@ pub async fn handle() -> Result<()> { output_text.push_str("\n\u{2139} Project index not found yet \u{2014} use ContextStream search first (not Explore/Grep) to find code as it builds; fall back to local tools only if search itself returns nothing.\n"); output_text.push_str("Keep hosted MCP configured, re-establish the intended checkout with `mcp__contextstream__init(folder_path=\"\")`, then run `mcp__contextstream__project(action=\"index\")`; the exact-checkout sync bridge supplies local bytes. Verify with `mcp__contextstream__project(action=\"index_status\")`.\n"); } + IndexInfo::Synced => { + output_text.push_str( + "\n\u{2713} This checkout is linked and the sync bridge uploads its changes as you edit, so search reflects it.\n", + ); + } + IndexInfo::NotLinked => { + output_text.push_str("\n\u{26a0} This checkout is not linked to ContextStream on this machine, so the sync bridge uploads nothing from it. Search shows what other machines uploaded, which can be another branch or days old, and can miss code that exists only here.\n"); + output_text.push_str("The bridge links it by itself when its Git repository names a project. To link it now, run `contextstream-mcp link --fix` in this folder (`--project-id ` if no project matches), then `mcp__contextstream__project(action=\"index\")`.\n"); + } + IndexInfo::BridgeStopped => { + output_text.push_str("\n\u{26a0} This checkout is linked, but the ContextStream sync bridge is not running, so its changes are not uploaded and search can be out of date. Run `contextstream-mcp doctor --only-configured --repair --scope=all` to start it.\n"); + } } if let Some(restored) = restored_context { @@ -282,6 +301,53 @@ enum IndexInfo { Stale, Indexing, NotIndexed, + /// The checkout is linked here and a running sync bridge uploads it. + Synced, + /// A Git checkout with no verified link on this machine: the bridge + /// uploads nothing from it. + NotLinked, + /// Linked, but no sync bridge is running to upload it. + BridgeStopped, +} + +/// What the sync bridge does for the checkout at `folder_path`. Machines on +/// hosted MCP never write `indexed-projects.json`, so this is how they know +/// whether search reflects this checkout. `None` outside Git and in agents' +/// task worktrees, which are never synced. +fn checkout_sync_status(folder_path: &str) -> Option { + let layout = mcp_session::checkout_repository_layout(folder_path)?; + let root = layout.repository_root.to_string_lossy().to_string(); + if crate::link::is_agent_worktree(&root) { + return None; + } + if !checkout_linked(&root) { + return Some(IndexInfo::NotLinked); + } + let health = crate::watch::sync_bridge_health(); + let running = matches!(health.state, crate::watch::SyncBridgeHealthState::Running) + && health.target_count > 0; + Some(if running { + IndexInfo::Synced + } else { + IndexInfo::BridgeStopped + }) +} + +/// Whether the checkout-local config links `root` to its project with a +/// verified repository fingerprint (what the bridge requires to sync it). +fn checkout_linked(root: &str) -> bool { + std::fs::read_to_string(Path::new(root).join(".contextstream/config.json")) + .ok() + .and_then(|raw| serde_json::from_str::(&raw).ok()) + .and_then(|config| { + config + .get("project_id") + .and_then(Value::as_str) + .and_then(|id| uuid::Uuid::parse_str(id.trim()).ok()) + }) + .is_some_and(|project_id| { + mcp_session::auto_init::checkout_binding_workspace(root, project_id).is_some() + }) } fn check_index_status(folder_path: &str) -> IndexInfo { @@ -607,6 +673,24 @@ mod tests { ); } + #[test] + fn an_unlinked_git_checkout_is_reported_as_not_synced() { + let directory = tempfile::tempdir().unwrap(); + let checkout = directory.path().join("app"); + std::fs::create_dir_all(checkout.join(".git")).unwrap(); + assert!(matches!( + checkout_sync_status(&checkout.to_string_lossy()), + Some(IndexInfo::NotLinked) + )); + // Plain folders and agents' task worktrees keep the generic message. + let plain = directory.path().join("notes"); + std::fs::create_dir_all(&plain).unwrap(); + assert!(checkout_sync_status(&plain.to_string_lossy()).is_none()); + let task = checkout.join(".claude/worktrees/agent-a1"); + std::fs::create_dir_all(task.join(".git")).unwrap(); + assert!(checkout_sync_status(&task.to_string_lossy()).is_none()); + } + #[test] fn synthetic_test_lesson_phrase_is_noise() { let lesson = json!({ "title": "Test lesson from audit", "severity": "low" }); diff --git a/crates/mcp-server/src/link.rs b/crates/mcp-server/src/link.rs index 3e36289..25f8142 100644 --- a/crates/mcp-server/src/link.rs +++ b/crates/mcp-server/src/link.rs @@ -152,6 +152,10 @@ async fn relink(root: &Path, target: &ScoredProject) -> Result<()> { ) .await .map_err(|error| anyhow::anyhow!("could not link {}: {error}", root.display()))?; + // The bridge uploads a watched checkout's edits as they happen. A folder + // linked just now may hold a branch or edits the server has never seen, + // so its first sync covers the whole checkout. + let _ = crate::hook_handlers::dirty_drain::mark_watch_force_full(&root.to_string_lossy()); Ok(()) } @@ -248,14 +252,9 @@ pub async fn audit_folders( outcomes } -/// Folders this machine has linked or indexed: every existing directory in -/// `mappings.json` and `indexed-projects.json`, reduced to checkout roots. -pub fn known_folders() -> Vec { - let Some(dir) = contextstream_dir() else { - return Vec::new(); - }; +/// Folders listed in `mappings.json` and `indexed-projects.json`. +fn registry_folders(mappings: &Value, indexed: &Value) -> Vec { let mut paths: Vec = Vec::new(); - let mappings = read_json(&dir.join("mappings.json")); if let Some(entries) = mappings .get("mappings") .and_then(Value::as_array) @@ -268,10 +267,51 @@ pub fn known_folders() -> Vec { .map(str::to_string), ); } - let indexed = read_json(&dir.join("indexed-projects.json")); if let Some(projects) = indexed.get("projects").and_then(Value::as_object) { paths.extend(projects.keys().cloned()); } + paths +} + +/// Folders an agent session worked in on this machine, from +/// `session-scopes.json` (the hooks keep a day of sessions). A machine set up +/// for hosted MCP has no other record of a checkout it never linked, so +/// without these the bridge would never sync it. +fn session_folders(session_scopes: &Value) -> Vec { + session_scopes + .get("sessions") + .and_then(Value::as_object) + .map(|sessions| { + sessions + .values() + .filter_map(|scope| scope.get("cwd").and_then(Value::as_str)) + .filter(|cwd| !cwd.trim().is_empty()) + .map(str::to_string) + .collect() + }) + .unwrap_or_default() +} + +/// Worktrees a coding agent makes for one task and removes afterwards. +pub(crate) fn is_agent_worktree(path: &str) -> bool { + path.replace('\\', "/").contains("/.claude/worktrees/") +} + +fn auditable_root(root: &Path) -> bool { + root.parent().is_some() && Some(root) != dirs::home_dir().as_deref() +} + +/// Folders this machine has linked or indexed (every existing directory in +/// `mappings.json` and `indexed-projects.json`) and the Git checkouts agent +/// sessions worked in, reduced to checkout roots. +pub fn known_folders() -> Vec { + let Some(dir) = contextstream_dir() else { + return Vec::new(); + }; + let paths = registry_folders( + &read_json(&dir.join("mappings.json")), + &read_json(&dir.join("indexed-projects.json")), + ); let mut roots: Vec = paths .into_iter() // Throwaway clones and worktrees under temporary directories are never @@ -284,10 +324,23 @@ pub fn known_folders() -> Vec { .map(|layout| layout.repository_root) .or_else(|| std::fs::canonicalize(&path).ok()) }) - .filter(|root| { - root.parent().is_some() && Some(root.as_path()) != dirs::home_dir().as_deref() - }) + .filter(|root| auditable_root(root)) .collect(); + // Only Git checkouts: the automatic repair links a folder only when its + // repository names the project, which a plain folder never does. + roots.extend( + session_folders(&read_json(&dir.join("session-scopes.json"))) + .into_iter() + .filter(|path| { + Path::new(path).is_dir() + && !mcp_session::project_link::is_generated_path(path) + && !is_agent_worktree(path) + }) + .filter_map(|path| { + mcp_session::checkout_repository_layout(&path).map(|layout| layout.repository_root) + }) + .filter(|root| auditable_root(root) && !is_agent_worktree(&root.to_string_lossy())), + ); roots.sort(); roots.dedup(); roots @@ -472,3 +525,45 @@ pub async fn run(command: LinkCommand) -> Result { outcome.relinked_to.is_some() || matches!(outcome.assessment.status, LinkStatus::Correct) })) } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn registry_folders_come_from_mappings_and_the_local_index() { + let mappings = json!({"mappings": [{"path": "/work/a"}, {"project_id": "x"}]}); + let indexed = json!({"projects": {"/work/b": {"project_id": "y"}}}); + assert_eq!( + registry_folders(&mappings, &indexed), + vec!["/work/a".to_string(), "/work/b".to_string()] + ); + // The older mappings layout is a bare array. + assert_eq!( + registry_folders(&json!([{"path": "/work/c"}]), &Value::Null), + vec!["/work/c".to_string()] + ); + } + + #[test] + fn session_folders_are_where_agent_sessions_worked() { + let scopes = json!({"sessions": { + "a": {"cwd": "/work/app", "workspace_id": "w", "project_id": null}, + "b": {"cwd": " ", "workspace_id": "w"}, + "c": {"workspace_id": "w"} + }}); + assert_eq!(session_folders(&scopes), vec!["/work/app".to_string()]); + assert!(session_folders(&Value::Null).is_empty()); + } + + #[test] + fn agent_task_worktrees_are_recognised() { + assert!(is_agent_worktree("/work/app/.claude/worktrees/agent-a1b2")); + assert!(is_agent_worktree( + "C:\\work\\app\\.claude\\worktrees\\agent-a1" + )); + assert!(!is_agent_worktree("/work/app-wt/feature")); + assert!(!is_agent_worktree("/work/app/.claude")); + } +} diff --git a/crates/mcp-server/src/watch.rs b/crates/mcp-server/src/watch.rs index 18bb319..2839e05 100644 --- a/crates/mcp-server/src/watch.rs +++ b/crates/mcp-server/src/watch.rs @@ -65,6 +65,9 @@ const REENUMERATE_INTERVAL: Duration = Duration::from_secs(120); /// (see `crate::link`). The first check runs shortly after start. const LINK_AUDIT_INTERVAL: Duration = Duration::from_secs(6 * 60 * 60); const LINK_AUDIT_FIRST_DELAY: Duration = Duration::from_secs(120); +/// The least time between two audits that hooks asked for (see +/// `request_link_audit`). A folder no project matches asks on every turn. +const LINK_AUDIT_REQUEST_COOLDOWN: Duration = Duration::from_secs(10 * 60); /// How often the singleton lock's heartbeat timestamp is refreshed. const LOCK_REFRESH_INTERVAL: Duration = Duration::from_secs(60); @@ -108,6 +111,7 @@ const WATCH_LOCK_FILE: &str = "watch.lock"; const WATCH_HEARTBEAT_FILE: &str = "watch.heartbeat.json"; const WATCH_STOP_REQUEST_FILE: &str = "watch.stop.json"; const WATCH_RELOAD_REQUEST_FILE: &str = "watch.reload"; +const LINK_AUDIT_REQUEST_FILE: &str = "watch.link-audit"; const BRIDGE_CONTROL_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(60); const BRIDGE_CONTROL_CLAIM_INTERVAL: Duration = Duration::from_secs(30); const BRIDGE_CONTROL_LEASE_SECONDS: u64 = 180; @@ -909,17 +913,80 @@ pub fn request_sync_bridge_reload() -> Result { } fn consume_sync_bridge_reload_request_at(dir: &Path) -> bool { - let path = dir.join(WATCH_RELOAD_REQUEST_FILE); - match std::fs::remove_file(&path) { + consume_request_marker(&dir.join(WATCH_RELOAD_REQUEST_FILE), "reload") +} + +fn consume_request_marker(path: &Path, kind: &str) -> bool { + match std::fs::remove_file(path) { Ok(()) => true, Err(error) if error.kind() == std::io::ErrorKind::NotFound => false, Err(error) => { - tracing::debug!("watch: could not consume reload request: {}", error); + tracing::debug!("watch: could not consume {kind} request: {}", error); false } } } +/// Ask the running bridge to check which project each folder on this machine +/// belongs to now, instead of at its next six-hourly audit. +/// +/// Hooks call this when an agent works in a checkout that has no verified +/// link here. The audit links it only when its Git repository names the +/// project (see `crate::link`), and the bridge then syncs it. Like the reload +/// marker, the request holds no path or scope and is consumed once. +pub fn request_link_audit() -> Result { + let Some(dir) = contextstream_dir() else { + return Ok(false); + }; + request_link_audit_at(&dir) +} + +fn request_link_audit_at(dir: &Path) -> Result { + if probe_singleton_lock_at(dir) != LockProbe::Held { + return Ok(false); + } + crate::setup::safe_edit::write_owned_file_if_changed( + &dir.join(LINK_AUDIT_REQUEST_FILE), + "link-audit\n", + )?; + Ok(true) +} + +/// Take a pending audit request, unless an audit is running (it may have +/// listed the folders before the request's folder was recorded) or one was +/// requested within the cooldown. A request left in place is taken later. +fn take_link_audit_request_at( + dir: &Path, + audit_running: bool, + last_requested: Option, + now: Instant, +) -> bool { + !audit_running + && last_requested + .is_none_or(|last| now.saturating_duration_since(last) >= LINK_AUDIT_REQUEST_COOLDOWN) + && consume_request_marker(&dir.join(LINK_AUDIT_REQUEST_FILE), "link audit") +} + +/// Run the background link audit unless one is already running. A repaired +/// link changes which folders the bridge may sync, so the bridge reloads. +fn spawn_link_audit(client: &ContextStreamClient, running: &Arc) { + if running + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return; + } + let client = client.clone(); + let running = running.clone(); + tokio::spawn(async move { + let relinked = crate::link::background_audit(&client).await; + if relinked > 0 { + let _ = request_sync_bridge_reload(); + } + running.store(false, Ordering::Release); + }); +} + fn consume_sync_bridge_stop_request_at(dir: &Path, lock: &SingletonLock) -> bool { let path = dir.join(WATCH_STOP_REQUEST_FILE); let Ok(Some((request, raw))) = read_watch_stop_request(&path) else { @@ -1975,6 +2042,7 @@ pub async fn run_watch() -> Result<()> { ); link_audit_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let link_audit_running = Arc::new(AtomicBool::new(false)); + let mut last_requested_link_audit: Option = None; let mut lock_tick = tokio::time::interval(LOCK_REFRESH_INTERVAL); lock_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); let mut bridge_heartbeat_tick = tokio::time::interval(BRIDGE_CONTROL_HEARTBEAT_INTERVAL); @@ -2049,6 +2117,18 @@ pub async fn run_watch() -> Result<()> { "watch: reloaded managed checkout mappings" ); } + let now = Instant::now(); + if link_audit_enabled() + && take_link_audit_request_at( + &dir, + link_audit_running.load(Ordering::Acquire), + last_requested_link_audit, + now, + ) + { + last_requested_link_audit = Some(now); + spawn_link_audit(&client, &link_audit_running); + } } let now = Instant::now(); let due = take_due_watch_projects(&pending, &inflight, &holds, now); @@ -2102,22 +2182,8 @@ pub async fn run_watch() -> Result<()> { lock.write_heartbeat(targets.len()); } _ = link_audit_tick.tick() => { - if link_audit_enabled() - && link_audit_running - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_ok() - { - let client = client.clone(); - let running = link_audit_running.clone(); - tokio::spawn(async move { - // A repaired link changes which folders the bridge - // may sync; the next re-enumeration picks it up. - let relinked = crate::link::background_audit(&client).await; - if relinked > 0 { - let _ = request_sync_bridge_reload(); - } - running.store(false, Ordering::Release); - }); + if link_audit_enabled() { + spawn_link_audit(&client, &link_audit_running); } } _ = bridge_heartbeat_tick.tick() => { @@ -2345,6 +2411,46 @@ mod tests { assert!(directory.path().join(WATCH_STOP_REQUEST_FILE).exists()); } + #[test] + fn link_audit_request_needs_a_running_bridge_and_waits_out_a_running_audit_or_cooldown() { + let directory = tempdir().expect("tempdir"); + let marker = directory.path().join(LINK_AUDIT_REQUEST_FILE); + assert!(!request_link_audit_at(directory.path()).unwrap()); + assert!(!marker.exists()); + + let _lock = acquire_singleton_lock_at(directory.path()).expect("lock"); + assert!(request_link_audit_at(directory.path()).unwrap()); + assert_eq!(std::fs::read_to_string(&marker).unwrap(), "link-audit\n"); + + let now = Instant::now(); + assert!(!take_link_audit_request_at( + directory.path(), + true, + None, + now + )); + assert!(!take_link_audit_request_at( + directory.path(), + false, + Some(now), + now + Duration::from_secs(60) + )); + assert!(marker.exists(), "a request left in place is taken later"); + assert!(take_link_audit_request_at( + directory.path(), + false, + Some(now), + now + LINK_AUDIT_REQUEST_COOLDOWN + )); + assert!(!marker.exists()); + assert!(!take_link_audit_request_at( + directory.path(), + false, + None, + now + )); + } + #[test] fn reload_request_is_one_shot_and_contains_no_scope_data() { let directory = tempdir().expect("tempdir");