From a93c9be1a950dd10c860425422897b84e1dfb257 Mon Sep 17 00:00:00 2001 From: kyle-elliott <77862648+kyle-elliott@users.noreply.github.com> Date: Fri, 9 Oct 2026 23:18:26 +0000 Subject: [PATCH] feat(mcp): return from index_codebase without waiting for the first index index_codebase now returns as soon as the codebase is registered and the first index is handed to the checkout's coordinator. A concurrent call reports the first index as already in progress, and a completed gate is reported as complete. A failed gate is still retried. No index_codebase path waits on the first-index gate any more. sync_status reports the first-index phase (registering, syncing, embedding with its job, ready, or failed with a reason). The phase is derived from the gate state and the coordinator's last job, is part of CoordinatorStatus (older daemons decode as none), and shows in semctl daemon status. The server manual, tool docs, README, retrieval skill, and skill evals describe the non-blocking tool, the sync_status polling loop, and the rule to follow the next: line of a failed tool call. Closes #37 --- README.md | 8 +- .../semctx/skills/codebase-retrieval/SKILL.md | 23 +- .../skills/codebase-retrieval/evals.json | 16 +- src/daemon/status.rs | 18 ++ src/engine/coordinator.rs | 17 +- src/engine/coordinator/tests.rs | 78 +++++- src/mcp/docs/instructions/server.md | 25 +- src/mcp/docs/tools/index_codebase.md | 6 +- src/mcp/docs/tools/sync_status.md | 10 + src/mcp/indexing.rs | 158 ++++++++---- src/mcp/indexing/tests.rs | 231 ++++++++++++++++++ src/mcp/readiness.rs | 71 +++++- src/mcp/readiness/tests.rs | 118 ++++++++- src/mcp/tool_types.rs | 4 +- src/mcp/tools.rs | 3 + src/query/inspection.rs | 27 +- src/query/inspection/tests.rs | 94 +++++++ tests/cli_workflows.rs | 83 ++++--- 18 files changed, 873 insertions(+), 117 deletions(-) create mode 100644 src/mcp/indexing/tests.rs create mode 100644 src/query/inspection/tests.rs diff --git a/README.md b/README.md index 24d1c6f..c7315a6 100644 --- a/README.md +++ b/README.md @@ -130,6 +130,12 @@ most 5 seconds for a first index before it reports that the index is still running. The edit actions have no deadline: a limit that fired after the files 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. + ### Coming from the old `semctx` CLI The first `semctl install` automatically retires a previous `semctx` install: it @@ -175,7 +181,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 +semctl daemon status # version, pid, uptime, sessions, permits, checkouts and their first-index phase 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 0e4084d..8330526 100644 --- a/plugins/semctx/skills/codebase-retrieval/SKILL.md +++ b/plugins/semctx/skills/codebase-retrieval/SKILL.md @@ -45,11 +45,14 @@ for the rest of the MCP session. A previously indexed repo already has consent for retrieval and watcher activation, so use it without asking again; that consent does not authorize edits outside the user's requested repository set. Only when a repo has never been indexed should you tell the user and call `index_codebase` -after explicit opt-in. That first-ever index is gated: -retrieval/catalog/graph tools never serve a partial index. They wait at most 5 -seconds, then fail with a "still running" error. Use `sync_status` to follow -progress and local Read/Grep until the index completes. Later re-syncs do not -block use of the last complete index. +after explicit opt-in. That first-ever index runs in the background: +`index_codebase` returns at once and never waits for embedding. Retrieval, +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. ## Rules @@ -85,9 +88,13 @@ block use of the last complete index. ## Degraded states — read the signals, don't guess -- Empty results right after attach → `sync_status`: `queued` / `running` - means the index is still building; retry shortly instead of concluding the - repo isn't indexed. +- Empty results right after attach → `sync_status`: a first-index phase of + registering, syncing, or embedding, or a job that is `queued` / `running`, + means the index is still building; check again in 10 to 15 seconds instead of + concluding the repo isn't indexed. +- A tool fails → the error ends with a `next:` line. Follow it. Retry once only + when it says the call is retryable. Otherwise use local Read/Grep for that + request. Do not repeat the same failing call in a loop. - A hit flagged `⚠ stale (edited since indexed)`, or a freshness footer → `Read` that file for current bytes. The MCP server auto-syncs edits (filesystem watcher + periodic re-sync); do not tell the user to run diff --git a/plugins/semctx/skills/codebase-retrieval/evals.json b/plugins/semctx/skills/codebase-retrieval/evals.json index 721a04b..f45d612 100644 --- a/plugins/semctx/skills/codebase-retrieval/evals.json +++ b/plugins/semctx/skills/codebase-retrieval/evals.json @@ -37,7 +37,7 @@ "expected_behavior": [ "Uses search_codebase (prefer: docs) or semctx grep — does not skip semctx on the false premise that markdown is not indexed", "Falls back to host tools only on a reported degraded state", - "Identifies that retrieval/catalog/graph tools wait for the initial embed job while sync_status remains available" + "Identifies that index_codebase returns without waiting for embedding, that retrieval/catalog/graph tools wait at most 5 seconds and then report a still-running error, and that sync_status remains available to follow the first-index phase" ] }, { @@ -62,8 +62,9 @@ "skills": ["codebase-retrieval"], "query": "I just opted a brand-new repository into indexing. Can I search it while its first embed job is still running?", "expected_behavior": [ - "Explains that first-time retrieval/catalog/graph calls wait for successful embedding completion and cannot see a partial initial index", - "Uses sync_status to monitor first-index progress because status bypasses the readiness gate", + "Explains that index_codebase returned at once, and that first-time retrieval/catalog/graph calls cannot see a partial initial index: they wait at most 5 seconds, then fail with a still-running error", + "Calls sync_status about every 10 to 15 seconds to follow the first-index phase (registering, syncing, embedding, ready, or failed) because status bypasses the readiness gate", + "Uses local Read/Grep for the repository until the phase is ready instead of looping on a retrieval call that reports it is still running", "Distinguishes later background re-syncs, which do not block retrieval from the last complete snapshot" ] }, @@ -108,6 +109,15 @@ "Uses an immutable codebase id or indexed local directory path for another checkout", "Does not present or reuse a friendly slug as a selector" ] + }, + { + "skills": ["codebase-retrieval"], + "query": "A semctx retrieval tool returned an error that ends with the line 'next: use local Read/Grep for this request.' What do you do next?", + "expected_behavior": [ + "Follows the next: line and answers this request with local Read/Grep instead of calling the same semctx tool again", + "Does not loop on the failing call, and retries once only when the next: line says the call is retryable", + "Does not claim the repository is unindexed or tell the user to reindex because of one failed call" + ] } ], "triggering": { diff --git a/src/daemon/status.rs b/src/daemon/status.rs index 62be9d6..49ca63b 100644 --- a/src/daemon/status.rs +++ b/src/daemon/status.rs @@ -102,6 +102,7 @@ mod tests { use crate::engine::CoordinatorStatus; use crate::engine::coordinator::WatcherState; use crate::engine::scheduler::{PermitUsage, SchedulerUsage}; + use crate::mcp::readiness::FirstIndexPhase; fn status() -> DaemonStatus { DaemonStatus { @@ -134,6 +135,9 @@ mod tests { trigger_overflow: false, last_outcome: Some("uploaded 3 files".to_string()), last_error: None, + first_index: Some(FirstIndexPhase::Embedding { + job_id: Some("job-1".to_string()), + }), }], } } @@ -164,6 +168,12 @@ mod tests { Some("uploaded 3 files") ); assert_eq!(coordinator.last_error, None); + assert_eq!( + coordinator.first_index, + Some(FirstIndexPhase::Embedding { + job_id: Some("job-1".to_string()) + }) + ); } /// A reader that is not this build must find the documented field names. @@ -184,6 +194,10 @@ mod tests { assert_eq!(value["permits"]["scan"]["total"], 8); assert_eq!(value["coordinators"][0]["leases"], 2); assert_eq!(value["coordinators"][0]["watcher"], "active"); + assert_eq!( + value["coordinators"][0]["first_index"], + serde_json::json!({"embedding": {"job_id": "job-1"}}) + ); } #[test] @@ -205,6 +219,10 @@ mod tests { ] ); assert!(lines[8].starts_with(" root /work/checkout"), "{rendered}"); + assert!( + lines[8].contains("first index embedding (job job-1)"), + "{rendered}" + ); } #[test] diff --git a/src/engine/coordinator.rs b/src/engine/coordinator.rs index 17e3073..ec1db0b 100644 --- a/src/engine/coordinator.rs +++ b/src/engine/coordinator.rs @@ -48,7 +48,7 @@ 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::InitialIndexGate; +use crate::mcp::readiness::{FirstIndexPhase, InitialIndexGate}; use crate::sync::policy::{SourcePolicy, event_may_affect_policy}; use crate::sync::{self, LastJob, SyncCache, SyncLimits, SyncOutcome, blocking}; @@ -199,13 +199,17 @@ pub(crate) struct CoordinatorStatus { 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 {} pending {}{} outcome {} error {}", + "root {} codebase {} leases {} watcher {} job {} running {} first index {} pending {}{} outcome {} error {}", self.root.display(), self.codebase_id.as_deref().unwrap_or("(unbound)"), self.leases, @@ -215,6 +219,9 @@ impl std::fmt::Display for CoordinatorStatus { }, 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)" @@ -394,6 +401,11 @@ impl CheckoutCoordinator { .await .as_ref() .map(|job| job.job_id.clone()); + let first_index = if let Some(gate) = self.gate().await { + Some(gate.phase(last_job_id.as_deref()).await) + } else { + None + }; let run = self.run.lock().await; CoordinatorStatus { root: self.root.clone(), @@ -402,6 +414,7 @@ impl CheckoutCoordinator { watcher: self.watcher_state(), last_job_id, running: run.running, + first_index, pending_triggers: self.triggers.max_capacity() - self.triggers.capacity(), trigger_overflow: self.overflow.load(Ordering::Acquire), last_outcome: run.last_outcome.clone(), diff --git a/src/engine/coordinator/tests.rs b/src/engine/coordinator/tests.rs index e7bb609..219d09b 100644 --- a/src/engine/coordinator/tests.rs +++ b/src/engine/coordinator/tests.rs @@ -18,7 +18,7 @@ use super::{ }; use crate::client::{Client, api}; use crate::engine::registry::CheckoutKey; -use crate::mcp::readiness::InitialIndexGate; +use crate::mcp::readiness::{FirstIndexPhase, InitialIndexGate}; use crate::sync::policy; /// Somewhere no test writes, so a coordinator that starts reading the @@ -289,6 +289,82 @@ async fn status_reports_the_last_job_and_the_watcher_state() { assert!(status.to_string().contains("job job")); } +/// Poll the coordinator's status until its first index reaches `expected`. +async fn wait_for_phase(coordinator: &CheckoutCoordinator, expected: FirstIndexPhase) { + tokio::time::timeout(Duration::from_secs(5), async { + while coordinator.status().await.first_index.as_ref() != Some(&expected) { + tokio::task::yield_now().await; + } + }) + .await + .unwrap_or_else(|_| panic!("the first index must reach {expected}")); +} + +/// Nothing waits for a first index any more, so status is the only way to +/// follow it. Each phase comes from the gate and from the last recorded job. +#[tokio::test] +async fn status_follows_a_first_index_through_its_phases() { + let (reconciler, mut runs) = CountingReconciler::new(); + let gate = Arc::new(InitialIndexGate::pending()); + let (coordinator, task) = build(reconciler, Some(gate.clone()), Some(0)).await; + assert_eq!( + coordinator.status().await.first_index, + Some(FirstIndexPhase::Registering) + ); + task.spawn(); + coordinator.trigger(Trigger::Startup); + + gate.register_codebase("codebase".to_string()).await; + runs.next_start().await; + wait_for_phase(&coordinator, FirstIndexPhase::Syncing).await; + + runs.release(1); + let embedding = FirstIndexPhase::Embedding { + job_id: Some("job".to_string()), + }; + wait_for_phase(&coordinator, embedding.clone()).await; + assert!( + coordinator + .status() + .await + .to_string() + .contains("first index embedding (job job)") + ); + + runs.release_polls(1); + wait_for_phase(&coordinator, FirstIndexPhase::Ready).await; +} + +/// A checkout that no `index_codebase` call claimed has no first index to +/// report. +#[tokio::test] +async fn status_has_no_first_index_phase_without_a_gate() { + let (reconciler, _runs) = CountingReconciler::new(); + let (coordinator, _task) = coordinator(reconciler).await; + + let status = coordinator.status().await; + + assert_eq!(status.first_index, None); + assert!(status.to_string().contains("first index (none)")); +} + +/// 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. +#[test] +fn a_status_without_a_first_index_field_decodes_as_none() { + let old = serde_json::json!({ + "root": "/work/checkout", "codebase_id": "codebase", "leases": 1, + "watcher": "active", "last_job_id": null, "running": false, + "pending_triggers": 0, "trigger_overflow": false, + "last_outcome": null, "last_error": null, + }); + + let status: super::CoordinatorStatus = + serde_json::from_value(old).expect("decode a status from an older daemon"); + + assert_eq!(status.first_index, None); +} + /// A second reconcile during embedding sees the same pending gate. It must not /// start a second poll: the poll that finished last would otherwise decide the /// result for every session, and a succeeded first index could turn failed. diff --git a/src/mcp/docs/instructions/server.md b/src/mcp/docs/instructions/server.md index 583d26f..300ec38 100644 --- a/src/mcp/docs/instructions/server.md +++ b/src/mcp/docs/instructions/server.md @@ -25,8 +25,8 @@ Code retrieval and graph navigation over semctx-indexed codebases. Tools default - **external_links** — cross-repo "jump to definition": imports that leave this codebase, resolved into the public API of other codebases you can see. - **list_codebases / current_context** — discover visible codebases and diagnose the effective server/tenant/codebase/checkout/index/graph binding. - **list_domains** — discover what's searchable and which tags are available for filtering. -- **sync_status** — check the background index job; if `list_files` / `search` are empty right after attaching, this tells you whether indexing is still in progress. -- **index_codebase** — explicit opt-in indexing. If startup or a tool says the current repository is not indexed, tell the user and ask permission; call this only after they agree. +- **sync_status** — check the background index job and the first-index phase; if `list_files` / `search` are empty right after attaching, this tells you whether indexing is still in progress. +- **index_codebase** — explicit opt-in indexing. If startup or a tool says the current repository is not indexed, tell the user and ask permission; call this only after they agree. It returns as soon as the first index starts; follow the first index with `sync_status`. - **rename_symbol / safe_delete_symbol / replace_symbol_body / insert_before_symbol / insert_after_symbol** — immediate local edit actions. Each internally consumes the server's grammar-validated plan, verifies source identity, graph generation, paths, and hashes, and goes through the client's normal approval flow. - **undo_edit** — reverse a completed symbolic edit by its returned edit id while retained preimages and current postimage hashes still match. @@ -38,12 +38,15 @@ without asking again; that consent does not authorize edits beyond the user's requested repository set. Ask only when a directory has never been indexed and would need `index_codebase`. -The first-ever `index_codebase` call is a readiness boundary: 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: -call `sync_status` to follow progress, and use local Read/Grep until it -completes. `sync_status` stays callable. Later re-syncs do not block retrieval -and continue to expose the last complete snapshot with freshness warnings. +The first-ever `index_codebase` call starts the first index and returns at once. +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. +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. ## Workflow @@ -68,3 +71,9 @@ Prefer the symbol-graph tools over search whenever you have an exact name — th Expanded search bodies share the server's total result-content budget. `top_k` is an upper bound, not a guarantee that every full body fits. Use expanded bodies directly instead of fetching the same files again. + +## When a tool fails + +A tool failure arrives as an error result with a `next:` line. Follow that line. +Retry once only when it says the call is retryable. Otherwise use local +Read/Grep for that request. Do not repeat the same failing call in a loop. diff --git a/src/mcp/docs/tools/index_codebase.md b/src/mcp/docs/tools/index_codebase.md index 02f5d46..b11d286 100644 --- a/src/mcp/docs/tools/index_codebase.md +++ b/src/mcp/docs/tools/index_codebase.md @@ -2,6 +2,10 @@ Explicitly opt a local directory into semctl indexing and keep it watched for th Call this only after the user has agreed to index a genuinely unindexed directory. Never infer first-time consent from a search request, an unindexed-repository notice, or a failed code tool. A repository that resolves to an existing index already has prior consent and should be synced/watched automatically through normal codebase-scoped tools without asking again. Omit `path` to index the MCP launch directory, or pass an absolute directory path. -For a first-ever index, the tool registers the codebase and starts the scan/upload in a detached task, then waits for the server to finish embedding every file. 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 for progress checks. If the initiating tool call times out, is cut off at the call deadline, or is cancelled, indexing and the readiness gate continue in the MCP server. +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. + +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. For a previously indexed repository, this starts the normal background re-sync/watcher immediately without a first-index gate. diff --git a/src/mcp/docs/tools/sync_status.md b/src/mcp/docs/tools/sync_status.md index e4e63d1..304cc1a 100644 --- a/src/mcp/docs/tools/sync_status.md +++ b/src/mcp/docs/tools/sync_status.md @@ -4,4 +4,14 @@ Omit `codebase` for the launch/current repository, or pass either a codebase id Reports whether a local checkout watcher is active. The total section always reports the catalog's file count and source bytes, so a no-op latest run cannot make an already-populated index look empty. When a latest job is known, it also reports the post-sync total chunk count, phase (`queued` → `running` → `done`, or `failed`), and per-run embedded/deleted/failed progress. While the first local scan is still preparing its server job, status says so rather than claiming nothing is happening. +While a first index is running, or after it ended, the answer has a `first index:` line after the codebase line. Its value is one of these phases: + +- registering: `index_codebase` has not yet registered the codebase. +- syncing: the codebase exists, and the scan and upload run. +- embedding: the upload is done, and the server embeds the files. The line names the server job when it is known. +- ready: the first index is complete, and retrieval tools serve it. +- failed: the first index ended without a usable index. The line gives the reason. Call `index_codebase` for the same path to retry. + +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. + 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/indexing.rs b/src/mcp/indexing.rs index c32242d..f9c71dc 100644 --- a/src/mcp/indexing.rs +++ b/src/mcp/indexing.rs @@ -1,18 +1,31 @@ //! The explicit first index behind the `index_codebase` tool. //! -//! The tool registers a checkout, hands it to the engine, and reports through -//! the checkout's first-index gate. Retrieval tools wait on that same gate. +//! The tool registers a checkout, hands it to the engine, and returns as soon +//! as the codebase exists. The first index then runs in the checkout's +//! coordinator, so no `index_codebase` call waits for it. Retrieval tools wait +//! on the checkout's first-index gate, and `sync_status` reports its phase. use std::{ + future::Future, path::{Path, PathBuf}, sync::Arc, }; +use super::readiness::FirstIndexPhase; use super::tool_types::IndexCodebaseArgs; use super::{ CallBudget, FailureKind, InitialIndexGate, McpServer, ToolError, canonical_directory, initial_gate_for_path, }; +use crate::client::Client; + +/// What every report of a running first index ends with. The agent can follow +/// the first index through `sync_status`, and it can use its own tools until +/// retrieval serves the new index. +const FOLLOW_PROGRESS: &str = "retrieval tools for this path report that the first index is \ + still running until it completes\n\ + call sync_status every 10 to 15 seconds to follow progress, and use local Read/Grep \ + meanwhile"; impl McpServer { /// Index the checkout that `args` names, or the launch checkout. @@ -32,21 +45,34 @@ impl McpServer { // later launches from `repo` incorrectly look unindexed. let dir = crate::codebase::working_copy_root(&dir).await; - // A concurrent or recent first-index call owns the gate. Await it - // rather than queueing another full upload. A gate that FAILED is not - // awaited: it falls through below, where the engine replaces it with a - // fresh one, so a failed first index stays retryable while a partial - // one can never be reported as ready. - if let Some(gate) = initial_gate_for_path(&self.shared.leases, &dir).await - && !matches!(gate.outcome().await, Some(Err(_))) - { - return first_index_report(gate.wait().await, &dir, None); + if let Some(report) = self.earlier_first_index(&dir).await { + return report; } - if let Some(started) = self.resume_recorded_index(&dir, budget).await? { return Ok(started); } - self.start_first_index(&dir).await + self.start_first_index(&dir, ensure_codebase).await + } + + /// The report of a first index that an earlier call already owns on `dir`, + /// or `None` when this call must start one. + /// + /// A concurrent or recent first-index call owns the gate. Report it rather + /// than queue another full upload, and never wait for it. A gate that + /// FAILED is not reported: it falls through, where the engine replaces it + /// with a fresh one, so a failed first index stays retryable while a + /// partial one can never be reported as ready. + async fn earlier_first_index(&self, dir: &Path) -> Option> { + let gate = initial_gate_for_path(&self.shared.leases, dir).await?; + let phase = gate.phase(None).await; + if matches!(phase, FirstIndexPhase::Failed { .. }) { + return None; + } + Some(first_index_report( + &phase, + dir, + gate.codebase().await.as_deref(), + )) } /// Start the background sync and the watcher for a checkout that already @@ -84,8 +110,19 @@ impl McpServer { ))) } - /// Register the checkout as a new codebase and wait for its first index. - async fn start_first_index(&self, dir: &Path) -> Result { + /// Register the checkout as a new codebase, and return once the codebase + /// exists. The first index continues in the checkout's coordinator. + /// + /// `ensure` finds or creates the codebase of the checkout. It is a + /// parameter so a test can register without a server. + async fn start_first_index( + &self, + dir: &Path, + ensure: impl FnOnce(Client, PathBuf) -> F, + ) -> Result + where + F: Future> + Send + 'static, + { // Claim the checkout and its gate before anything is registered on the // server. The coordinator waits for the codebase this call registers // instead of registering one of its own, so one first index creates one @@ -109,10 +146,16 @@ impl McpServer { ) })?; if !gate.claim_registration().await { - return first_index_report(gate.wait().await, dir, None); + // Another call registers this codebase and owns the first index. + return first_index_report( + &gate.phase(None).await, + dir, + gate.codebase().await.as_deref(), + ); } - let id = register_codebase(&self.shared.base, &gate, dir).await?; + let registering = ensure(self.shared.base.clone(), dir.to_path_buf()); + let id = register_codebase(&gate, dir, registering).await?; let client = self .shared .base @@ -122,28 +165,31 @@ impl McpServer { if !self.shared.pinned && self.dir().await == dir { *self.shared.bound.lock().await = Some(client.clone()); } - first_index_report(gate.wait().await, dir, Some(&id)) + Ok(running_report("first index started", dir, Some(&id))) } } -/// Register `dir` on the server and report the result to `gate`. +/// Find or create the codebase of `dir`. `base` has no deadline, because the +/// registration outlives the call that asked for it. +async fn ensure_codebase(base: Client, dir: PathBuf) -> anyhow::Result { + crate::codebase::ensure(&base, &dir).await +} + +/// Run `ensure` and report its result to `gate`. /// /// The registration runs in a task of its own. The call deadline can drop the /// future of the tool while the request is in flight, and a gate that was /// claimed but never finished would hold every later call on this path. The -/// task reports to the gate whatever happens to the call. It uses `base` -/// without a deadline, because it outlives the call. +/// task reports to the gate whatever happens to the call. async fn register_codebase( - base: &crate::client::Client, gate: &Arc, dir: &Path, + ensure: impl Future> + Send + 'static, ) -> Result { let registration = tokio::spawn({ - let base = base.clone(); - let dir = dir.to_path_buf(); let gate = Arc::clone(gate); async move { - match crate::codebase::ensure(&base, &dir).await { + match ensure.await { Ok(id) => { gate.register_codebase(id.clone()).await; Ok(id) @@ -169,30 +215,54 @@ async fn register_codebase( } } -/// The tool's answer for the result of a first index. `codebase` names the -/// codebase when this call registered it. +/// The `codebase ` line of a report, or nothing when the codebase is not +/// known. +fn codebase_line(codebase: Option<&str>) -> String { + codebase.map_or_else(String::new, |id| format!("codebase {id}\n")) +} + +/// The report of a first index that is running. `headline` says whether this +/// call started it or found it. +fn running_report(headline: &str, dir: &Path, codebase: Option<&str>) -> String { + format!( + "{headline}\n{}path {}\n{FOLLOW_PROGRESS}", + codebase_line(codebase), + dir.display() + ) +} + +/// The tool's answer for a first index that another call owns. `codebase` names +/// the codebase when the gate knows it. fn first_index_report( - outcome: Result<(), String>, + phase: &FirstIndexPhase, dir: &Path, codebase: Option<&str>, ) -> Result { - let path = dir.display(); - match (outcome, codebase) { - (Ok(()), None) => Ok(format!( - "initial indexing complete\npath {path}\nretrieval tools are now available" - )), - (Ok(()), Some(id)) => Ok(format!( - "initial indexing complete\ncodebase {id}\npath {path}\nretrieval tools are now available" - )), - (Err(e), None) => Err(ToolError::new( - "index_codebase", - FailureKind::IndexFailed, - format!("initial indexing failed for {path}: {e}"), + match phase { + FirstIndexPhase::Ready => Ok(format!( + "first index complete\n{}path {}\nretrieval tools are available", + codebase_line(codebase), + dir.display() )), - (Err(e), Some(id)) => Err(ToolError::new( - "index_codebase", - FailureKind::IndexFailed, - format!("initial indexing failed for {path} (codebase {id}): {e}"), + FirstIndexPhase::Failed { reason } => { + let path = dir.display(); + let detail = match codebase { + Some(id) => format!("first index failed for {path} (codebase {id}): {reason}"), + None => format!("first index failed for {path}: {reason}"), + }; + Err(ToolError::new( + "index_codebase", + FailureKind::IndexFailed, + detail, + )) + } + running => Ok(running_report( + &format!("first index already in progress ({running})"), + dir, + codebase, )), } } + +#[cfg(test)] +mod tests; diff --git a/src/mcp/indexing/tests.rs b/src/mcp/indexing/tests.rs new file mode 100644 index 0000000..e86502d --- /dev/null +++ b/src/mcp/indexing/tests.rs @@ -0,0 +1,231 @@ +//! `index_codebase` never waits for a first index. +//! +//! The engine of these tests counts reconciles instead of performing them, so +//! a first index stays open until a test finishes its gate. A call that waited +//! for the gate would therefore hit the timeout of the test. + +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use tempfile::TempDir; + +use super::{FirstIndexPhase, first_index_report}; +use crate::client::Client; +use crate::mcp::tests::{first_index, server}; +use crate::mcp::tool_types::IndexCodebaseArgs; +use crate::mcp::{CallBudget, McpServer, initial_gate_for_path}; + +/// Longer than any call without a wait needs. A call that waits for the open +/// first index never ends inside it. +const NO_WAIT: Duration = Duration::from_secs(1); + +/// A checkout on disk, named as `index_codebase` names it. +async fn checkout() -> (TempDir, PathBuf) { + let temp = tempfile::tempdir().expect("temporary checkout"); + let dir = crate::codebase::working_copy_root(temp.path()).await; + (temp, dir) +} + +fn session() -> McpServer { + server(Client::for_test("codebase", None), "launch", false) +} + +fn call_for(dir: &Path) -> IndexCodebaseArgs { + IndexCodebaseArgs { + path: Some(dir.display().to_string()), + } +} + +/// Register `dir` through a registration that succeeds at once. +async fn start(server: &McpServer, dir: &Path) -> Result { + server + .start_first_index(dir, |_, _| async { Ok("codebase-1".to_string()) }) + .await +} + +#[tokio::test] +async fn a_new_first_index_returns_once_the_codebase_is_registered() { + let (_temp, dir) = checkout().await; + let server = session(); + + let text = tokio::time::timeout(NO_WAIT, start(&server, &dir)) + .await + .expect("the call must not wait for the first index") + .expect("the codebase registers"); + + assert!( + text.starts_with("first index started\ncodebase codebase-1\npath "), + "{text}" + ); + assert!( + text.contains("call sync_status every 10 to 15 seconds"), + "{text}" + ); + assert!(text.contains("local Read/Grep"), "{text}"); + let gate = initial_gate_for_path(&server.shared.leases, &dir) + .await + .expect("the session holds the first-index gate"); + assert_eq!(gate.outcome().await, None, "the first index is still open"); + assert_eq!(gate.phase(None).await, FirstIndexPhase::Syncing); +} + +/// `sync_status` reads the coordinator that owns the checkout. It must find +/// the phase of a first index that is still open, and it must not wait. +#[tokio::test] +async fn the_checkout_status_carries_the_phase_of_the_open_first_index() { + let (_temp, dir) = checkout().await; + let server = session(); + start(&server, &dir).await.expect("the codebase registers"); + let client = Client::for_test("codebase-1", Some(dir.clone())); + + let status = tokio::time::timeout(NO_WAIT, server.checkout_status(&client)) + .await + .expect("status must not wait for the first index") + .expect("the session holds the checkout"); + + assert_eq!(status.first_index, Some(FirstIndexPhase::Syncing)); +} + +#[tokio::test] +async fn a_second_call_reports_the_first_index_in_progress_without_waiting() { + let (_temp, dir) = checkout().await; + let server = session(); + start(&server, &dir).await.expect("the codebase registers"); + + let text = tokio::time::timeout( + NO_WAIT, + server.index_checkout(call_for(&dir), &CallBudget::unbounded()), + ) + .await + .expect("a second call must not wait for the first index") + .expect("a first index in progress is not an error"); + + assert!( + text.starts_with("first index already in progress (syncing)\ncodebase codebase-1\npath "), + "{text}" + ); + assert!( + text.contains("call sync_status every 10 to 15 seconds"), + "{text}" + ); +} + +/// Two calls race for one checkout. The one that does not win the claim must +/// not register a second codebase, and it must not wait for the winner. +#[tokio::test] +async fn a_call_that_loses_the_registration_claim_reports_progress() { + let (_temp, dir) = checkout().await; + let server = session(); + let gate = first_index(&server, "codebase", &dir).await; + assert!( + gate.claim_registration().await, + "another call holds the claim" + ); + + let text = tokio::time::timeout( + NO_WAIT, + server.start_first_index(&dir, |_, _| async { + Err(anyhow::anyhow!("a second registration must not run")) + }), + ) + .await + .expect("the call must not wait for the other registration") + .expect("the other call's first index is not an error"); + + assert!( + text.starts_with("first index already in progress (registering)\npath "), + "{text}" + ); + assert!(text.contains("call sync_status"), "{text}"); +} + +#[tokio::test] +async fn a_completed_first_index_is_reported_complete() { + let (_temp, dir) = checkout().await; + let server = session(); + let gate = first_index(&server, "codebase", &dir).await; + gate.register_codebase("codebase-1".into()).await; + gate.finish(Ok(())).await; + + let text = tokio::time::timeout( + NO_WAIT, + server.index_checkout(call_for(&dir), &CallBudget::unbounded()), + ) + .await + .expect("a completed first index needs no wait") + .expect("a completed first index is not an error"); + + assert!( + text.starts_with("first index complete\ncodebase codebase-1\npath "), + "{text}" + ); + assert!(text.ends_with("retrieval tools are available"), "{text}"); +} + +/// A failed first index stays retryable: the call does not answer from the +/// failed gate, so it goes on to start a new first index. +#[tokio::test] +async fn a_failed_first_index_is_not_reported_so_the_call_retries() { + let (_temp, dir) = checkout().await; + let server = session(); + let gate = first_index(&server, "codebase", &dir).await; + + assert!( + server.earlier_first_index(&dir).await.is_some(), + "an open first index answers the call" + ); + + gate.finish(Err("embedding job 7 failed".into())).await; + + assert!( + server.earlier_first_index(&dir).await.is_none(), + "a failed first index must fall through to a retry" + ); +} + +#[tokio::test] +async fn a_failed_registration_is_a_tool_error_and_fails_the_gate() { + let (_temp, dir) = checkout().await; + let server = session(); + + let error = server + .start_first_index(&dir, |_, _| async { + Err(anyhow::anyhow!("the server is unreachable")) + }) + .await + .expect_err("a failed registration is an error"); + + let message = error.to_string(); + assert!(message.starts_with("index_codebase failed: "), "{message}"); + assert!(message.contains("the server is unreachable"), "{message}"); + let gate = initial_gate_for_path(&server.shared.leases, &dir) + .await + .expect("the session holds the first-index gate"); + assert!( + matches!(gate.outcome().await, Some(Err(reason)) if reason.contains("unreachable")), + "the gate must report the failure to every later call" + ); +} + +#[test] +fn a_call_that_finds_a_failed_first_index_reports_it_with_the_retry_step() { + let phase = FirstIndexPhase::Failed { + reason: "embedding job 7 failed".to_string(), + }; + + let message = first_index_report(&phase, Path::new("/work/checkout"), Some("codebase-1")) + .expect_err("a failed first index is an error") + .to_string(); + + assert!( + message.starts_with( + "index_codebase failed: first index failed for /work/checkout \ + (codebase codebase-1): embedding job 7 failed\n" + ), + "{message}" + ); + assert!( + message.ends_with("next: call index_codebase for this path to retry the first index."), + "{message}" + ); +} diff --git a/src/mcp/readiness.rs b/src/mcp/readiness.rs index d6fa732..a35f6ec 100644 --- a/src/mcp/readiness.rs +++ b/src/mcp/readiness.rs @@ -6,10 +6,12 @@ //! this session did bring in is waited for. use std::collections::HashMap; +use std::fmt; use std::path::Path; use std::sync::Arc; use std::time::Duration; +use serde::{Deserialize, Serialize}; use tokio::sync::{Mutex, Notify, RwLock, RwLockReadGuard}; use tokio::time::Instant; @@ -191,6 +193,58 @@ struct InitialIndexState { result: Option>, } +/// How far one first index has come. `sync_status` and the daemon status +/// report it, so an agent can follow a first index that no tool call waits for. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum FirstIndexPhase { + /// `index_codebase` has not yet registered a codebase for the checkout. + Registering, + /// The codebase exists. The scan and the upload run, or wait for a permit. + Syncing, + /// The upload is done and the server embeds the files. `job_id` names the + /// server job when the coordinator has recorded it. + Embedding { job_id: Option }, + /// The server embedded every file. Retrieval serves this index. + Ready, + /// The first index ended without a usable index. Another `index_codebase` + /// call retries it. + Failed { reason: String }, +} + +impl FirstIndexPhase { + /// The phase of a first index, from one snapshot of its gate. + /// + /// `last_job_id` is the coordinator's last recorded job. Once the upload is + /// done, that job is the embedding job of this first index. + fn of(state: &InitialIndexState, last_job_id: Option<&str>) -> Self { + match &state.result { + Some(Ok(())) => Self::Ready, + Some(Err(reason)) => Self::Failed { + reason: reason.clone(), + }, + None if state.codebase_id.is_none() => Self::Registering, + None if !state.polling => Self::Syncing, + None => Self::Embedding { + job_id: last_job_id.map(str::to_string), + }, + } + } +} + +impl fmt::Display for FirstIndexPhase { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Registering => formatter.write_str("registering"), + Self::Syncing => formatter.write_str("syncing"), + Self::Embedding { job_id: Some(id) } => write!(formatter, "embedding (job {id})"), + Self::Embedding { job_id: None } => formatter.write_str("embedding"), + Self::Ready => formatter.write_str("ready"), + Self::Failed { reason } => write!(formatter, "failed — {reason}"), + } + } +} + impl InitialIndexGate { pub(crate) fn pending() -> Self { Self { @@ -212,8 +266,9 @@ impl InitialIndexGate { /// Take responsibility for registering this first index's codebase. /// /// Exactly one caller is answered `true` for one gate. Every other caller - /// waits, so one explicit `index_codebase` on a checkout registers one - /// codebase however many callers ask at once. + /// reports the first index as already in progress, so one explicit + /// `index_codebase` on a checkout registers one codebase however many + /// callers ask at once. pub(crate) async fn claim_registration(&self) -> bool { let mut state = self.state.lock().await; if state.registering { @@ -251,6 +306,18 @@ impl InitialIndexGate { self.state.lock().await.result.clone() } + /// The phase of this first index now. Never waits. + pub(crate) async fn phase(&self, last_job_id: Option<&str>) -> FirstIndexPhase { + let state = self.state.lock().await; + FirstIndexPhase::of(&state, last_job_id) + } + + /// The codebase that `index_codebase` registered for this first index, if + /// it has registered one. Never waits, unlike [`Self::registered_codebase`]. + pub(crate) async fn codebase(&self) -> Option { + self.state.lock().await.codebase_id.clone() + } + /// Wait until the first index has a codebase, or until it ends without one. /// /// The coordinator calls this before its startup reconcile. A first index diff --git a/src/mcp/readiness/tests.rs b/src/mcp/readiness/tests.rs index 1480955..c5f5599 100644 --- a/src/mcp/readiness/tests.rs +++ b/src/mcp/readiness/tests.rs @@ -8,7 +8,10 @@ use std::time::Duration; use tokio::time::Instant; -use super::{FIRST_INDEX_WAIT, NotReady, ready_for_codebases}; +use super::{ + FIRST_INDEX_WAIT, FirstIndexPhase, InitialIndexGate, InitialIndexState, NotReady, + ready_for_codebases, +}; use crate::client::Client; use crate::mcp::CallBudget; use crate::mcp::tests::{first_index, server}; @@ -183,3 +186,116 @@ async fn a_scoped_wait_serves_a_finished_index_and_reports_a_failed_one() { )) ); } + +fn snapshot( + codebase: Option<&str>, + polling: bool, + result: Option>, +) -> InitialIndexState { + InitialIndexState { + registering: true, + polling, + codebase_id: codebase.map(str::to_string), + result, + } +} + +fn embedding(job: Option<&str>) -> FirstIndexPhase { + FirstIndexPhase::Embedding { + job_id: job.map(str::to_string), + } +} + +fn failed(reason: &str) -> FirstIndexPhase { + FirstIndexPhase::Failed { + reason: reason.to_string(), + } +} + +/// The phase is a pure function of one gate snapshot and the coordinator's +/// last job, so each row below is one state that a first index can be in. +#[test] +fn each_gate_state_maps_to_one_phase() { + let cases = [ + // No codebase yet, whether or not a caller claimed the registration. + ( + InitialIndexState::default(), + None, + FirstIndexPhase::Registering, + ), + ( + snapshot(None, false, None), + None, + FirstIndexPhase::Registering, + ), + // A codebase and no poll: the scan and the upload. + ( + snapshot(Some("A"), false, None), + None, + FirstIndexPhase::Syncing, + ), + // A job of an earlier attempt is not the job of this upload. + ( + snapshot(Some("A"), false, None), + Some("earlier-job"), + FirstIndexPhase::Syncing, + ), + // A poll claimed: the server embeds the files. + ( + snapshot(Some("A"), true, None), + Some("job-1"), + embedding(Some("job-1")), + ), + (snapshot(Some("A"), true, None), None, embedding(None)), + // A result ends the first index, whatever the other fields say. + ( + snapshot(Some("A"), true, Some(Ok(()))), + Some("job-1"), + FirstIndexPhase::Ready, + ), + ( + snapshot(None, false, Some(Err("registration failed".into()))), + None, + failed("registration failed"), + ), + ( + snapshot(Some("A"), true, Some(Err("embedding failed".into()))), + Some("job-1"), + failed("embedding failed"), + ), + ]; + + for (state, last_job_id, expected) in cases { + assert_eq!(FirstIndexPhase::of(&state, last_job_id), expected); + } +} + +#[tokio::test] +async fn a_gate_reports_each_phase_as_the_first_index_advances() { + let gate = InitialIndexGate::pending(); + assert_eq!(gate.phase(None).await, FirstIndexPhase::Registering); + assert_eq!(gate.codebase().await, None); + + gate.register_codebase("A".into()).await; + assert_eq!(gate.phase(None).await, FirstIndexPhase::Syncing); + assert_eq!(gate.codebase().await.as_deref(), Some("A")); + + assert!(gate.claim_poll().await); + assert_eq!(gate.phase(Some("job-1")).await, embedding(Some("job-1"))); + + gate.finish(Ok(())).await; + assert_eq!(gate.phase(Some("job-1")).await, FirstIndexPhase::Ready); +} + +/// The phase never waits. A tool call that asks for it while the first index +/// runs must come back at once. +#[tokio::test(start_paused = true)] +async fn asking_for_the_phase_does_not_wait_for_the_first_index() { + let gate = InitialIndexGate::pending(); + + let phase = tokio::time::timeout(Duration::from_millis(1), gate.phase(None)) + .await + .expect("the phase is read without waiting"); + + assert_eq!(phase, FirstIndexPhase::Registering); +} diff --git a/src/mcp/tool_types.rs b/src/mcp/tool_types.rs index fed8bff..129d861 100644 --- a/src/mcp/tool_types.rs +++ b/src/mcp/tool_types.rs @@ -456,7 +456,7 @@ pub struct ListFilesArgs { pub struct IndexCodebaseArgs { /// Local directory to register and index. Omit for the launch/current /// directory. Calling this tool is the explicit indexing opt-in; agents must - /// ask the user before calling it. A first-ever index waits for embedding to - /// complete before retrieval tools are released. + /// ask the user before calling it. A first-ever index runs in the + /// background: the call returns at once, and `sync_status` reports progress. pub path: Option, } diff --git a/src/mcp/tools.rs b/src/mcp/tools.rs index f3c8f04..3ba06b7 100644 --- a/src/mcp/tools.rs +++ b/src/mcp/tools.rs @@ -762,6 +762,9 @@ impl McpServer { .as_ref() .and_then(|status| status.last_job_id.as_deref()), watching, + status + .as_ref() + .and_then(|status| status.first_index.as_ref()), ) .await } diff --git a/src/query/inspection.rs b/src/query/inspection.rs index ad0946f..330376c 100644 --- a/src/query/inspection.rs +++ b/src/query/inspection.rs @@ -7,6 +7,7 @@ use super::render::{ truncation_note, }; use super::{Client, ToolError, api, bound_codebase, urlencode}; +use crate::mcp::readiness::FirstIndexPhase; pub(super) const FILES_PAGE_MAX: u32 = 1000; @@ -156,10 +157,14 @@ pub async fn list_files( /// The catalog totals are deliberately independent of the latest job: a no-op /// sync has a 0-file plan even when the codebase already contains thousands of /// indexed 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. pub async fn sync_status( client: &Client, job_id: Option<&str>, local_watch_active: bool, + first_index: Option<&FirstIndexPhase>, ) -> Result { let codebase_id = bound_codebase(client, "sync_status")?; let totals = catalog_totals(client, codebase_id).await; @@ -178,8 +183,12 @@ pub async fn sync_status( } else { "not active (no local checkout selected)" }; - let mut out = - format!("codebase {codebase_id}\nlocal checkout watch: {watch}\ntotal indexed state:"); + let mut out = format!("codebase {codebase_id}"); + if let Some(phase) = first_index { + write!(out, "\n{}", first_index_line(phase)).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 { Ok((files, bytes)) => { write!( @@ -221,6 +230,17 @@ pub async fn sync_status( Ok(out) } +/// The `first index:` line of `sync_status`. A failed first index also names +/// the call that retries it. +fn first_index_line(phase: &FirstIndexPhase) -> String { + match phase { + FirstIndexPhase::Failed { .. } => { + format!("first index: {phase}; call index_codebase to retry") + } + _ => format!("first index: {phase}"), + } +} + async fn catalog_totals(client: &Client, codebase_id: &str) -> anyhow::Result<(u32, i64)> { let mut page = 0u32; let mut bytes = 0i64; @@ -470,3 +490,6 @@ pub async fn list_domains(client: &Client) -> Result { Err(e) => Err(ToolError::from_client("list_domains", &e)), } } + +#[cfg(test)] +mod tests; diff --git a/src/query/inspection/tests.rs b/src/query/inspection/tests.rs new file mode 100644 index 0000000..f995b0b --- /dev/null +++ b/src/query/inspection/tests.rs @@ -0,0 +1,94 @@ +//! `sync_status` text for the phases of a first index. +//! +//! 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 crate::client::Client; +use crate::client::stub::{Reply, Stub}; +use crate::mcp::readiness::FirstIndexPhase; + +const EMPTY_CATALOG_PAGE: &str = r#"{"success":true,"data":[],"page":0,"pageSize":1000,"total":0}"#; + +async fn catalog_stub() -> (Stub, Client) { + let stub = Stub::serve(vec![Reply::Answer { + status: "200 OK", + headers: Vec::new(), + body: EMPTY_CATALOG_PAGE.to_string(), + }]) + .await; + let client = Client::for_test_server(&stub.url).with_codebase("A".to_string()); + (stub, client) +} + +#[test] +fn each_phase_has_one_plain_line() { + for (phase, line) in [ + (FirstIndexPhase::Registering, "first index: registering"), + (FirstIndexPhase::Syncing, "first index: syncing"), + ( + FirstIndexPhase::Embedding { + job_id: Some("job-7".to_string()), + }, + "first index: embedding (job job-7)", + ), + ( + FirstIndexPhase::Embedding { job_id: None }, + "first index: embedding", + ), + (FirstIndexPhase::Ready, "first index: ready"), + ] { + assert_eq!(first_index_line(&phase), line); + } +} + +#[test] +fn a_failed_first_index_names_the_reason_and_the_retry() { + let phase = FirstIndexPhase::Failed { + reason: "embedding job 7 failed".to_string(), + }; + + assert_eq!( + first_index_line(&phase), + "first index: failed — embedding job 7 failed; call index_codebase to retry" + ); +} + +/// An agent reads the top of the answer first, and nothing waits for a first +/// index. The phase therefore comes straight after the codebase line. +#[tokio::test] +async fn sync_status_reports_every_phase_after_the_codebase_line() { + let (_stub, client) = catalog_stub().await; + for phase in [ + FirstIndexPhase::Registering, + FirstIndexPhase::Syncing, + FirstIndexPhase::Embedding { + job_id: Some("job-7".to_string()), + }, + FirstIndexPhase::Ready, + FirstIndexPhase::Failed { + reason: "embedding job 7 failed".to_string(), + }, + ] { + let text = sync_status(&client, None, true, Some(&phase)) + .await + .expect("sync_status works while a first index is pending"); + let lines: Vec<&str> = text.lines().collect(); + + assert_eq!(lines[0], "codebase A", "{text}"); + assert_eq!(lines[1], first_index_line(&phase), "{text}"); + assert_eq!(lines[2], "local checkout watch: active", "{text}"); + } +} + +#[tokio::test] +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) + .await + .expect("sync_status answers"); + + assert!(!text.contains("first index"), "{text}"); + assert_eq!(text.lines().nth(1), Some("local checkout watch: active")); +} diff --git a/tests/cli_workflows.rs b/tests/cli_workflows.rs index 7f2caf7..9c58ec5 100644 --- a/tests/cli_workflows.rs +++ b/tests/cli_workflows.rs @@ -714,10 +714,26 @@ fn readiness_server( }) } -#[tokio::test] -async fn scoped_mcp_search_waits_for_initial_embedding_and_propagates_failure() { - use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; +type ServerLines = tokio::io::Lines>; + +/// The next message the server writes. A message that never comes fails the +/// test after ten seconds instead of hanging it. +async fn next_message(stdout: &mut ServerLines, what: &str) -> (String, Value) { + let next = tokio::time::timeout(Duration::from_secs(10), stdout.next_line()); + let line = next.await.expect(what).unwrap().unwrap(); + let message = serde_json::from_str(&line).unwrap(); + (line, message) +} +async fn send(stdin: &mut tokio::process::ChildStdin, message: &Value) { + use tokio::io::AsyncWriteExt; + let line = format!("{message}\n"); + stdin.write_all(line.as_bytes()).await.unwrap(); +} + +#[tokio::test] +async fn index_codebase_returns_at_once_and_scoped_search_waits_for_the_first_embedding() { + use tokio::io::AsyncBufReadExt; for failed in [false, true] { let checkout = Checkout::new(); fs::write(checkout.config.join("semctl/config.toml"), "").unwrap(); @@ -739,26 +755,23 @@ async fn scoped_mcp_search_waits_for_initial_embedding_and_propagates_failure() "capabilities": {}, "clientInfo": {"name": "readiness-test", "version": "1"} } }); - stdin - .write_all(format!("{initialize}\n").as_bytes()) - .await - .unwrap(); - tokio::time::timeout(Duration::from_secs(10), stdout.next_line()) - .await - .unwrap() - .unwrap() - .unwrap(); - stdin - .write_all(b"{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n") - .await - .unwrap(); + send(&mut stdin, &initialize).await; + next_message(&mut stdout, "initialize must answer").await; + send( + &mut stdin, + &json!({"jsonrpc": "2.0", "method": "notifications/initialized"}), + ) + .await; let index = json!({"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "index_codebase", "arguments": {} }}); - stdin - .write_all(format!("{index}\n").as_bytes()) - .await - .unwrap(); + send(&mut stdin, &index).await; + // The embedding job is still open here, so this answer cannot wait for it. + let (started, response) = + next_message(&mut stdout, "index_codebase must not wait for embedding").await; + assert_eq!(response["id"], 2, "{started}"); + assert_eq!(response["result"]["isError"], false, "{started}"); + assert!(started.contains("call sync_status"), "{started}"); tokio::time::timeout(Duration::from_secs(10), polled.notified()) .await .expect("initial job must start"); @@ -770,37 +783,23 @@ async fn scoped_mcp_search_waits_for_initial_embedding_and_propagates_failure() let search = json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": { "name": "search_codebase", "arguments": arguments }}); - stdin - .write_all(format!("{search}\n").as_bytes()) - .await - .unwrap(); + send(&mut stdin, &search).await; } - let unrelated = tokio::time::timeout(Duration::from_secs(10), stdout.next_line()) - .await - .expect("unrelated search must remain available") - .unwrap() - .unwrap(); - assert_eq!( - serde_json::from_str::(&unrelated).unwrap()["id"], - 5, - "{unrelated}" - ); + let (unrelated, response) = + next_message(&mut stdout, "unrelated search must remain available").await; + assert_eq!(response["id"], 5, "{unrelated}"); assert_eq!(server.request_count("/v1/search"), 1); complete.store(true, Ordering::Release); let mut ids = Vec::new(); - for _ in 0..3 { - let line = tokio::time::timeout(Duration::from_secs(10), stdout.next_line()) - .await - .expect("initial index must release waiting tools") - .unwrap() - .unwrap(); - let response: Value = serde_json::from_str(&line).unwrap(); + for _ in 0..2 { + let (line, response) = + next_message(&mut stdout, "initial index must release waiting tools").await; ids.push(response["id"].as_u64().unwrap()); assert_eq!(line.contains("embedding failed"), failed, "{line}"); assert_eq!(response["result"]["isError"], failed, "{line}"); } ids.sort_unstable(); - assert_eq!(ids, [2, 3, 4]); + assert_eq!(ids, [3, 4]); drop(stdin); let output = tokio::time::timeout(Duration::from_secs(30), child.wait_with_output()) .await