diff --git a/README.md b/README.md index c7315a6..642c1ea 100644 --- a/README.md +++ b/README.md @@ -133,8 +133,10 @@ changed would report a failed edit that was applied. The `index_codebase` tool does not wait for the first index. It registers the codebase, starts the scan and upload in the background, and returns at once with the codebase id and the path. The `sync_status` tool then reports the first-index -phase: registering, syncing, embedding, ready, or failed. `semctl daemon status` -reports the same phase for each checkout. +phase: registering, syncing, embedding, ready, or failed. While a sync runs, it +also reports a `current sync:` line with the latest step, such as the number of +uploaded files. `semctl daemon status` reports the same phase and step for each +checkout. ### Coming from the old `semctx` CLI @@ -181,7 +183,7 @@ A daemon started by hand skips the job check, and clients attach to it. Two commands inspect and end a running daemon: ```sh -semctl daemon status # version, pid, uptime, sessions, permits, checkouts and their first-index phase +semctl daemon status # version, pid, uptime, sessions, permits, checkouts, their first-index phase, and the progress of a running sync semctl daemon status --json # the same facts as one JSON object semctl daemon stop # end every session and exit ``` diff --git a/plugins/semctx/skills/codebase-retrieval/SKILL.md b/plugins/semctx/skills/codebase-retrieval/SKILL.md index 8330526..8017dfc 100644 --- a/plugins/semctx/skills/codebase-retrieval/SKILL.md +++ b/plugins/semctx/skills/codebase-retrieval/SKILL.md @@ -50,9 +50,10 @@ after explicit opt-in. That first-ever index runs in the background: catalog, and graph tools never serve a partial index. They wait at most 5 seconds, then fail with a "still running" error. Call `sync_status` every 10 to 15 seconds to follow the first-index phase (registering, syncing, embedding, -ready, or failed), and use local Read/Grep meanwhile. If the phase is failed, -call `index_codebase` for the same path to retry. Later re-syncs do not block -use of the last complete index. +ready, or failed), and use local Read/Grep meanwhile. While the phase is +syncing, the `current sync:` line shows how many files the upload has sent. If +the phase is failed, call `index_codebase` for the same path to retry. Later +re-syncs do not block use of the last complete index. ## Rules diff --git a/src/commands/index.rs b/src/commands/index.rs index 1a365e4..34b5b65 100644 --- a/src/commands/index.rs +++ b/src/commands/index.rs @@ -134,25 +134,7 @@ pub async fn run(args: IndexArgs, cli: &Cli) -> Result<()> { } fn report_sync_progress(progress: &SyncProgress) { - info!("{}", sync_progress_message(progress)); -} - -fn sync_progress_message(progress: &SyncProgress) -> String { - match progress { - SyncProgress::Preparing => "preparing index".to_string(), - SyncProgress::Scanning { root } => format!("scanning files in {}", root.display()), - SyncProgress::Planning { - files, - cached_files, - } => format!( - "scanned {files} files ({cached_files} filter decisions reused) — checking for changes" - ), - SyncProgress::Uploading { - uploaded_files, - total_files, - } => format!("uploading: {uploaded_files}/{total_files} files"), - SyncProgress::Finalizing => "finalizing upload".to_string(), - } + info!("{progress}"); } fn poll_interval(elapsed: std::time::Duration) -> std::time::Duration { @@ -219,11 +201,7 @@ async fn wait_for_job(client: &Client, job_id: &str) -> Result<()> { #[cfg(test)] mod tests { - use std::path::PathBuf; - - use super::{ - JobProgress, SyncProgress, api, poll_interval, sync_progress_message, terminal_result, - }; + use super::{JobProgress, api, poll_interval, terminal_result}; #[test] fn polling_is_responsive_for_fast_jobs_then_backs_off() { @@ -282,33 +260,4 @@ mod tests { assert_eq!(progress.processed, 12); assert_eq!(progress.total, 15); } - - #[test] - fn upload_progress_reports_completed_and_total_files() { - let message = sync_progress_message(&SyncProgress::Uploading { - uploaded_files: 3, - total_files: 8, - }); - assert_eq!(message, "uploading: 3/8 files"); - } - - #[test] - fn scan_summary_reports_verified_filter_decisions() { - let message = sync_progress_message(&SyncProgress::Planning { - files: 80, - cached_files: 73, - }); - assert_eq!( - message, - "scanned 80 files (73 filter decisions reused) — checking for changes" - ); - } - - #[test] - fn scan_progress_names_the_effective_root() { - let message = sync_progress_message(&SyncProgress::Scanning { - root: PathBuf::from("repo"), - }); - assert_eq!(message, "scanning files in repo"); - } } diff --git a/src/daemon/status.rs b/src/daemon/status.rs index 49ca63b..a80b71f 100644 --- a/src/daemon/status.rs +++ b/src/daemon/status.rs @@ -103,6 +103,7 @@ mod tests { use crate::engine::coordinator::WatcherState; use crate::engine::scheduler::{PermitUsage, SchedulerUsage}; use crate::mcp::readiness::FirstIndexPhase; + use crate::sync::SyncProgress; fn status() -> DaemonStatus { DaemonStatus { @@ -138,6 +139,10 @@ mod tests { first_index: Some(FirstIndexPhase::Embedding { job_id: Some("job-1".to_string()), }), + sync_progress: Some(SyncProgress::Uploading { + uploaded_files: 3, + total_files: 10, + }), }], } } @@ -174,6 +179,13 @@ mod tests { job_id: Some("job-1".to_string()) }) ); + assert_eq!( + coordinator.sync_progress, + Some(SyncProgress::Uploading { + uploaded_files: 3, + total_files: 10 + }) + ); } /// A reader that is not this build must find the documented field names. @@ -198,6 +210,10 @@ mod tests { value["coordinators"][0]["first_index"], serde_json::json!({"embedding": {"job_id": "job-1"}}) ); + assert_eq!( + value["coordinators"][0]["sync_progress"], + serde_json::json!({"uploading": {"uploaded_files": 3, "total_files": 10}}) + ); } #[test] @@ -223,6 +239,7 @@ mod tests { lines[8].contains("first index embedding (job job-1)"), "{rendered}" ); + assert!(lines[8].contains("sync uploading 3/10 files"), "{rendered}"); } #[test] diff --git a/src/engine/coordinator.rs b/src/engine/coordinator.rs index ec1db0b..0a8673b 100644 --- a/src/engine/coordinator.rs +++ b/src/engine/coordinator.rs @@ -48,10 +48,16 @@ use super::registry::CheckoutKey; use super::scheduler; use super::watch_hub::{WatchBatch, WatchRegistration, can_change_tree}; use crate::client::{self, Client}; -use crate::mcp::readiness::{FirstIndexPhase, InitialIndexGate}; +use crate::mcp::readiness::InitialIndexGate; use crate::sync::policy::{SourcePolicy, event_may_affect_policy}; use crate::sync::{self, LastJob, SyncCache, SyncLimits, SyncOutcome, blocking}; +mod progress; +mod status; + +use progress::ProgressSlot; +pub(crate) use status::CoordinatorStatus; + /// Reasons a reconcile can be due, before any coalescing. const TRIGGER_CAPACITY: usize = 64; @@ -100,6 +106,8 @@ pub(crate) struct ReconcileRun { pub(crate) root: PathBuf, pub(crate) cache: Arc>, pub(crate) limits: SyncLimits, + /// Where the reconcile reports its milestones for status callers. + pub(crate) progress: Arc, } /// A reconcile in progress. Boxed and `'static`, so an implementation must own @@ -128,9 +136,15 @@ pub(crate) struct SyncReconciler; impl Reconciler for SyncReconciler { fn reconcile(&self, run: ReconcileRun) -> Reconciled { Box::pin(async move { - sync::sync(&run.client, &run.root, &run.cache, &run.limits) - .await - .map_err(|error| format!("{error:#}")) + sync::sync_with_progress( + &run.client, + &run.root, + &run.cache, + &run.limits, + |progress| run.progress.publish(progress), + ) + .await + .map_err(|error| format!("{error:#}")) }) } @@ -183,57 +197,6 @@ struct RunState { last_error: Option, } -/// One checkout, as `sync_status` and the daemon status see it. -/// -/// `semctl daemon status` reports one of these per checkout, so this is the -/// single source of truth for that part of the status line as well. -#[derive(Clone, Debug, Serialize, Deserialize)] -pub(crate) struct CoordinatorStatus { - pub(crate) root: PathBuf, - pub(crate) codebase_id: Option, - pub(crate) leases: usize, - pub(crate) watcher: WatcherState, - pub(crate) last_job_id: Option, - pub(crate) running: bool, - pub(crate) pending_triggers: usize, - pub(crate) trigger_overflow: bool, - pub(crate) last_outcome: Option, - pub(crate) last_error: Option, - /// The phase of this checkout's first index. `None` when the checkout has - /// no first-index gate, and for a daemon that predates the field. - #[serde(default)] - pub(crate) first_index: Option, -} - -impl std::fmt::Display for CoordinatorStatus { - fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!( - formatter, - "root {} codebase {} leases {} watcher {} job {} running {} first index {} pending {}{} outcome {} error {}", - self.root.display(), - self.codebase_id.as_deref().unwrap_or("(unbound)"), - self.leases, - match &self.watcher { - WatcherState::Active => "active", - WatcherState::Unavailable(reason) => reason, - }, - self.last_job_id.as_deref().unwrap_or("(none)"), - self.running, - self.first_index - .as_ref() - .map_or_else(|| "(none)".to_string(), ToString::to_string), - self.pending_triggers, - if self.trigger_overflow { - " (overflowed)" - } else { - "" - }, - self.last_outcome.as_deref().unwrap_or("(none)"), - self.last_error.as_deref().unwrap_or("(none)"), - ) - } -} - /// Everything one coordinator needs to exist. The registry fills it; the watch /// registration is already decided, because registering is blocking work that /// must not happen under the registry's lock. @@ -270,6 +233,8 @@ pub(crate) struct CheckoutCoordinator { /// unconditional, because what was dropped cannot be judged. overflow: Arc, run: Mutex, + /// The latest milestone of the reconcile that is running. + progress: Arc, leases: AtomicUsize, /// When the lease count last reached zero, for the registry's sweeper. released_at: StdMutex>, @@ -309,6 +274,7 @@ impl CheckoutCoordinator { triggers, overflow: setup.overflow, run: Mutex::new(RunState::default()), + progress: Arc::new(ProgressSlot::new()), leases: AtomicUsize::new(0), released_at: StdMutex::new(None), task: StdMutex::new(None), @@ -415,6 +381,7 @@ impl CheckoutCoordinator { last_job_id, running: run.running, first_index, + sync_progress: self.progress.latest(), pending_triggers: self.triggers.max_capacity() - self.triggers.capacity(), trigger_overflow: self.overflow.load(Ordering::Acquire), last_outcome: run.last_outcome.clone(), @@ -558,14 +525,20 @@ impl CheckoutCoordinator { // process. It is held for the whole reconcile, including its uploads, // because the scan's retained content is what the bound protects. let permit = scheduler::permit(scan_permits).await; - let result = reconciler - .reconcile(ReconcileRun { - client: client.clone(), - root: self.root.clone(), - cache: self.cache.clone(), - limits: limits.clone(), - }) - .await; + let result = { + // The scope clears the milestone when the reconcile ends, fails, + // or is cancelled by dropping this future. + let _progress = self.progress.scope(); + reconciler + .reconcile(ReconcileRun { + client: client.clone(), + root: self.root.clone(), + cache: self.cache.clone(), + limits: limits.clone(), + progress: self.progress.clone(), + }) + .await + }; drop(permit); self.record(trigger, &result).await; // Any run can report a first index, because a run only starts once the diff --git a/src/engine/coordinator/progress.rs b/src/engine/coordinator/progress.rs new file mode 100644 index 0000000..70aa110 --- /dev/null +++ b/src/engine/coordinator/progress.rs @@ -0,0 +1,53 @@ +//! The latest milestone of the reconcile that is running. +//! +//! A reconciler publishes each milestone from a synchronous callback, and a +//! status caller reads the latest one. The slot is a `watch` channel because +//! publishing and reading never wait and never poison, so the callback cannot +//! stall an upload and a status call cannot stall a reconcile. + +use tokio::sync::watch; + +use crate::sync::SyncProgress; + +/// One coordinator's progress slot. +pub(crate) struct ProgressSlot { + latest: watch::Sender>, +} + +impl ProgressSlot { + pub(super) fn new() -> Self { + Self { + latest: watch::Sender::new(None), + } + } + + /// Record a milestone of the reconcile that is running. Only the latest + /// one is kept. + pub(crate) fn publish(&self, progress: &SyncProgress) { + self.latest.send_replace(Some(progress.clone())); + } + + /// The latest milestone, or `None` when no reconcile is running or none + /// reported one yet. + pub(super) fn latest(&self) -> Option { + self.latest.borrow().clone() + } + + /// Hold this guard for the length of one reconcile. Its drop clears the + /// slot, so a reconcile that ends, fails, panics, or is cancelled leaves + /// no stale milestone behind. + pub(super) fn scope(&self) -> ProgressScope<'_> { + ProgressScope { slot: self } + } +} + +/// Clears its slot when dropped. See [`ProgressSlot::scope`]. +pub(super) struct ProgressScope<'slot> { + slot: &'slot ProgressSlot, +} + +impl Drop for ProgressScope<'_> { + fn drop(&mut self) { + self.slot.latest.send_replace(None); + } +} diff --git a/src/engine/coordinator/status.rs b/src/engine/coordinator/status.rs new file mode 100644 index 0000000..54481ad --- /dev/null +++ b/src/engine/coordinator/status.rs @@ -0,0 +1,70 @@ +//! One checkout's status, as `sync_status` and the daemon status see it. + +use std::path::PathBuf; + +use serde::{Deserialize, Serialize}; + +use super::WatcherState; +use crate::mcp::readiness::FirstIndexPhase; +use crate::sync::SyncProgress; + +/// One checkout, as `sync_status` and the daemon status see it. +/// +/// `semctl daemon status` reports one of these per checkout, so this is the +/// single source of truth for that part of the status line as well. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub(crate) struct CoordinatorStatus { + pub(crate) root: PathBuf, + pub(crate) codebase_id: Option, + pub(crate) leases: usize, + pub(crate) watcher: WatcherState, + pub(crate) last_job_id: Option, + pub(crate) running: bool, + pub(crate) pending_triggers: usize, + pub(crate) trigger_overflow: bool, + pub(crate) last_outcome: Option, + pub(crate) last_error: Option, + /// The phase of this checkout's first index. `None` when the checkout has + /// no first-index gate, and for a daemon that predates the field. + #[serde(default)] + pub(crate) first_index: Option, + /// The latest milestone of the reconcile that is running now. `None` when + /// no reconcile reported one, and for a daemon that predates the field. + /// + /// This reuses [`SyncProgress`], the type the sync engine reports with, + /// so the wire shape and the engine cannot drift apart. + #[serde(default)] + pub(crate) sync_progress: Option, +} + +impl std::fmt::Display for CoordinatorStatus { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "root {} codebase {} leases {} watcher {} job {} running {} sync {} first index {} pending {}{} outcome {} error {}", + self.root.display(), + self.codebase_id.as_deref().unwrap_or("(unbound)"), + self.leases, + match &self.watcher { + WatcherState::Active => "active", + WatcherState::Unavailable(reason) => reason, + }, + self.last_job_id.as_deref().unwrap_or("(none)"), + self.running, + self.sync_progress + .as_ref() + .map_or_else(|| "(none)".to_string(), ToString::to_string), + self.first_index + .as_ref() + .map_or_else(|| "(none)".to_string(), ToString::to_string), + self.pending_triggers, + if self.trigger_overflow { + " (overflowed)" + } else { + "" + }, + self.last_outcome.as_deref().unwrap_or("(none)"), + self.last_error.as_deref().unwrap_or("(none)"), + ) + } +} diff --git a/src/engine/coordinator/tests.rs b/src/engine/coordinator/tests.rs index 219d09b..c29a1a5 100644 --- a/src/engine/coordinator/tests.rs +++ b/src/engine/coordinator/tests.rs @@ -19,7 +19,7 @@ use super::{ use crate::client::{Client, api}; use crate::engine::registry::CheckoutKey; use crate::mcp::readiness::{FirstIndexPhase, InitialIndexGate}; -use crate::sync::policy; +use crate::sync::{SyncProgress, policy}; /// Somewhere no test writes, so a coordinator that starts reading the /// filesystem fails instead of touching a real checkout. @@ -36,10 +36,24 @@ struct CountingReconciler { release: Arc, /// Each job poll takes one permit, so a test can hold a first index open. finish_polls: Arc, + /// The milestone each run publishes before it waits for its release. + milestone: Option, + /// How each run ends: `None` succeeds, and `Some` fails with the reason. + failure: Option, } impl CountingReconciler { fn new() -> (Arc, Runs) { + Self::build(None, None) + } + + /// A reconcile that reports `milestone` when it starts, and fails with + /// `failure` instead of succeeding when the test releases it. + fn reporting(milestone: SyncProgress, failure: Option<&str>) -> (Arc, Runs) { + Self::build(Some(milestone), failure.map(str::to_string)) + } + + fn build(milestone: Option, failure: Option) -> (Arc, Runs) { let (started, starts) = mpsc::unbounded_channel(); let release = Arc::new(Semaphore::new(0)); let finish_polls = Arc::new(Semaphore::new(0)); @@ -49,6 +63,8 @@ impl CountingReconciler { started, release: release.clone(), finish_polls: finish_polls.clone(), + milestone, + failure, }); ( reconciler.clone(), @@ -107,16 +123,24 @@ impl Runs { } impl Reconciler for CountingReconciler { - fn reconcile(&self, _run: super::ReconcileRun) -> Reconciled { + fn reconcile(&self, run: super::ReconcileRun) -> Reconciled { self.runs.fetch_add(1, Ordering::AcqRel); let _ = self.started.send(()); let release = self.release.clone(); + let milestone = self.milestone.clone(); + let failure = self.failure.clone(); Box::pin(async move { + if let Some(milestone) = &milestone { + run.progress.publish(milestone); + } let permit = release.acquire().await; // Only a closed semaphore fails, and no test closes one. if let Ok(permit) = permit { permit.forget(); } + if let Some(reason) = failure { + return Err(reason); + } Ok(SyncOutcome { codebase_id: "codebase".to_string(), job_id: "job".to_string(), @@ -348,10 +372,106 @@ async fn status_has_no_first_index_phase_without_a_gate() { assert!(status.to_string().contains("first index (none)")); } +fn uploading(uploaded_files: usize, total_files: usize) -> SyncProgress { + SyncProgress::Uploading { + uploaded_files, + total_files, + } +} + +/// Poll the coordinator's status until its sync progress is `expected`. +async fn wait_for_progress(coordinator: &CheckoutCoordinator, expected: Option) { + tokio::time::timeout(Duration::from_secs(5), async { + while coordinator.status().await.sync_progress != expected { + tokio::task::yield_now().await; + } + }) + .await + .unwrap_or_else(|_| panic!("the sync progress must become {expected:?}")); +} + +/// `sync_status` and `semctl daemon status` report how far the running +/// reconcile has come. The milestone must be visible while the reconcile runs +/// and gone when it ends. +#[tokio::test] +async fn status_shows_the_running_reconciles_progress_until_it_ends() { + let (reconciler, mut runs) = CountingReconciler::reporting(uploading(3, 10), None); + let (coordinator, task) = coordinator(reconciler).await; + assert_eq!(coordinator.status().await.sync_progress, None); + task.spawn(); + + coordinator.trigger(Trigger::Explicit); + runs.next_start().await; + wait_for_progress(&coordinator, Some(uploading(3, 10))).await; + let running = coordinator.status().await; + assert!(running.running); + assert!( + running.to_string().contains("sync uploading 3/10 files"), + "{running}" + ); + + runs.release(1); + wait_for_progress(&coordinator, None).await; + assert!( + coordinator + .status() + .await + .to_string() + .contains("sync (none)") + ); +} + +/// A failed reconcile must not leave its last milestone behind. The status +/// would otherwise show an upload that no longer runs. +#[tokio::test] +async fn a_failed_reconcile_clears_its_progress() { + let (reconciler, mut runs) = CountingReconciler::reporting(uploading(3, 10), Some("boom")); + let (coordinator, task) = coordinator(reconciler).await; + task.spawn(); + + coordinator.trigger(Trigger::Explicit); + runs.next_start().await; + wait_for_progress(&coordinator, Some(uploading(3, 10))).await; + runs.release(1); + + // The error is recorded after the progress is cleared, so the status that + // shows the error must show no progress. + let status = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let status = coordinator.status().await; + if status.last_error.is_some() { + return status; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("a failed run records its error"); + assert_eq!(status.last_error.as_deref(), Some("boom")); + assert_eq!(status.sync_progress, None); +} + +/// Cancelling the coordinator drops the run in progress. Its milestone must go +/// with it. +#[tokio::test] +async fn a_cancelled_reconcile_clears_its_progress() { + let (reconciler, mut runs) = CountingReconciler::reporting(uploading(3, 10), None); + let (coordinator, task) = coordinator(reconciler).await; + task.spawn(); + coordinator.trigger(Trigger::Explicit); + runs.next_start().await; + wait_for_progress(&coordinator, Some(uploading(3, 10))).await; + + coordinator.cancel(); + + wait_for_progress(&coordinator, None).await; +} + /// The daemon status crosses the process boundary. A daemon that predates the -/// first-index field sends none, and the reader must still decode its status. +/// first-index and sync-progress fields sends neither, and the reader must +/// still decode its status. #[test] -fn a_status_without_a_first_index_field_decodes_as_none() { +fn a_status_from_an_older_daemon_decodes_without_the_newer_fields() { let old = serde_json::json!({ "root": "/work/checkout", "codebase_id": "codebase", "leases": 1, "watcher": "active", "last_job_id": null, "running": false, @@ -363,6 +483,28 @@ fn a_status_without_a_first_index_field_decodes_as_none() { serde_json::from_value(old).expect("decode a status from an older daemon"); assert_eq!(status.first_index, None); + assert_eq!(status.sync_progress, None); +} + +#[test] +fn a_status_carries_its_sync_progress_across_the_wire() { + let current = serde_json::json!({ + "root": "/work/checkout", "codebase_id": "codebase", "leases": 1, + "watcher": "active", "last_job_id": null, "running": true, + "pending_triggers": 0, "trigger_overflow": false, + "last_outcome": null, "last_error": null, + "sync_progress": {"uploading": {"uploaded_files": 3, "total_files": 10}}, + }); + + let status: super::CoordinatorStatus = + serde_json::from_value(current).expect("decode a status with progress"); + let encoded = serde_json::to_value(&status).expect("encode the status"); + + assert_eq!(status.sync_progress, Some(uploading(3, 10))); + assert_eq!( + encoded["sync_progress"], + serde_json::json!({"uploading": {"uploaded_files": 3, "total_files": 10}}) + ); } /// A second reconcile during embedding sees the same pending gate. It must not diff --git a/src/mcp/docs/instructions/server.md b/src/mcp/docs/instructions/server.md index 300ec38..4586a83 100644 --- a/src/mcp/docs/instructions/server.md +++ b/src/mcp/docs/instructions/server.md @@ -43,7 +43,9 @@ It never waits for embedding. Retrieval, catalog, and graph tools never serve a partial first index. They wait at most 5 seconds for it to complete. If it still runs, they fail with a "still running" error. `sync_status` stays callable. It reports the first-index phase: registering, syncing, embedding, ready, or failed. -Call it every 10 to 15 seconds, and use local Read/Grep until the phase is ready. +While the phase is syncing, its `current sync:` line shows how many files the +upload has sent. Call it every 10 to 15 seconds, and use local Read/Grep until +the phase is ready. If the phase is failed, call `index_codebase` for the same path to retry. Later re-syncs do not block retrieval and continue to expose the last complete snapshot with freshness warnings. diff --git a/src/mcp/docs/tools/index_codebase.md b/src/mcp/docs/tools/index_codebase.md index b11d286..b213230 100644 --- a/src/mcp/docs/tools/index_codebase.md +++ b/src/mcp/docs/tools/index_codebase.md @@ -4,7 +4,7 @@ Call this only after the user has agreed to index a genuinely unindexed director For a first-ever index, the tool registers the codebase, starts the scan and upload in a background task, and returns at once. It does not wait for the server to embed the files. The answer names the codebase id and the path. Call `sync_status` every 10 to 15 seconds to follow the first index, and use local Read/Grep until it completes. If the call times out or is cancelled, the first index continues in the MCP server. -All codebase-scoped retrieval/catalog/graph tools targeting that repository check the same readiness gate, so they cannot observe a partial first index: each waits at most 5 seconds, then fails with a "still running" error. `sync_status` deliberately remains available and reports the first-index phase: registering, syncing, embedding, ready, or failed. +All codebase-scoped retrieval/catalog/graph tools targeting that repository check the same readiness gate, so they cannot observe a partial first index: each waits at most 5 seconds, then fails with a "still running" error. `sync_status` deliberately remains available and reports the first-index phase: registering, syncing, embedding, ready, or failed. While the sync runs, it also reports a `current sync:` line, such as `current sync: uploading 1200/3607 files`. A second call for the same path while the first index runs does not start another one. It answers "already in progress" with the same instruction. After the first index completes, a call answers that it is complete. If the first index failed, call this tool again with the same path to retry it. diff --git a/src/mcp/docs/tools/sync_status.md b/src/mcp/docs/tools/sync_status.md index 304cc1a..1913f1c 100644 --- a/src/mcp/docs/tools/sync_status.md +++ b/src/mcp/docs/tools/sync_status.md @@ -14,4 +14,14 @@ While a first index is running, or after it ended, the answer has a `first index The line is absent when the MCP server has no first index for the checkout, such as a checkout that was indexed in an earlier session. Call this tool every 10 to 15 seconds while a first index runs. +While a sync runs for the checkout, the answer has a `current sync:` line after the `first index:` line. This holds for a first index and for a later re-sync. When there is no `first index:` line, the `current sync:` line follows the codebase line. The line shows the latest step of the sync: + +- `current sync: preparing index`: the sync started. +- `current sync: scanning files in `: the sync reads the checkout. +- `current sync: scanned files ( filter decisions reused) — checking for changes`: the scan is done, and the sync compares the files with the server. +- `current sync: uploading / files`: the sync uploaded `` of `` files. `` counts only the files that the server needs, not all files in the checkout. +- `current sync: finalizing upload`: all files are uploaded, and the sync completes the upload. + +During a first index, the `first index: syncing` line and the `current sync:` line together show how far the first index has come. The `current sync:` line is absent while no sync runs, for example while the server embeds the files after the upload. Read the `first index:` line for the phase. + Unlike retrieval/catalog/graph tools, this tool does not wait on a first-index readiness gate, so it can monitor that initial job. diff --git a/src/mcp/tools.rs b/src/mcp/tools.rs index 3ba06b7..24f16dd 100644 --- a/src/mcp/tools.rs +++ b/src/mcp/tools.rs @@ -765,6 +765,9 @@ impl McpServer { status .as_ref() .and_then(|status| status.first_index.as_ref()), + status + .as_ref() + .and_then(|status| status.sync_progress.as_ref()), ) .await } diff --git a/src/query/inspection.rs b/src/query/inspection.rs index 330376c..636b9d9 100644 --- a/src/query/inspection.rs +++ b/src/query/inspection.rs @@ -8,6 +8,7 @@ use super::render::{ }; use super::{Client, ToolError, api, bound_codebase, urlencode}; use crate::mcp::readiness::FirstIndexPhase; +use crate::sync::SyncProgress; pub(super) const FILES_PAGE_MAX: u32 = 1000; @@ -160,11 +161,16 @@ pub async fn list_files( /// /// `first_index` is the phase of the checkout's first index. No tool call waits /// for a first index, so this line is how an agent follows it. +/// +/// `sync_progress` is the latest milestone of the reconcile that runs now. Its +/// line follows the first-index line, so an agent reads how far a first index +/// has come. There is no line when no reconcile runs. pub async fn sync_status( client: &Client, job_id: Option<&str>, local_watch_active: bool, first_index: Option<&FirstIndexPhase>, + sync_progress: Option<&SyncProgress>, ) -> Result { let codebase_id = bound_codebase(client, "sync_status")?; let totals = catalog_totals(client, codebase_id).await; @@ -187,6 +193,9 @@ pub async fn sync_status( if let Some(phase) = first_index { write!(out, "\n{}", first_index_line(phase)).expect("writing to a String cannot fail"); } + if let Some(progress) = sync_progress { + write!(out, "\n{}", current_sync_line(progress)).expect("writing to a String cannot fail"); + } write!(out, "\nlocal checkout watch: {watch}\ntotal indexed state:") .expect("writing to a String cannot fail"); match totals { @@ -241,6 +250,12 @@ fn first_index_line(phase: &FirstIndexPhase) -> String { } } +/// The `current sync:` line of `sync_status`. The milestone text is the one +/// `semctl index` shows, so both commands use the same words. +fn current_sync_line(progress: &SyncProgress) -> String { + format!("current sync: {progress}") +} + async fn catalog_totals(client: &Client, codebase_id: &str) -> anyhow::Result<(u32, i64)> { let mut page = 0u32; let mut bytes = 0i64; diff --git a/src/query/inspection/tests.rs b/src/query/inspection/tests.rs index f995b0b..2533e06 100644 --- a/src/query/inspection/tests.rs +++ b/src/query/inspection/tests.rs @@ -1,12 +1,13 @@ -//! `sync_status` text for the phases of a first index. +//! `sync_status` text for the phases of a first index and for a running sync. //! //! The tests use a loopback stub that answers every request with an empty //! catalog page, so no test needs a server or a network. -use super::{first_index_line, sync_status}; +use super::{current_sync_line, first_index_line, sync_status}; use crate::client::Client; use crate::client::stub::{Reply, Stub}; use crate::mcp::readiness::FirstIndexPhase; +use crate::sync::SyncProgress; const EMPTY_CATALOG_PAGE: &str = r#"{"success":true,"data":[],"page":0,"pageSize":1000,"total":0}"#; @@ -70,7 +71,7 @@ async fn sync_status_reports_every_phase_after_the_codebase_line() { reason: "embedding job 7 failed".to_string(), }, ] { - let text = sync_status(&client, None, true, Some(&phase)) + let text = sync_status(&client, None, true, Some(&phase), None) .await .expect("sync_status works while a first index is pending"); let lines: Vec<&str> = text.lines().collect(); @@ -85,10 +86,85 @@ async fn sync_status_reports_every_phase_after_the_codebase_line() { async fn sync_status_has_no_first_index_line_for_a_checkout_without_one() { let (_stub, client) = catalog_stub().await; - let text = sync_status(&client, None, true, None) + let text = sync_status(&client, None, true, None, None) .await .expect("sync_status answers"); assert!(!text.contains("first index"), "{text}"); assert_eq!(text.lines().nth(1), Some("local checkout watch: active")); } + +#[test] +fn the_current_sync_line_names_the_milestone_in_the_shared_words() { + assert_eq!( + current_sync_line(&SyncProgress::Uploading { + uploaded_files: 1200, + total_files: 3607, + }), + "current sync: uploading 1200/3607 files" + ); + assert_eq!( + current_sync_line(&SyncProgress::Planning { + files: 3607, + cached_files: 3500, + }), + "current sync: scanned 3607 files (3500 filter decisions reused) — checking for changes" + ); +} + +/// The phase says that the first index syncs. The next line says how far the +/// sync has come, so the two lines stay together. +#[tokio::test] +async fn sync_status_puts_the_current_sync_line_after_the_first_index_line() { + let (_stub, client) = catalog_stub().await; + let progress = SyncProgress::Uploading { + uploaded_files: 1200, + total_files: 3607, + }; + + let text = sync_status( + &client, + None, + true, + Some(&FirstIndexPhase::Syncing), + Some(&progress), + ) + .await + .expect("sync_status answers while a first index syncs"); + let lines: Vec<&str> = text.lines().collect(); + + assert_eq!(lines[0], "codebase A", "{text}"); + assert_eq!(lines[1], "first index: syncing", "{text}"); + assert_eq!( + lines[2], "current sync: uploading 1200/3607 files", + "{text}" + ); + assert_eq!(lines[3], "local checkout watch: active", "{text}"); +} + +/// A re-sync of an indexed checkout has no first-index gate. Its progress line +/// then follows the codebase line. +#[tokio::test] +async fn sync_status_reports_a_running_resync_without_a_first_index() { + let (_stub, client) = catalog_stub().await; + + let text = sync_status(&client, None, true, None, Some(&SyncProgress::Finalizing)) + .await + .expect("sync_status answers during a re-sync"); + let lines: Vec<&str> = text.lines().collect(); + + assert_eq!(lines[0], "codebase A", "{text}"); + assert_eq!(lines[1], "current sync: finalizing upload", "{text}"); + assert_eq!(lines[2], "local checkout watch: active", "{text}"); +} + +#[tokio::test] +async fn sync_status_has_no_current_sync_line_when_no_sync_runs() { + let (_stub, client) = catalog_stub().await; + + let text = sync_status(&client, None, true, Some(&FirstIndexPhase::Ready), None) + .await + .expect("sync_status answers"); + + assert!(!text.contains("current sync"), "{text}"); +} diff --git a/src/query/mod.rs b/src/query/mod.rs index af75e96..9c5f6b2 100644 --- a/src/query/mod.rs +++ b/src/query/mod.rs @@ -252,8 +252,9 @@ async fn catalog_hashes(client: &Client) -> anyhow::Result Option { let content = std::fs::read_to_string(path).ok()?; Some(blake3::hash(content.as_bytes()).to_hex().to_string()) @@ -717,7 +718,8 @@ mod tests { let p = dir.path().join("f.rs"); std::fs::write(&p, "fn main() {}\n").unwrap(); - // Must equal exactly what `crate::sync::sync` writes to the manifest. + // Must equal exactly what `crate::sync::sync_with_progress` writes to + // the manifest. let want = blake3::hash("fn main() {}\n".as_bytes()) .to_hex() .to_string(); diff --git a/src/sync/mod.rs b/src/sync/mod.rs index 6b6c6fd..ac3dcd8 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -16,15 +16,19 @@ mod cache; pub(crate) mod policy; mod scan; mod source; +#[cfg(test)] +mod tests; mod upload; pub(crate) mod walker; pub(crate) use cache::SyncCache; +use std::fmt; use std::path::{Path, PathBuf}; use std::sync::Arc; use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; use tokio::sync::{Mutex, OwnedSemaphorePermit, Semaphore}; use tracing::{debug, warn}; @@ -54,7 +58,7 @@ impl SyncLimits { } } -/// What a [`sync`] queued, for the caller to report on. +/// What [`sync_with_progress`] queued, for the caller to report on. pub struct SyncOutcome { pub codebase_id: String, pub job_id: String, @@ -73,10 +77,13 @@ impl SyncOutcome { } } -/// A user-visible milestone from a one-shot sync. Background indexing uses the -/// quiet [`sync`] wrapper; interactive callers can use [`sync_with_progress`] -/// to show what is happening before the server-side embed job is queued. -#[derive(Clone, Debug, PartialEq, Eq)] +/// A user-visible milestone of a sync, from [`sync_with_progress`]. +/// +/// The coordinator keeps the latest milestone of the reconcile that is +/// running, and the daemon status carries it, so this is also a wire type. +/// [`Display`](fmt::Display) is the one source of its human text. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum SyncProgress { Preparing, Scanning { @@ -93,6 +100,27 @@ pub enum SyncProgress { Finalizing, } +impl fmt::Display for SyncProgress { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Preparing => formatter.write_str("preparing index"), + Self::Scanning { root } => write!(formatter, "scanning files in {}", root.display()), + Self::Planning { + files, + cached_files, + } => write!( + formatter, + "scanned {files} files ({cached_files} filter decisions reused) — checking for changes" + ), + Self::Uploading { + uploaded_files, + total_files, + } => write!(formatter, "uploading {uploaded_files}/{total_files} files"), + Self::Finalizing => formatter.write_str("finalizing upload"), + } + } +} + /// Identifies the index job `sync_status` reports on. One coordinator keeps one /// of these, so status for one checkout never reports another checkout's sync. #[derive(Clone)] @@ -105,21 +133,14 @@ pub(crate) struct LastJob { /// the caller decides whether to poll it. Shared by the `semctl index` command /// and the `semctl mcp` startup / watch / periodic auto-index. /// +/// Reports each scan, plan, and completed upload-batch milestone through +/// `on_progress`. A caller that does not show progress passes a closure that +/// ignores it. +/// /// Holds the `cache` lock for the whole call: that gives exclusive cache access /// *and* serializes overlapping syncs to one codebase into one job at a time. /// Step-by-step progress is logged at `debug`; callers emit the `info`-level /// summary so a no-op periodic tick stays quiet. -pub(crate) async fn sync( - client: &Client, - dir: &Path, - cache: &Mutex, - limits: &SyncLimits, -) -> Result { - sync_with_progress(client, dir, cache, limits, |_| {}).await -} - -/// The interactive form of [`sync`], reporting scan, plan, and completed -/// upload-batch progress through `on_progress`. pub(crate) async fn sync_with_progress( client: &Client, dir: &Path, diff --git a/src/sync/tests.rs b/src/sync/tests.rs new file mode 100644 index 0000000..070537d --- /dev/null +++ b/src/sync/tests.rs @@ -0,0 +1,75 @@ +//! The text and the wire shape of a sync milestone. + +use std::path::PathBuf; + +use super::SyncProgress; + +/// `semctl index`, `sync_status`, and `semctl daemon status` all show this +/// text, so one test pins every milestone. +#[test] +fn each_milestone_has_one_plain_line() { + for (progress, line) in [ + (SyncProgress::Preparing, "preparing index"), + ( + SyncProgress::Scanning { + root: PathBuf::from("repo"), + }, + "scanning files in repo", + ), + ( + SyncProgress::Planning { + files: 80, + cached_files: 73, + }, + "scanned 80 files (73 filter decisions reused) — checking for changes", + ), + ( + SyncProgress::Uploading { + uploaded_files: 3, + total_files: 8, + }, + "uploading 3/8 files", + ), + (SyncProgress::Finalizing, "finalizing upload"), + ] { + assert_eq!(progress.to_string(), line); + } +} + +/// An older daemon and a newer client must agree on these tags. +#[test] +fn milestones_use_snake_case_tags_on_the_wire() { + for (progress, wire) in [ + (SyncProgress::Preparing, serde_json::json!("preparing")), + ( + SyncProgress::Scanning { + root: PathBuf::from("/work/checkout"), + }, + serde_json::json!({"scanning": {"root": "/work/checkout"}}), + ), + ( + SyncProgress::Planning { + files: 80, + cached_files: 73, + }, + serde_json::json!({"planning": {"files": 80, "cached_files": 73}}), + ), + ( + SyncProgress::Uploading { + uploaded_files: 3, + total_files: 8, + }, + serde_json::json!({"uploading": {"uploaded_files": 3, "total_files": 8}}), + ), + (SyncProgress::Finalizing, serde_json::json!("finalizing")), + ] { + assert_eq!( + serde_json::to_value(&progress).expect("serialize the milestone"), + wire + ); + assert_eq!( + serde_json::from_value::(wire).expect("decode the milestone"), + progress + ); + } +}