diff --git a/Cargo.lock b/Cargo.lock index 67d7297..eda7f82 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -606,6 +606,15 @@ version = "2.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6" +[[package]] +name = "file-id" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1fc6a637b6dc58414714eddd9170ff187ecb0933d4c7024d1abbd23a3cc26e9" +dependencies = [ + "windows-sys 0.60.2", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -652,6 +661,15 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "futures" version = "0.3.32" @@ -833,9 +851,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68df315d2857b2d8d2898be54a85e1d001bbbe0dbb5f8ef847b48dd3a23c4527" dependencies = [ "cfg-if", - "nix", + "nix 0.30.1", "widestring", - "windows", + "windows 0.61.3", ] [[package]] @@ -1114,6 +1132,26 @@ dependencies = [ "serde_core", ] +[[package]] +name = "inotify" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "533e68a5842e734946fe159fb03fc9bbbb254f590dd0d8ad321ae5ff7beca2c1" +dependencies = [ + "bitflags", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e05c02b5e89bff3b946cedeca278abc628fe811e604f027c45a8aa3cf793d0eb" +dependencies = [ + "libc", +] + [[package]] name = "ipnet" version = "2.12.0" @@ -1202,6 +1240,26 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "kqueue" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "273c0752728918e0ac4976f2b275b6fefb9ecd400585dec929419f3844cd87b5" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags", + "libc", +] + [[package]] name = "lazy_static" version = "1.5.0" @@ -1309,6 +1367,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -1335,12 +1394,64 @@ dependencies = [ "libc", ] +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "normalize-line-endings" version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "61807f77802ff30975e01f4f071c8ba10c022052f98b3294119f3e615d13e5be" +[[package]] +name = "notify" +version = "8.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3" +dependencies = [ + "bitflags", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio", + "notify-types", + "walkdir", + "windows-sys 0.60.2", +] + +[[package]] +name = "notify-debouncer-full" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c02b49179cfebc9932238d04d6079912d26de0379328872846118a0fa0dbb302" +dependencies = [ + "file-id", + "log", + "notify", + "notify-types", + "walkdir", +] + +[[package]] +name = "notify-types" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a" +dependencies = [ + "bitflags", +] + [[package]] name = "num-bigint" version = "0.4.6" @@ -1399,12 +1510,13 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" [[package]] name = "oxcode-cli" -version = "0.3.0" +version = "0.4.0" dependencies = [ "anyhow", "assert_cmd", "axoupdater", "clap", + "notify-debouncer-full", "oxcode-core", "predicates", "rmcp", @@ -1416,7 +1528,7 @@ dependencies = [ [[package]] name = "oxcode-core" -version = "0.3.0" +version = "0.4.0" dependencies = [ "ignore", "oxcode-model", @@ -1451,7 +1563,7 @@ dependencies = [ [[package]] name = "oxcode-model" -version = "0.3.0" +version = "0.4.0" dependencies = [ "serde", "serde_json", @@ -1716,6 +1828,20 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "process-wrap" +version = "9.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e842efad9119158434d193c6682e2ebee4b44d6ad801d7b349623b3f57cdf55" +dependencies = [ + "futures", + "indexmap", + "nix 0.31.3", + "tokio", + "tracing", + "windows 0.62.2", +] + [[package]] name = "pxfm" version = "0.1.29" @@ -1949,6 +2075,7 @@ dependencies = [ "futures", "pastey", "pin-project-lite", + "process-wrap", "rmcp-macros", "schemars", "serde", @@ -3056,11 +3183,23 @@ version = "0.61.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893" dependencies = [ - "windows-collections", + "windows-collections 0.2.0", "windows-core 0.61.2", - "windows-future", + "windows-future 0.2.1", "windows-link 0.1.3", - "windows-numerics", + "windows-numerics 0.2.0", +] + +[[package]] +name = "windows" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580" +dependencies = [ + "windows-collections 0.3.2", + "windows-core 0.62.2", + "windows-future 0.3.2", + "windows-numerics 0.3.1", ] [[package]] @@ -3072,6 +3211,15 @@ dependencies = [ "windows-core 0.61.2", ] +[[package]] +name = "windows-collections" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610" +dependencies = [ + "windows-core 0.62.2", +] + [[package]] name = "windows-core" version = "0.61.2" @@ -3106,7 +3254,18 @@ checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e" dependencies = [ "windows-core 0.61.2", "windows-link 0.1.3", - "windows-threading", + "windows-threading 0.1.0", +] + +[[package]] +name = "windows-future" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb" +dependencies = [ + "windows-core 0.62.2", + "windows-link 0.2.1", + "windows-threading 0.2.1", ] [[package]] @@ -3153,6 +3312,16 @@ dependencies = [ "windows-link 0.1.3", ] +[[package]] +name = "windows-numerics" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26" +dependencies = [ + "windows-core 0.62.2", + "windows-link 0.2.1", +] + [[package]] name = "windows-result" version = "0.3.4" @@ -3258,6 +3427,15 @@ dependencies = [ "windows-link 0.1.3", ] +[[package]] +name = "windows-threading" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37" +dependencies = [ + "windows-link 0.2.1", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" diff --git a/Cargo.toml b/Cargo.toml index 6884afc..5d8712d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -28,7 +28,7 @@ serde_json = "1.0.145" streaming-iterator = "0.1.9" tempfile = "3.23.0" thiserror = "2.0.17" -tokio = { version = "1.52", features = ["rt-multi-thread", "macros", "io-std", "sync"] } +tokio = { version = "1.52", features = ["rt-multi-thread", "macros", "io-std", "sync", "time"] } toml = "1.1.2" tree-sitter = "0.26" diff --git a/claude-plugin/.claude-plugin/plugin.json b/claude-plugin/.claude-plugin/plugin.json index 987e130..4b40469 100644 --- a/claude-plugin/.claude-plugin/plugin.json +++ b/claude-plugin/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "oxcode", "version": "0.4.0", - "description": "PageRank-curated code intelligence for coding agents. Bundles the oxcode MCP server over stdio (self-updating from v0.1.2): oxcode_index builds/refreshes the index; explore/search/callers/callees/symbol/files/status query it. explore also surfaces n-ary impl/module hyperedges and a crate dependency graph for architecture-altitude context.", + "description": "PageRank-curated code intelligence for coding agents. Bundles the oxcode MCP server over stdio (self-updating from v0.1.2): oxcode_watch builds the index and keeps it current as files change — a file lock elects a single writer per folder while other instances serve reads; explore/search/callers/callees/symbol/files/status query it. explore also surfaces n-ary impl/module hyperedges and a crate dependency graph for architecture-altitude context.", "author": { "name": "oxgraph", "email": "michael@snowmead.com" }, "homepage": "https://oxgraph.io", "repository": "https://github.com/oxgraph/oxcode", diff --git a/claude-plugin/README.md b/claude-plugin/README.md index 3988a0a..7d951f4 100644 --- a/claude-plugin/README.md +++ b/claude-plugin/README.md @@ -3,32 +3,39 @@ Bundles the **oxcode MCP server** so coding agents get PageRank-curated code intelligence over your indexed repository with no manual `.mcp.json` wiring. -Once enabled, the plugin starts `oxcode mcp` over stdio and exposes eight tools -(surfaced as `mcp__oxcode__`) — seven read-only queries plus `oxcode_index`, -which builds or refreshes the index: +Once enabled, the plugin starts `oxcode mcp` over stdio and exposes one watcher +tool plus seven read-only query tools (surfaced as `mcp__oxcode__`). Call +`oxcode_watch` once and the index is built and kept current as files change — no +manual index command: | Tool | What it answers | | ---------------- | --------------------------------------------------------------------- | -| `oxcode_index` | Build or refresh the index for a project (optional `path`, defaults to cwd). The only tool that writes; run it first if `oxcode_status` reports no database. | -| `oxcode_explore` | One-call answer: top symbols by graph centrality + source, relations, blast radius, call flow. **Use this first.** | +| `oxcode_watch` | Build the index and keep it current as files change. One instance per folder is elected the writer (holds a file lock and re-indexes); others serve reads and take over if the writer exits. **Call this first.** | +| `oxcode_explore` | One-call answer: top symbols by graph centrality + source, relations, blast radius, call flow. | | `oxcode_search` | Search indexed symbols by keyword (optionally by kind). | | `oxcode_callers` | Incoming call graph for a symbol. | | `oxcode_callees` | Outgoing call graph for a symbol. | | `oxcode_symbol` | Describe one symbol by selector (qualified name, `name:`, `element:`, `file::`). | | `oxcode_files` | Search indexed files by keyword. | -| `oxcode_status` | Indexed project's database status (element/relation counts, paths). | +| `oxcode_status` | Database status (element/relation counts, paths) plus this instance's watch role and re-index count. | -`oxcode_index` and `oxcode_explore` declare `taskSupport: "optional"`: a client +`oxcode_watch` and `oxcode_explore` declare `taskSupport: "optional"`: a client that supports MCP tasks (SEP-1686) may run them as background tasks and poll for the result instead of blocking; otherwise they run as ordinary synchronous calls. +Run many agents at once: across all `oxcode mcp` processes pointed at one folder, a +`.oxcode/watch.lock` file lock elects exactly one writer (the process that watches +and re-indexes) while the rest serve reads. If the writer exits, a standby takes +over automatically — so there is never more than one writer, and the index stays +fresh as processes come and go. + ## Prerequisites The plugin ships configuration only — it **cannot bundle the `oxcode` binary**. You must have it installed and on your `PATH` once; from then on it keeps itself -current (see [Updates](#updates)). A project must be indexed before the query -tools can answer anything — either run `oxcode index` once, or call the -`oxcode_index` MCP tool from the agent. +current (see [Updates](#updates)). There is nothing to build by hand: calling +`oxcode_watch` creates `.oxcode/index.oxgdb/` and keeps it current (`O(change)` +per update). 1. **Install the CLI** (binary is `oxcode`; the crate is `oxcode-cli` because the bare `oxcode` name is taken on crates.io): @@ -42,15 +49,8 @@ tools can answer anything — either run `oxcode index` once, or call the cargo install oxcode-cli # build from source ``` -2. **Index your project** (creates `.oxcode/index.oxgdb/`). The MCP server opens - this database lazily and errors if it was never built; re-indexing after - changes is `O(change)`. Do it from the CLI, or ask the agent to run the - `oxcode_index` tool: - - ```sh - cd your-project - oxcode index - ``` +That's it — point the agent at your project, have it call `oxcode_watch`, then ask +a question. A manual `oxcode index` still works if you want to pre-build the index. ## Install diff --git a/crates/oxcode-cli/Cargo.toml b/crates/oxcode-cli/Cargo.toml index 1a089cc..3c9e909 100644 --- a/crates/oxcode-cli/Cargo.toml +++ b/crates/oxcode-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "oxcode-cli" -version = "0.3.0" +version = "0.4.0" description = "Index source code into a graph and serve it to coding agents — `oxcode` CLI." categories = ["development-tools", "command-line-utilities"] readme = "README.md" @@ -23,7 +23,8 @@ path = "src/main.rs" anyhow.workspace = true axoupdater.workspace = true clap.workspace = true -oxcode-core = { path = "../oxcode-core", version = "0.3.0" } +notify-debouncer-full = "0.7" +oxcode-core = { path = "../oxcode-core", version = "0.4.0" } rmcp.workspace = true serde.workspace = true serde_json.workspace = true @@ -35,5 +36,5 @@ predicates.workspace = true # The MCP integration tests in `src/mcp.rs` act as an in-process MCP client over # `tokio::io::duplex`, which needs rmcp's client role. This is dev-only and does # not change the published binary's feature set. -rmcp = { workspace = true, features = ["client"] } +rmcp = { workspace = true, features = ["client", "transport-child-process"] } tempfile.workspace = true diff --git a/crates/oxcode-cli/src/mcp.rs b/crates/oxcode-cli/src/mcp.rs index 77c277a..b4dd607 100644 --- a/crates/oxcode-cli/src/mcp.rs +++ b/crates/oxcode-cli/src/mcp.rs @@ -1,27 +1,63 @@ //! The `oxcode mcp` server: tools mapped onto `oxcode_core::ProjectIndex`. //! -//! Exposes oxcode's read-only queries to coding agents over MCP (stdio). Run it -//! with `oxcode mcp`; configure your agent to launch that command. +//! Exposes oxcode's read-only queries plus a single-writer file watcher +//! (`oxcode_watch`) to coding agents over MCP (stdio). Run it with `oxcode mcp`; +//! configure your agent to launch that command. Across many MCP processes pointed +//! at one folder, a `.oxcode/watch.lock` file lock elects exactly one writer (the +//! process that watches and re-indexes); the rest serve reads. -use std::{collections::HashMap, path::PathBuf, sync::Arc}; +use std::{ + collections::{HashMap, HashSet}, + fs::{File, OpenOptions, TryLockError}, + path::{Path, PathBuf}, + sync::{ + Arc, + atomic::{AtomicU64, Ordering}, + }, + time::Duration, +}; -use oxcode_core::{GraphDirection, IndexProgress, NodeKind, ProjectIndex}; +use notify_debouncer_full::{ + DebounceEventResult, Debouncer, RecommendedCache, new_debouncer, + notify::{RecommendedWatcher, RecursiveMode}, +}; +use oxcode_core::{GraphDirection, IndexStats, NodeKind, ProjectIndex}; use rmcp::{ - ErrorData as McpError, Peer, RoleServer, ServerHandler, ServiceExt, + ErrorData as McpError, ServerHandler, ServiceExt, handler::server::{router::tool::ToolRouter, wrapper::Parameters}, - model::{ - CallToolResult, Content, Meta, ProgressNotificationParam, ServerCapabilities, ServerInfo, - TasksCapability, - }, + model::{CallToolResult, Content, ServerCapabilities, ServerInfo, TasksCapability}, schemars, task_handler, task_manager::OperationProcessor, tool, tool_handler, tool_router, transport::stdio, }; use serde::Deserialize; -use tokio::sync::Mutex; +use tokio::sync::{ + Mutex, + mpsc::{UnboundedReceiver, unbounded_channel}, +}; -/// Runs the MCP server over stdio until the client disconnects. +/// Default debounce window for the file watcher: collapse an editor's save burst +/// (write + rename of a temp file, etc.) into one re-index. Overridable with +/// `OXCODE_WATCH_DEBOUNCE_MS`. +const DEFAULT_DEBOUNCE: Duration = Duration::from_millis(400); + +/// Default failover poll interval: how often a standby retries the writer lock so +/// it can take over when the current writer exits. Overridable with +/// `OXCODE_WATCH_POLL_MS`. +const DEFAULT_POLL: Duration = Duration::from_secs(3); + +/// Filename of the advisory single-writer lock, inside the `.oxcode` index dir. +const WATCH_LOCK_FILE: &str = "watch.lock"; + +/// Directory names whose filesystem events never warrant a re-index: the index +/// store itself (`.oxcode`, the load-bearing entry that prevents a write → +/// event → re-index feedback loop) plus the dirs source discovery already +/// skips. Mirrors `oxcode_core`'s scan skip list. +const WATCH_SKIP_DIRS: &[&str] = &[".oxcode", ".git", "target", "node_modules", "vendor"]; + +/// Runs the MCP server over stdio until the client disconnects. The index is not +/// touched until a client calls `oxcode_watch` (writer) or queries (reader). pub(crate) fn serve() -> anyhow::Result<()> { let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() @@ -33,22 +69,24 @@ pub(crate) fn serve() -> anyhow::Result<()> { }) } -/// Server instructions steering agents to the one-call `oxcode_explore` tool. -const INSTRUCTIONS: &str = "This server answers questions about the indexed code repository in the \ -working directory. If `oxcode_status` reports no database, call `oxcode_index` first to build it \ -(it accepts an optional `path`, defaults to the working directory, and re-indexing after changes is \ -incremental). For almost any code-understanding question, call `oxcode_explore` first with the \ -user's question verbatim: it returns the most relevant symbols (ranked by graph centrality), their \ -source, the relationships among them, the n-ary hyperedges they belong to (trait impl groups and \ -container/module membership, ranked by hypergraph PageRank — the architecture-altitude layer), the \ -blast radius, and the call flow — in one call. Use \ +/// Server instructions steering agents to `oxcode_watch` then `oxcode_explore`. +const INSTRUCTIONS: &str = "This server answers questions about the code repository in the working \ +directory. First call `oxcode_watch` (optional `path`, defaults to the working directory): it builds \ +the index if needed and keeps it current as files change. Only one MCP instance watches a given \ +folder at a time — a file lock elects a single writer; other instances serve reads and take over \ +automatically if the writer exits. Then, for almost any code-understanding question, call \ +`oxcode_explore` first with the user's question verbatim: it returns the most relevant symbols \ +(ranked by graph centrality), their source, the relationships among them, the n-ary hyperedges they \ +belong to (trait impl groups and container/module membership, ranked by hypergraph PageRank — the \ +architecture-altitude layer), the blast radius, and the call flow — in one call. Use \ `oxcode_callers`/`oxcode_callees`/`oxcode_symbol` to follow specific edges, and \ `oxcode_search`/`oxcode_files` only when explore did not surface the target. Prefer these query \ -tools over shelling out to grep or reading files; `oxcode_index` is the only tool that writes \ -(it maintains `.oxcode/`). Do not edit source files."; +tools over shelling out to grep or reading files. Every tool except `oxcode_watch` is read-only; do \ +not edit source files."; -/// MCP server over oxcode's queries plus an `oxcode_index` build tool, caching -/// one opened index per root and driving task-augmented calls through an +/// MCP server over oxcode's read-only queries plus the `oxcode_watch` file +/// watcher. Caches one opened index per root it writes, elects a single writer +/// per root via a file lock, and drives task-augmented calls through an /// [`OperationProcessor`]. #[derive(Clone)] pub(crate) struct OxcodeServer { @@ -57,9 +95,32 @@ pub(crate) struct OxcodeServer { reason = "stored per rmcp's #[tool_router] convention; the #[tool_handler]-generated request router reads it through macro-expanded code the dead-code pass does not attribute" )] tool_router: ToolRouter, + /// Opened readers cached per root this process writes (evicted on reindex). indexes: Arc>>>, /// Backs the rmcp `#[task_handler]` lifecycle for task-augmented tool calls. operations: Arc>, + /// Roots this process is the elected writer for (holds the lock + watcher). + writers: Arc>>>, + /// Roots this process is a standby for (lost the lock; a failover task polls). + standbys: Arc>>, + /// File-watcher debounce window. + debounce: Duration, + /// Failover poll interval for standbys. + poll: Duration, +} + +/// State for a root this process has been elected to write. Dropping it (on +/// process exit) releases the advisory lock and stops the watcher. +struct WriterState { + /// Held advisory `flock`; the kernel frees it on drop or process crash, so a + /// standby can take over. The file itself is never removed. + _lock_file: File, + /// Live debouncer; dropping it stops the watch thread. The `std::sync::Mutex` + /// makes `WriterState: Sync` regardless of the platform watcher's `Sync`-ness. + /// `None` when the watcher failed to start (the lock still elects this writer). + _watcher: std::sync::Mutex>>, + /// Number of reindexes this process has performed for the root (observability). + reindexes: Arc, } /// A code question to answer in one curated call. @@ -126,22 +187,95 @@ pub(crate) struct StatusParams { pub path: Option, } -/// A project root to build or refresh the index for. +/// A project root to watch and keep indexed. #[derive(Debug, Deserialize, schemars::JsonSchema)] -pub(crate) struct IndexParams { - /// Project root to index; defaults to the server's working directory. +pub(crate) struct WatchParams { + /// Project root to watch; defaults to the server's working directory. pub path: Option, } #[tool_router] impl OxcodeServer { - /// Builds an empty server; the index is opened lazily on the first tool call. + /// Builds a server with intervals from the environment (or defaults). Nothing + /// is indexed or watched until a client calls `oxcode_watch` or queries. #[must_use] pub(crate) fn new() -> Self { + Self::new_with( + env_duration("OXCODE_WATCH_DEBOUNCE_MS", DEFAULT_DEBOUNCE), + env_duration("OXCODE_WATCH_POLL_MS", DEFAULT_POLL), + ) + } + + /// Builds a server with explicit debounce + failover-poll windows (tests use + /// tiny values). + #[must_use] + fn new_with(debounce: Duration, poll: Duration) -> Self { Self { tool_router: Self::tool_router(), indexes: Arc::new(Mutex::new(HashMap::new())), operations: Arc::new(Mutex::new(OperationProcessor::new())), + writers: Arc::new(std::sync::Mutex::new(HashMap::new())), + standbys: Arc::new(std::sync::Mutex::new(HashSet::new())), + debounce, + poll, + } + } + + #[tool( + description = "Start (or join) watching a project so its index is built and kept current as files change. Exactly one MCP instance per folder becomes the writer (it holds a file lock and re-indexes on changes); other instances become readers that just serve queries and automatically take over if the writer exits. Call this once before querying. Optional `path` defaults to the working directory.", + execution(task_support = "optional") + )] + async fn oxcode_watch( + &self, + Parameters(params): Parameters, + ) -> Result { + let root = resolve_root(params.path); + + // Idempotent: already participating for this root. + if self.is_writer(&root) { + return json_result(&watch_body(&root, "writer", true, None)); + } + if self.is_standby(&root) { + return json_result(&watch_body(&root, "standby", false, None)); + } + + // The lock lives inside `.oxcode/`, which `.gitignore`s itself. + let index_directory = oxcode_core::index_dir(&root); + ensure_index_dir(&index_directory) + .map_err(|error| McpError::internal_error(error.to_string(), None))?; + let lock_file = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(index_directory.join(WATCH_LOCK_FILE)) + .map_err(|error| McpError::internal_error(format!("open watch lock: {error}"), None))?; + + match lock_file.try_lock() { + Ok(()) => { + let stats = self + .promote_to_writer(root.clone(), lock_file) + .await + .map_err(|error| McpError::internal_error(error.to_string(), None))?; + eprintln!("oxcode: elected as writer for {}", root.display()); + json_result(&watch_body(&root, "writer", true, Some(&stats))) + } + Err(TryLockError::WouldBlock) => { + self.standbys + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(root.clone()); + tokio::spawn(self.clone().failover_loop(root.clone(), lock_file)); + eprintln!( + "oxcode: standby — another instance is watching {}", + root.display() + ); + json_result(&watch_body(&root, "standby", false, None)) + } + Err(TryLockError::Error(error)) => Err(McpError::internal_error( + format!("acquire watch lock: {error}"), + None, + )), } } @@ -160,59 +294,6 @@ impl OxcodeServer { json_result(&report) } - #[tool( - description = "Build or refresh the oxcode index for a project (defaults to the working directory), writing .oxcode/index.oxgdb/. Run this first when oxcode_status reports no database, and after code changes to refresh it (re-indexing is incremental). Reports scan/extract/resolve/store progress when invoked with a progress token.", - execution(task_support = "optional") - )] - async fn oxcode_index( - &self, - Parameters(params): Parameters, - meta: Meta, - peer: Peer, - ) -> Result { - let root = resolve_root(params.path); - let index_root = root.clone(); - let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); - let handle = tokio::task::spawn_blocking(move || { - oxcode_core::index_project_with_progress(&index_root, |progress| { - // The receiver is dropped when no progress token was supplied; - // a failed send just means nobody is listening. - let _ = tx.send(progress); - }) - }); - - // Forward each stage milestone as an MCP progress notification when the - // client opted in with a progress token. Draining the channel until the - // sender drops also serves as the await point for the blocking index. - if let Some(token) = meta.get_progress_token() { - while let Some(progress) = rx.recv().await { - let _ = peer - .notify_progress(ProgressNotificationParam { - progress_token: token.clone(), - progress: f64::from(progress.step), - total: Some(f64::from(progress.total)), - message: Some(progress.stage.label().to_owned()), - }) - .await; - } - } else { - drop(rx); - } - - let stats = handle - .await - .map_err(|error| { - McpError::internal_error(format!("oxcode index task failed: {error}"), None) - })? - .map_err(|error| McpError::internal_error(error.to_string(), None))?; - - // The on-disk database just changed; drop any cached reader for this - // root so the next query reopens the freshly reconciled index. - self.indexes.lock().await.remove(&root); - - json_result(&stats) - } - #[tool( description = "Search indexed symbols by keyword, optionally restricted to symbol kinds." )] @@ -270,15 +351,21 @@ impl OxcodeServer { } #[tool( - description = "Show the indexed project's database status (element/relation counts, paths)." + description = "Show the project's database status (element/relation counts, paths) plus this instance's watch role (writer/standby/reader) and how many times it has re-indexed." )] async fn oxcode_status( &self, Parameters(params): Parameters, ) -> Result { let root = resolve_root(params.path); - let status = blocking(move || oxcode_core::project_status(&root)).await?; - json_result(&status) + let (role, watching, reindexes) = self.watch_state(&root); + let status_root = root.clone(); + let database = blocking(move || oxcode_core::project_status(&status_root)).await?; + let body = serde_json::json!({ + "watch": { "role": role, "watching": watching, "reindexes": reindexes }, + "database": database, + }); + json_result(&body) } /// Shared call-graph path for callers/callees. @@ -295,19 +382,283 @@ impl OxcodeServer { json_result(&report) } - /// Returns a cached opened index for `path` (default cwd), opening on first use. + /// Opens the index for `path` (default cwd). If this process is the writer for + /// the root, the opened reader is cached and evicted on each reindex; any other + /// process opens fresh per query so it reflects the writer's latest commit. A + /// missing index is not built here — call `oxcode_watch` first. async fn index_for(&self, path: Option) -> Result, McpError> { let root = resolve_root(path); - if let Some(index) = self.indexes.lock().await.get(&root) { - return Ok(Arc::clone(index)); + if self.is_writer(&root) { + if let Some(index) = self.indexes.lock().await.get(&root) { + return Ok(Arc::clone(index)); + } + let open_root = root.clone(); + let index = Arc::new(blocking(move || ProjectIndex::open(&open_root)).await?); + self.indexes.lock().await.insert(root, Arc::clone(&index)); + return Ok(index); + } + if !oxcode_core::database_dir(&root).exists() { + return Err(McpError::invalid_params( + format!( + "no index yet for {} — call oxcode_watch to build and keep it current", + root.display() + ), + None, + )); } + // Reader: open fresh so the writer's latest committed snapshot is visible. let open_root = root.clone(); - let index = Arc::new(blocking(move || ProjectIndex::open(&open_root)).await?); - self.indexes.lock().await.insert(root, Arc::clone(&index)); - Ok(index) + Ok(Arc::new( + blocking(move || ProjectIndex::open(&open_root)).await?, + )) + } + + /// Builds/refreshes `root`, starts its watcher, and records this process as the + /// writer. Caller must already hold the advisory lock (`lock_file`). + async fn promote_to_writer( + &self, + root: PathBuf, + lock_file: File, + ) -> anyhow::Result { + let write_lock = Arc::new(Mutex::new(())); + let reindexes = Arc::new(AtomicU64::new(0)); + let stats = run_reindex(&self.indexes, &root, &write_lock, &reindexes).await?; + let watcher = self.spawn_watch(&root, write_lock, Arc::clone(&reindexes)); + let state = Arc::new(WriterState { + _lock_file: lock_file, + _watcher: std::sync::Mutex::new(watcher), + reindexes, + }); + self.writers + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(root.clone(), state); + self.standbys + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&root); + Ok(stats) + } + + /// Failover: poll the writer lock; when the current writer exits and frees it, + /// promote this process to writer (build + watch). Runs until promotion. + async fn failover_loop(self, root: PathBuf, lock_file: File) { + loop { + tokio::time::sleep(self.poll).await; + if self.is_writer(&root) { + break; + } + match lock_file.try_lock() { + Ok(()) => { + self.take_over(root, lock_file).await; + break; + } + Err(TryLockError::WouldBlock) => continue, + Err(TryLockError::Error(error)) => { + eprintln!( + "oxcode: failover lock error for {}: {error}", + root.display() + ); + break; + } + } + } + } + + /// Promotes this process to writer for `root` after winning the freed lock, + /// logging the outcome to stderr. + async fn take_over(&self, root: PathBuf, lock_file: File) { + match self.promote_to_writer(root.clone(), lock_file).await { + Ok(_) => eprintln!( + "oxcode: promoted to writer after previous writer released {}", + root.display() + ), + Err(error) => { + eprintln!( + "oxcode: failover index failed for {}: {error}", + root.display() + ) + } + } + } + + /// Starts a recursive debounced watcher on `root` and a task that re-indexes + /// (serialized by `write_lock`) on each debounced change. Returns `None` if the + /// watcher could not be started. + fn spawn_watch( + &self, + root: &Path, + write_lock: Arc>, + reindexes: Arc, + ) -> Option> { + let (tick_tx, tick_rx) = unbounded_channel::<()>(); + let mut debouncer = + match new_debouncer(self.debounce, None, move |result: DebounceEventResult| { + // Tick on any batch that touches at least one indexable path. A + // batch confined to skip dirs (notably `.oxcode/`, which our own + // re-index writes) is dropped — this is what breaks the feedback + // loop. Watcher errors are transient; the next real event re-syncs. + if let Ok(events) = result + && events + .iter() + .flat_map(|event| event.paths.iter()) + .any(|path| !is_ignored_path(path)) + { + let _ = tick_tx.send(()); + } + }) { + Ok(debouncer) => debouncer, + Err(error) => { + eprintln!( + "oxcode: file watcher unavailable for {}: {error}", + root.display() + ); + return None; + } + }; + if let Err(error) = debouncer.watch(root, RecursiveMode::Recursive) { + eprintln!("oxcode: cannot watch {}: {error}", root.display()); + return None; + } + tokio::spawn(watch_loop( + Arc::clone(&self.indexes), + root.to_path_buf(), + write_lock, + reindexes, + tick_rx, + )); + Some(debouncer) + } + + /// Whether this process is the elected writer for `root`. + fn is_writer(&self, root: &Path) -> bool { + self.writers + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains_key(root) + } + + /// Whether this process is a standby (failover participant) for `root`. + fn is_standby(&self, root: &Path) -> bool { + self.standbys + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains(root) + } + + /// This process's role for `root`, plus whether it is watching and its reindex + /// count (0 for non-writers). + fn watch_state(&self, root: &Path) -> (&'static str, bool, u64) { + if let Some(state) = self + .writers + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(root) + { + return ("writer", true, state.reindexes.load(Ordering::Relaxed)); + } + if self.is_standby(root) { + return ("standby", false, 0); + } + ("reader", false, 0) } } +/// Re-indexes `root` on each debounced change tick until the watcher stops. +async fn watch_loop( + indexes: Arc>>>, + root: PathBuf, + write_lock: Arc>, + reindexes: Arc, + mut tick_rx: UnboundedReceiver<()>, +) { + while tick_rx.recv().await.is_some() { + // Collapse a burst of ticks that landed during the last re-index into one run. + while tick_rx.try_recv().is_ok() {} + match run_reindex(&indexes, &root, &write_lock, &reindexes).await { + Ok(_) => eprintln!( + "oxcode: re-indexed {} (#{})", + root.display(), + reindexes.load(Ordering::Relaxed) + ), + Err(error) => eprintln!("oxcode: re-index failed for {}: {error}", root.display()), + } + } +} + +/// Runs `index_project` for `root` under `write_lock` (serializing this process's +/// writers), evicts the cached reader so the next query reopens the fresh index, +/// and bumps the reindex counter. An unchanged tree is a cheap digest no-op. +async fn run_reindex( + indexes: &Arc>>>, + root: &Path, + write_lock: &Mutex<()>, + reindexes: &AtomicU64, +) -> anyhow::Result { + let _guard = write_lock.lock().await; + let root_owned = root.to_path_buf(); + let stats = + tokio::task::spawn_blocking(move || oxcode_core::index_project(&root_owned)).await??; + // Bump the counter before evicting the cache: the eviction is what lets a + // concurrent reader observe the new commit, so ordering the increment first + // guarantees "new symbol visible" implies "reindex counted". + reindexes.fetch_add(1, Ordering::Relaxed); + indexes.lock().await.remove(root); + Ok(stats) +} + +/// Whether a changed path falls in a directory source discovery skips, so its +/// events should not trigger a re-index. Mirrors `oxcode_core`'s scan skip list; +/// `.oxcode/` is the load-bearing entry that prevents a self-triggered loop. +fn is_ignored_path(path: &Path) -> bool { + path.components().any(|component| { + matches!(component, std::path::Component::Normal(name) + if WATCH_SKIP_DIRS.iter().any(|skip| name == std::ffi::OsStr::new(skip))) + }) +} + +/// Creates the `.oxcode` index dir and its self-ignoring `.gitignore` so the lock +/// file is never committed. Idempotent. +fn ensure_index_dir(index_directory: &Path) -> std::io::Result<()> { + std::fs::create_dir_all(index_directory)?; + let gitignore = index_directory.join(".gitignore"); + if !gitignore.exists() { + std::fs::write(&gitignore, "*\n")?; + } + Ok(()) +} + +/// Reads a millisecond duration from `key`, falling back to `default`. +fn env_duration(key: &str, default: Duration) -> Duration { + std::env::var(key) + .ok() + .and_then(|value| value.parse::().ok()) + .map(Duration::from_millis) + .unwrap_or(default) +} + +/// Builds the JSON body for an `oxcode_watch` response. +fn watch_body( + root: &Path, + role: &str, + watching: bool, + stats: Option<&IndexStats>, +) -> serde_json::Value { + let mut body = serde_json::json!({ + "root": root.display().to_string(), + "role": role, + "watching": watching, + }); + if let Some(stats) = stats { + body["index"] = serde_json::to_value(stats).unwrap_or(serde_json::Value::Null); + } else if !watching { + body["message"] = serde_json::json!( + "another oxcode instance is watching this root; standing by to take over if it exits" + ); + } + body +} + #[tool_handler] #[task_handler(processor = self.operations)] impl ServerHandler for OxcodeServer { @@ -322,9 +673,13 @@ impl ServerHandler for OxcodeServer { } } -/// Resolves the project root from an optional path argument. +/// Resolves the project root from an optional path argument, canonicalizing +/// best-effort so the reader cache, the writer registry, and the lock file all key +/// on the same absolute path (FS events report canonical paths). Falls back to the +/// raw path when it does not exist yet. fn resolve_root(path: Option) -> PathBuf { - PathBuf::from(path.unwrap_or_else(|| ".".to_owned())) + let raw = PathBuf::from(path.unwrap_or_else(|| ".".to_owned())); + std::fs::canonicalize(&raw).unwrap_or(raw) } /// Parses caller-supplied kind strings into `NodeKind`, dropping unknown ones. @@ -369,17 +724,13 @@ fn json_result(value: &T) -> Result>>, - } - - impl ClientHandler for TestClient { - fn on_progress( - &self, - params: ProgressNotificationParam, - _context: NotificationContext, - ) -> impl Future + MaybeSendFuture + '_ { - if let Ok(mut sink) = self.progress.lock() { - sink.push(params); - } - ready(()) - } - } + struct TestClient; + + impl ClientHandler for TestClient {} - /// Wires a fresh `OxcodeServer` to a `TestClient` over an in-memory duplex - /// pipe and returns the connected client service. - async fn connect( - progress: Arc>>, - ) -> RunningService { + /// Wires a fresh `OxcodeServer` (with the given intervals) to a `TestClient` + /// over an in-memory duplex pipe and returns the connected client service. + async fn connect(debounce: Duration, poll: Duration) -> RunningService { let (server_transport, client_transport) = tokio::io::duplex(4096); tokio::spawn(async move { - let server = OxcodeServer::new() + let server = OxcodeServer::new_with(debounce, poll) .serve(server_transport) .await .expect("server serve"); let _ = server.waiting().await; }); - TestClient { progress } + TestClient .serve(client_transport) .await .expect("client connect") @@ -459,19 +795,53 @@ mod tests { .expect("text content") } - /// Waits (bounded) for the progress sink to hold at least `n` notifications, - /// since the client dispatches them asynchronously. - async fn wait_for_progress(progress: &Arc>>, n: usize) { + /// Calls `oxcode_watch` for `path` and returns the parsed JSON response. + async fn watch( + client: &RunningService, + path: &str, + ) -> serde_json::Value { + let result = client + .call_tool(tool_call( + "oxcode_watch", + serde_json::json!({ "path": path }), + )) + .await + .expect("watch call"); + serde_json::from_str(result_text(&result)).expect("watch json") + } + + /// Polls `oxcode_search` (bounded) until `name` actually appears as a match. + /// Inspects the parsed `matches` array — not a substring of the JSON, which + /// would falsely match the echoed `query` field. + async fn poll_symbol_indexed( + client: &RunningService, + path: &str, + name: &str, + ) -> bool { for _ in 0..100 { - if progress.lock().expect("lock").len() >= n { - return; + let searched = client + .call_tool(tool_call( + "oxcode_search", + serde_json::json!({ "path": path, "query": name }), + )) + .await + .expect("search call"); + let report: serde_json::Value = + serde_json::from_str(result_text(&searched)).expect("search json"); + // Keyword search is fuzzy, so check for an exact-named match rather + // than "any match" (which would falsely fire on weak candidates). + let matched = report["matches"] + .as_array() + .is_some_and(|matches| matches.iter().any(|entry| entry["symbol"]["name"] == name)); + if matched { + return true; } - tokio::time::sleep(Duration::from_millis(20)).await; + tokio::time::sleep(Duration::from_millis(50)).await; } + false } - /// Polls `tasks/get` until the task reaches a terminal status (or times out), - /// returning the last observed status. + /// Polls `tasks/get` until the task reaches a terminal status (or times out). async fn poll_until_terminal( client: &RunningService, task_id: &str, @@ -498,18 +868,44 @@ mod tests { status } + /// `flock` is per open-file-description on macOS/Linux: a second independent + /// open of the same path cannot take the lock the first holds. This pins the + /// platform behavior the writer election depends on. + #[test] + fn watch_lock_is_exclusive_per_handle() { + let temp = tempfile::TempDir::new().expect("temp dir"); + let path = temp.path().join("watch.lock"); + let first = OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(&path) + .expect("open first"); + first.try_lock().expect("first acquires"); + let second = OpenOptions::new() + .read(true) + .write(true) + .open(&path) + .expect("open second"); + assert!( + matches!(second.try_lock(), Err(TryLockError::WouldBlock)), + "a second handle cannot take the held lock" + ); + } + #[tokio::test] - async fn lists_tools_with_task_support() { - let client = connect(Arc::default()).await; + async fn lists_tools_with_watch_and_explore_task_support() { + let client = connect(DEFAULT_DEBOUNCE, DEFAULT_POLL).await; let tools = client.list_all_tools().await.expect("list tools"); - let index = tools - .iter() - .find(|tool| tool.name == "oxcode_index") - .expect("oxcode_index is registered"); assert!( - index.input_schema.get("required").is_none(), - "oxcode_index path is optional, so the schema has no required fields" + tools.iter().any(|tool| tool.name == "oxcode_watch"), + "oxcode_watch is registered" + ); + assert!( + tools.iter().all(|tool| tool.name != "oxcode_index"), + "the old write tool is gone" ); let task_support = |name: &str| { @@ -519,34 +915,22 @@ mod tests { .and_then(|tool| tool.execution.as_ref()) .and_then(|execution| execution.task_support) }; - assert_eq!(task_support("oxcode_index"), Some(TaskSupport::Optional)); + assert_eq!(task_support("oxcode_watch"), Some(TaskSupport::Optional)); assert_eq!(task_support("oxcode_explore"), Some(TaskSupport::Optional)); - // Read-only query tools are not task-augmentable. assert_eq!(task_support("oxcode_search"), None); assert_eq!(task_support("oxcode_status"), None); } #[tokio::test] - async fn indexes_then_explores_through_mcp() { + async fn watch_elects_writer_and_serves_queries() { let project = rust_project(); let path = project.path().to_string_lossy().into_owned(); - let client = connect(Arc::default()).await; + let client = connect(Duration::from_millis(50), Duration::from_millis(150)).await; + + let watched = watch(&client, &path).await; + assert_eq!(watched["role"], "writer", "first watcher is the writer"); + assert_eq!(watched["watching"], true); - // Synchronous (non-task) build returns IndexStats directly. - let indexed = client - .call_tool(tool_call( - "oxcode_index", - serde_json::json!({ "path": path }), - )) - .await - .expect("index call"); - assert_ne!(indexed.is_error, Some(true), "index succeeded"); - let stats: serde_json::Value = - serde_json::from_str(result_text(&indexed)).expect("IndexStats json"); - assert!(stats["files"].as_u64().unwrap_or(0) >= 1, "indexed a file"); - assert!(stats["symbols"].as_u64().unwrap_or(0) >= 2, "found symbols"); - - // The freshly built index is queryable in the same session (cache evicted). let explored = client .call_tool(tool_call( "oxcode_explore", @@ -554,64 +938,103 @@ mod tests { )) .await .expect("explore call"); - assert_ne!(explored.is_error, Some(true), "explore succeeded"); assert!( result_text(&explored).contains("entry"), - "explore surfaces the `entry` symbol" + "writer's index is queryable" ); } #[tokio::test] - async fn index_emits_stage_progress() { + async fn second_watcher_on_same_root_is_standby() { let project = rust_project(); let path = project.path().to_string_lossy().into_owned(); - let progress = Arc::new(StdMutex::new(Vec::new())); - let client = connect(Arc::clone(&progress)).await; - - // A cancellable request always carries a progress token, so the server - // streams stage notifications back to `TestClient::on_progress`. - let params = tool_call("oxcode_index", serde_json::json!({ "path": path })); - let handle = client - .send_cancellable_request( - ClientRequest::CallToolRequest(Request::new(params)), - PeerRequestOptions::no_options(), - ) + let writer_client = connect(Duration::from_millis(50), Duration::from_millis(150)).await; + let standby_client = connect(Duration::from_millis(50), Duration::from_millis(150)).await; + + assert_eq!(watch(&writer_client, &path).await["role"], "writer"); + // Second server, same root: the lock is held, so it becomes a standby. + assert_eq!(watch(&standby_client, &path).await["role"], "standby"); + + // The standby still answers queries off the shared on-disk index. + let explored = standby_client + .call_tool(tool_call( + "oxcode_explore", + serde_json::json!({ "path": path, "query": "entry" }), + )) .await - .expect("send index"); - handle.await_response().await.expect("index response"); - - wait_for_progress(&progress, 4).await; - let captured = progress.lock().expect("lock").clone(); - let messages: Vec<_> = captured - .iter() - .filter_map(|notification| notification.message.clone()) - .collect(); - assert_eq!( - messages, - [ - "scanning sources", - "extracting symbols", - "resolving references", - "reconciling database", - ] + .expect("reader explore"); + assert!(result_text(&explored).contains("entry")); + } + + #[tokio::test] + async fn query_without_watch_errors_when_no_index() { + let project = rust_project(); + let path = project.path().to_string_lossy().into_owned(); + let client = connect(DEFAULT_DEBOUNCE, DEFAULT_POLL).await; + + // No oxcode_watch, no prior index: a query must not build; it hints instead. + let result = client + .call_tool(tool_call( + "oxcode_explore", + serde_json::json!({ "path": path, "query": "entry" }), + )) + .await; + assert!( + result.is_err(), + "query before oxcode_watch errors with a hint, never silently builds" ); - let steps: Vec = captured.iter().map(|n| n.progress).collect(); - assert_eq!(steps, [1.0, 2.0, 3.0, 4.0], "progress increases 1..=4"); + } + + #[tokio::test] + async fn writer_auto_reindexes_on_change() { + let project = rust_project(); + let path = project.path().to_string_lossy().into_owned(); + let client = connect(Duration::from_millis(50), Duration::from_millis(150)).await; + + assert_eq!(watch(&client, &path).await["role"], "writer"); + + // Let the FS-event stream establish before the change: FSEvents (and + // other backends) have a startup window where a change can land as + // initial state and go unreported. + tokio::time::sleep(Duration::from_millis(300)).await; + std::fs::write( + project.path().join("src/extra.rs"), + "pub fn brand_new_symbol() {}\n", + ) + .expect("write extra.rs"); + + let found = poll_symbol_indexed(&client, &path, "brand_new_symbol").await; assert!( - captured.iter().all(|n| n.total == Some(4.0)), - "every step reports a total of 4" + found, + "the writer's watcher re-indexed and surfaced the symbol" + ); + + let status: serde_json::Value = serde_json::from_str(result_text( + &client + .call_tool(tool_call( + "oxcode_status", + serde_json::json!({ "path": path }), + )) + .await + .expect("status call"), + )) + .expect("status json"); + assert_eq!(status["watch"]["role"], "writer"); + assert!( + status["watch"]["reindexes"].as_u64().unwrap_or(0) >= 2, + "writer reindexed at least the initial build and the change" ); } #[tokio::test] - async fn task_augmented_index_completes() { + async fn task_augmented_watch_completes() { let project = rust_project(); let path = project.path().to_string_lossy().into_owned(); - let client = connect(Arc::default()).await; + let client = connect(Duration::from_millis(50), Duration::from_millis(150)).await; // Task-augment the call: typed `call_tool` cannot carry a task field, so // send the request directly and expect an immediate CreateTaskResult. - let mut params = tool_call("oxcode_index", serde_json::json!({ "path": path })); + let mut params = tool_call("oxcode_watch", serde_json::json!({ "path": path })); params.task = serde_json::json!({ "ttl": 60_000 }).as_object().cloned(); let created = client .send_request(ClientRequest::CallToolRequest(Request::new(params))) @@ -625,15 +1048,13 @@ mod tests { other => panic!("expected CreateTaskResult, got {other:?}"), }; - // Poll tasks/get to a terminal status before fetching the result - // (rmcp's tasks/result destructively consumes the completed entry). let status = poll_until_terminal(&client, &task_id).await; - assert_eq!(status, TaskStatus::Completed, "task ran to completion"); + assert_eq!( + status, + TaskStatus::Completed, + "watch task ran to completion" + ); - // tasks/result returns the deferred CallToolResult carrying the same - // IndexStats a synchronous call would have produced. rmcp decodes the - // payload (a serialized CallToolResult) straight into the matching - // ServerResult::CallToolResult variant; accept the raw payload too. let payload = client .send_request(ClientRequest::GetTaskResultRequest(Request::new( GetTaskResultParams { @@ -651,10 +1072,9 @@ mod tests { .to_owned(), other => panic!("expected the deferred tool result, got {other:?}"), }; - let stats: serde_json::Value = serde_json::from_str(&text).expect("IndexStats json"); assert!( - stats["symbols"].as_u64().unwrap_or(0) >= 2, - "deferred result carries IndexStats" + text.contains("writer"), + "deferred watch result reports the elected writer role" ); } } diff --git a/crates/oxcode-cli/tests/multiprocess.rs b/crates/oxcode-cli/tests/multiprocess.rs new file mode 100644 index 0000000..e795c06 --- /dev/null +++ b/crates/oxcode-cli/tests/multiprocess.rs @@ -0,0 +1,225 @@ +//! Real multi-process end-to-end test for the lock-elected file watcher. +//! +//! This spawns several actual `oxcode mcp` processes against one shared temp +//! repository and drives each as an MCP client over its stdio. It is the only +//! test that proves the cross-process guarantee — a single in-process server +//! cannot: exactly one process is elected writer (it holds `.oxcode/watch.lock` +//! and re-indexes), the others serve reads, and when the writer exits a standby +//! takes over. Every assertion is observed over MCP (`oxcode_watch` role + +//! `oxcode_status`), never by scraping logs. + +use std::{path::Path, time::Duration}; + +use rmcp::{ + RoleClient, ServiceExt, + model::{CallToolRequestParams, CallToolResult}, + service::RunningService, + transport::{ConfigureCommandExt, TokioChildProcess}, +}; + +/// A no-op MCP client; the spawned servers are what we exercise. +#[derive(Clone, Default)] +struct Probe; + +impl rmcp::ClientHandler for Probe {} + +type Client = RunningService; + +/// Spawns a real `oxcode mcp` process with fast watcher intervals and connects a +/// client to it over stdio. Auto-update is disabled so it never re-execs. +async fn spawn_server(bin: &Path) -> Client { + let command = tokio::process::Command::new(bin).configure(|cmd| { + cmd.arg("mcp") + .env("OXCODE_NO_AUTO_UPDATE", "1") + .env("OXCODE_WATCH_DEBOUNCE_MS", "40") + .env("OXCODE_WATCH_POLL_MS", "150"); + }); + Probe + .serve(TokioChildProcess::new(command).expect("spawn oxcode mcp")) + .await + .expect("connect client") +} + +/// Writes a minimal two-function Rust project into a fresh temp dir and returns +/// the dir plus its canonical path string (all tool calls target this `path`, +/// exercising the "mcp launched outside the folder" case). +fn rust_project() -> (tempfile::TempDir, String) { + let temp = tempfile::TempDir::new().expect("temp dir"); + std::fs::create_dir_all(temp.path().join("src")).expect("mkdir src"); + std::fs::write( + temp.path().join("src/lib.rs"), + "pub fn helper() {}\npub fn entry() {\n helper();\n}\n", + ) + .expect("write lib.rs"); + let path = std::fs::canonicalize(temp.path()) + .expect("canonicalize") + .to_string_lossy() + .into_owned(); + (temp, path) +} + +/// Extracts the single text content block from a tool result. +fn result_text(result: &CallToolResult) -> &str { + result + .content + .first() + .and_then(|content| content.as_text()) + .map(|text| text.text.as_str()) + .expect("text content") +} + +/// Calls a tool and parses its JSON body. +async fn call(client: &Client, name: &'static str, args: serde_json::Value) -> serde_json::Value { + let mut params = CallToolRequestParams::new(name); + params.arguments = args.as_object().cloned(); + let result = client.call_tool(params).await.expect("tool call"); + serde_json::from_str(result_text(&result)).expect("tool json") +} + +/// Calls `oxcode_watch` and returns the elected role. +async fn watch_role(client: &Client, path: &str) -> String { + call(client, "oxcode_watch", serde_json::json!({ "path": path })).await["role"] + .as_str() + .expect("role") + .to_owned() +} + +/// Returns this process's watch state for `path` as `(role, reindexes)`. +async fn watch_state(client: &Client, path: &str) -> (String, u64) { + let status = call(client, "oxcode_status", serde_json::json!({ "path": path })).await; + let role = status["watch"]["role"].as_str().expect("role").to_owned(); + let reindexes = status["watch"]["reindexes"].as_u64().unwrap_or(0); + (role, reindexes) +} + +/// Whether an `oxcode_search` for `name` returns an exact-named match (keyword +/// search is fuzzy, so "any match" would false-positive). +async fn search_finds(client: &Client, path: &str, name: &str) -> bool { + let report = call( + client, + "oxcode_search", + serde_json::json!({ "path": path, "query": name }), + ) + .await; + report["matches"] + .as_array() + .is_some_and(|matches| matches.iter().any(|entry| entry["symbol"]["name"] == name)) +} + +/// Bounded poll: true once `client` can search up `name` (≤ ~12s). +async fn poll_search_finds(client: &Client, path: &str, name: &str) -> bool { + for _ in 0..120 { + if search_finds(client, path, name).await { + return true; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + false +} + +/// Index of the client currently reporting the `writer` role, if any. +async fn writer_index(clients: &[Client], path: &str) -> Option { + for (index, client) in clients.iter().enumerate() { + if watch_state(client, path).await.0 == "writer" { + return Some(index); + } + } + None +} + +#[tokio::test] +async fn single_writer_election_reads_and_failover() { + let bin = assert_cmd::cargo::cargo_bin("oxcode"); + let (_project, path) = rust_project(); + + // Three independent `oxcode mcp` processes against the same repo. + let mut clients = vec![ + spawn_server(&bin).await, + spawn_server(&bin).await, + spawn_server(&bin).await, + ]; + + // 1. Single-writer election: each calls oxcode_watch; exactly one is writer. + let mut writer = None; + for (index, client) in clients.iter().enumerate() { + let role = watch_role(client, &path).await; + if role == "writer" { + assert!(writer.is_none(), "a second writer was elected"); + writer = Some(index); + } else { + assert_eq!( + role, "standby", + "non-writers that called watch are standbys" + ); + } + } + let writer = writer.expect("exactly one process was elected writer"); + + // 2. Every process can read (the writer built the shared index). + for client in &clients { + assert!( + poll_search_finds(client, &path, "entry").await, + "every instance reads the shared index" + ); + } + + // 3. Only the writer re-indexes; all instances see the change. Settle first so the writer's + // FS-event stream is established before the edit. + tokio::time::sleep(Duration::from_millis(500)).await; + std::fs::write( + std::path::Path::new(&path).join("src/extra.rs"), + "pub fn brand_new_symbol() {}\n", + ) + .expect("write extra.rs"); + + for client in &clients { + assert!( + poll_search_finds(client, &path, "brand_new_symbol").await, + "every instance reflects the writer's re-index" + ); + } + for (index, client) in clients.iter().enumerate() { + let (role, reindexes) = watch_state(client, &path).await; + if index == writer { + assert_eq!(role, "writer"); + assert!( + reindexes >= 2, + "writer ran the initial build plus the change" + ); + } else { + assert_eq!(role, "standby"); + assert_eq!(reindexes, 0, "standbys never re-index"); + } + } + + // 4. Failover: tear the writer down; a standby must take over and re-index. + let writer_client = clients.remove(writer); + writer_client.cancel().await.ok(); // child exits → OS frees watch.lock + + // A surviving standby becomes the new writer within a few poll intervals. + let mut promoted = None; + for _ in 0..120 { + promoted = writer_index(&clients, &path).await; + if promoted.is_some() { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + let promoted = promoted.expect("a standby was promoted to writer after the writer exited"); + + // The new writer keeps the index current: a fresh edit propagates. + tokio::time::sleep(Duration::from_millis(500)).await; + std::fs::write( + std::path::Path::new(&path).join("src/more.rs"), + "pub fn later_symbol() {}\n", + ) + .expect("write more.rs"); + assert!( + poll_search_finds(&clients[promoted], &path, "later_symbol").await, + "the promoted writer re-indexes new changes" + ); + + for client in clients { + client.cancel().await.ok(); + } +} diff --git a/crates/oxcode-core/Cargo.toml b/crates/oxcode-core/Cargo.toml index a402ece..7396226 100644 --- a/crates/oxcode-core/Cargo.toml +++ b/crates/oxcode-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "oxcode-core" -version = "0.3.0" +version = "0.4.0" description = "OxGraph-native code indexing and navigation engine." categories = ["development-tools"] readme = "README.md" @@ -15,7 +15,7 @@ rust-version.workspace = true [dependencies] ignore.workspace = true -oxcode-model = { path = "../oxcode-model", version = "0.3.0" } +oxcode-model = { path = "../oxcode-model", version = "0.4.0" } oxgraph.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/oxcode-model/Cargo.toml b/crates/oxcode-model/Cargo.toml index 7824c40..158586a 100644 --- a/crates/oxcode-model/Cargo.toml +++ b/crates/oxcode-model/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "oxcode-model" -version = "0.3.0" +version = "0.4.0" description = "Storage-neutral code graph model types for oxcode." categories = ["development-tools"] readme = "README.md" diff --git a/npm/README.md b/npm/README.md index c33264f..3301202 100644 --- a/npm/README.md +++ b/npm/README.md @@ -35,6 +35,8 @@ This is equivalent to running `oxcode mcp` directly — if you already have the binary, pointing your client at `command: "oxcode", args: ["mcp"]` is simpler and avoids the Node hop. -Index a project first (`oxcode index`, or call the `oxcode_index` tool) so the -read tools have data. See the [oxcode repo](https://github.com/oxgraph/oxcode) -for the full CLI and tool list. +Call the `oxcode_watch` tool once: it builds the index and keeps it current as +files change. Across many `oxcode mcp` processes on one folder, a file lock elects +a single writer (the one watcher/re-indexer) while the rest serve reads, and a +standby takes over if the writer exits. See the +[oxcode repo](https://github.com/oxgraph/oxcode) for the full CLI and tool list. diff --git a/npm/package.json b/npm/package.json index 61f4a53..43415e8 100644 --- a/npm/package.json +++ b/npm/package.json @@ -1,6 +1,6 @@ { "name": "@snowmead/oxcode-mcp", - "version": "0.2.0", + "version": "0.4.0", "description": "npm launcher for the oxcode MCP server — PageRank-curated code intelligence for coding agents. Requires the self-updating `oxcode` binary on PATH.", "mcpName": "io.github.snowmead/oxcode", "type": "commonjs", diff --git a/server.json b/server.json index cc9ab32..8eeab55 100644 --- a/server.json +++ b/server.json @@ -6,12 +6,12 @@ "url": "https://github.com/oxgraph/oxcode", "source": "github" }, - "version": "0.2.0", + "version": "0.4.0", "packages": [ { "registryType": "npm", "identifier": "@snowmead/oxcode-mcp", - "version": "0.2.0", + "version": "0.4.0", "transport": { "type": "stdio" },