diff --git a/docs/management-api.openapi.json b/docs/management-api.openapi.json index 02ba7af..2f1d0d2 100644 --- a/docs/management-api.openapi.json +++ b/docs/management-api.openapi.json @@ -270,6 +270,86 @@ } } }, + "/api/namespaces": { + "post": { + "summary": "Create a namespace with a fixed embedding configuration", + "description": "Mirrors the `codesearch create` CLI command so it can be routed through a running serve process instead of contending for the DuckDB write lock.", + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "type": "object", + "required": [ + "name" + ], + "properties": { + "name": { + "type": "string", + "description": "Namespace to create" + }, + "embedding_target": { + "type": "string", + "enum": [ + "onnx", + "api" + ], + "description": "Embedding backend (default onnx)" + }, + "embedding_model": { + "type": "string", + "description": "Model id; required for api" + }, + "embedding_dimensions": { + "type": "integer", + "description": "Vector dimensionality (default 384)" + }, + "no_embeddings": { + "type": "boolean", + "description": "Keyword + call-graph only" + } + } + } + } + } + }, + "responses": { + "200": { + "description": "Namespace created", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "created": { + "type": "boolean" + }, + "namespace": { + "type": "string" + }, + "embedding_target": { + "type": "string" + }, + "embedding_model": { + "type": "string" + }, + "embedding_dimensions": { + "type": "integer" + } + } + } + } + } + }, + "400": { + "description": "Invalid request" + }, + "500": { + "description": "Internal error" + } + } + } + }, "/api/search": { "post": { "tags": [ @@ -1503,4 +1583,4 @@ } } } -} +} \ No newline at end of file diff --git a/src/cli/mod.rs b/src/cli/mod.rs index 43b1de1..40f0e03 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -1,5 +1,7 @@ use clap::{Subcommand, ValueEnum}; +pub mod server_client; + /// Default port for the MCP HTTP server started by `codesearch serve`. pub const DEFAULT_MCP_PORT: u16 = 8677; diff --git a/src/cli/server_client.rs b/src/cli/server_client.rs new file mode 100644 index 0000000..7c1100d --- /dev/null +++ b/src/cli/server_client.rs @@ -0,0 +1,292 @@ +//! Route write commands through a running `serve` process. +//! +//! DuckDB is single-writer per file. When `codesearch serve` holds the database +//! open, a one-shot CLI write command (`create`, `index`, `delete`) can't take +//! the write lock and would fail. This module detects such a server via the +//! run-info file it writes (see `connector::adapter::management::runinfo`), +//! confirms it's alive with a `/health` probe, and forwards the command to the +//! management API. Read commands don't come here — they open the DB read-only, +//! which DuckDB allows concurrently with the writer. +//! +//! Policy (chosen deliberately): if a server *owns* this data-dir but the API +//! call fails, we return an error telling the user to retry or pass `--local`. +//! We never silently fall back to opening the DB, because the server still holds +//! the lock and that fallback would just reproduce the confusing lock error. + +use std::path::Path; +use std::time::Duration; + +use anyhow::{anyhow, bail, Context, Result}; + +use crate::read_runinfo; + +/// How the CLI resolves where (if anywhere) to route a write command. +pub enum RouteTarget { + /// A live server owns this data-dir; forward to its management base URL. + Server(String), + /// No server detected — open the DB directly, as before. + Local, +} + +/// Short timeout for the liveness probe. A running server on loopback answers +/// `/health` in well under this; if it doesn't, treat it as not usable. +const HEALTH_TIMEOUT: Duration = Duration::from_millis(500); + +/// Decide how a write command should run. +/// +/// - `force_local` (from `--local`) short-circuits to [`RouteTarget::Local`]. +/// - `explicit_server` (from `--server `) forces routing to that URL +/// without reading run-info (still liveness-probed). +/// - Otherwise: read run-info for `data_dir`; if present and its `/health` +/// answers, route to it; if absent, run locally; if present but dead (stale +/// file), run locally (the DB is free, so direct access will succeed). +pub async fn resolve_route( + data_dir: &Path, + force_local: bool, + explicit_server: Option<&str>, +) -> Result { + if force_local { + return Ok(RouteTarget::Local); + } + + if let Some(url) = explicit_server { + let base = url.trim_end_matches('/').to_string(); + if probe_health(&base).await { + return Ok(RouteTarget::Server(base)); + } + bail!("no codesearch server reachable at {base} (from --server)"); + } + + let Some(info) = read_runinfo(data_dir) else { + return Ok(RouteTarget::Local); + }; + let base = info.mgmt_base_url(); + if probe_health(&base).await { + // Warn (don't block) on a version mismatch: the CLI may send a request + // shape an older server doesn't understand. + let cli_version = env!("CARGO_PKG_VERSION"); + if info.version != cli_version { + tracing::warn!( + "codesearch server at {base} is version {}, CLI is {cli_version}", + info.version + ); + } + Ok(RouteTarget::Server(base)) + } else { + // Stale run-info (server gone). The DB lock is free, so run locally. + tracing::info!("run-info present but server not responding; running locally"); + Ok(RouteTarget::Local) + } +} + +/// GET `{base}/health`; true iff it answers 2xx within the timeout. +async fn probe_health(base: &str) -> bool { + let Ok(client) = reqwest::Client::builder().timeout(HEALTH_TIMEOUT).build() else { + return false; + }; + match client.get(format!("{base}/health")).send().await { + Ok(resp) => resp.status().is_success(), + Err(_) => false, + } +} + +/// A client bound to one server's management base URL. +pub struct ServerClient { + base: String, + client: reqwest::Client, +} + +impl ServerClient { + pub fn new(base: String) -> Self { + Self { + base, + client: reqwest::Client::new(), + } + } + + /// `POST /api/namespaces` — create a namespace. Mirrors `codesearch create`. + pub async fn create_namespace( + &self, + name: &str, + embedding_target: &str, + embedding_model: Option<&str>, + embedding_dimensions: usize, + no_embeddings: bool, + ) -> Result { + let body = serde_json::json!({ + "name": name, + "embedding_target": embedding_target, + "embedding_model": embedding_model, + "embedding_dimensions": embedding_dimensions, + "no_embeddings": no_embeddings, + }); + let resp = self + .client + .post(format!("{}/api/namespaces", self.base)) + .json(&body) + .send() + .await + .context("failed to reach the running codesearch server")?; + let value = read_json_or_error(resp, "create namespace").await?; + Ok(format!( + "Created namespace '{}' via the running server.", + value + .get("namespace") + .and_then(|v| v.as_str()) + .unwrap_or(name) + )) + } + + /// `DELETE /api/repositories/{id}` — delete a repository by id or path. + pub async fn delete_repository(&self, id_or_path: &str) -> Result { + let resp = self + .client + .delete(format!( + "{}/api/repositories/{}", + self.base, + urlencode(id_or_path) + )) + .send() + .await + .context("failed to reach the running codesearch server")?; + read_json_or_error(resp, "delete repository").await?; + Ok(format!("Deleted '{id_or_path}' via the running server.")) + } + + /// `POST /api/stream/index` — index a repository, consuming the SSE stream + /// and surfacing the final `done`/`error` event as the command result. + pub async fn index_repository( + &self, + path: &str, + name: Option<&str>, + force: bool, + ) -> Result { + let body = serde_json::json!({ "path": path, "name": name, "force": force }); + let resp = self + .client + .post(format!("{}/api/stream/index", self.base)) + .json(&body) + .send() + .await + .context("failed to reach the running codesearch server")?; + if !resp.status().is_success() { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + bail!("server returned {status} for index: {text}"); + } + consume_index_sse(resp).await + } +} + +/// Read a JSON response, mapping a non-2xx status to a useful error that +/// includes the server's `{"error": …}` message when present. +async fn read_json_or_error(resp: reqwest::Response, op: &str) -> Result { + let status = resp.status(); + let text = resp.text().await.unwrap_or_default(); + if status.is_success() { + return serde_json::from_str(&text) + .with_context(|| format!("{op}: could not parse server response")); + } + let msg = serde_json::from_str::(&text) + .ok() + .and_then(|v| v.get("error").and_then(|e| e.as_str()).map(str::to_string)) + .unwrap_or(text); + Err(anyhow!("{op} failed on the server ({status}): {msg}")) +} + +/// Consume the index SSE stream, returning the final human-readable line built +/// from the terminal `done` event, or an error from an `error` event. +async fn consume_index_sse(resp: reqwest::Response) -> Result { + use futures_util::StreamExt; + + let mut stream = resp.bytes_stream(); + let mut buf = String::new(); + let mut current_event: Option = None; + let mut last_error: Option = None; + let mut done: Option = None; + + while let Some(chunk) = stream.next().await { + let bytes = chunk.context("index stream interrupted")?; + buf.push_str(&String::from_utf8_lossy(&bytes)); + + // SSE frames are separated by a blank line; process complete lines. + while let Some(nl) = buf.find('\n') { + let line = buf[..nl].trim_end_matches('\r').to_string(); + buf.drain(..=nl); + + if let Some(name) = line.strip_prefix("event:") { + current_event = Some(name.trim().to_string()); + } else if let Some(data) = line.strip_prefix("data:") { + let data = data.trim(); + let payload: serde_json::Value = serde_json::from_str(data).unwrap_or_default(); + match current_event.as_deref() { + Some("done") => done = Some(payload), + Some("error") => { + last_error = Some( + payload + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("indexing failed") + .to_string(), + ) + } + _ => {} + } + } + } + } + + if let Some(err) = last_error { + bail!("{err}"); + } + let done = done.ok_or_else(|| anyhow!("index stream ended without a result"))?; + let name = done.get("name").and_then(|v| v.as_str()).unwrap_or(""); + let files = done.get("file_count").and_then(|v| v.as_u64()).unwrap_or(0); + let chunks = done + .get("chunk_count") + .and_then(|v| v.as_u64()) + .unwrap_or(0); + Ok(format!( + "Indexed '{name}' via the running server: {files} files, {chunks} chunks." + )) +} + +/// Minimal path-segment encoding for the repository id/path in a URL. Percent- +/// encodes the characters that would otherwise break the path segment. +fn urlencode(s: &str) -> String { + let mut out = String::with_capacity(s.len()); + for b in s.bytes() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => { + out.push(b as char) + } + _ => out.push_str(&format!("%{b:02X}")), + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn urlencode_escapes_slashes_and_spaces() { + assert_eq!(urlencode("a/b c"), "a%2Fb%20c"); + assert_eq!(urlencode("plain-id_1.2~"), "plain-id_1.2~"); + } + + #[tokio::test] + async fn resolve_route_force_local_skips_detection() { + let dir = tempfile::tempdir().unwrap(); + let route = resolve_route(dir.path(), true, None).await.unwrap(); + assert!(matches!(route, RouteTarget::Local)); + } + + #[tokio::test] + async fn resolve_route_no_runinfo_is_local() { + let dir = tempfile::tempdir().unwrap(); + let route = resolve_route(dir.path(), false, None).await.unwrap(); + assert!(matches!(route, RouteTarget::Local)); + } +} diff --git a/src/connector/adapter/management/handlers/mod.rs b/src/connector/adapter/management/handlers/mod.rs index 6fbe135..17183ff 100644 --- a/src/connector/adapter/management/handlers/mod.rs +++ b/src/connector/adapter/management/handlers/mod.rs @@ -15,6 +15,7 @@ pub mod couplings; pub mod graph; pub mod graph_view; pub mod llm; +pub mod namespaces; pub mod repositories; pub mod search; diff --git a/src/connector/adapter/management/handlers/namespaces.rs b/src/connector/adapter/management/handlers/namespaces.rs new file mode 100644 index 0000000..f8e0f6b --- /dev/null +++ b/src/connector/adapter/management/handlers/namespaces.rs @@ -0,0 +1,126 @@ +//! Namespace endpoints. +//! +//! - `POST /api/namespaces` — create a namespace with a fixed embedding config. +//! +//! This mirrors the `codesearch create` CLI command. It exists so that command +//! can be routed through a running `serve` process: `create` only writes +//! namespace configuration (no embedding model is loaded), but it still needs a +//! DuckDB *write* lock, which the running server holds. Rather than fight over +//! the lock, the CLI POSTs here and the server performs the write. + +use axum::extract::State; +use axum::http::StatusCode; +use axum::Json; +use serde::Deserialize; +use serde_json::{json, Value}; +use std::path::Path; + +use crate::{DuckdbVectorRepository, NamespaceEmbeddingConfig}; +use crate::{DEFAULT_ONNX_EMBEDDING_MODEL, NO_EMBEDDINGS_MODEL}; + +use super::super::error::{ApiError, ApiResult}; +use super::super::server::AppState; + +/// Default embedding dimensionality when the request omits it (all-MiniLM-L6-v2). +const DEFAULT_EMBEDDING_DIMENSIONS: usize = 384; + +/// Body of `POST /api/namespaces`. Field meanings match the `create` CLI flags. +#[derive(Debug, Deserialize)] +pub struct CreateNamespaceRequest { + /// Namespace to create. Required — the server has no per-request "current" + /// namespace default the way the CLI derives one from the working dir. + pub name: String, + /// `"onnx"` (default) or `"api"`. + #[serde(default)] + pub embedding_target: Option, + /// Model identifier; required for `"api"`, defaulted for `"onnx"`. + #[serde(default)] + pub embedding_model: Option, + /// Vector dimensionality. Defaults to 384. + #[serde(default)] + pub embedding_dimensions: Option, + /// Create without embeddings — keyword + call-graph search only. + #[serde(default)] + pub no_embeddings: bool, +} + +/// `POST /api/namespaces` — create a namespace with a fixed embedding config. +/// +/// Resolves the same `(target, model, dimensions)` triple the CLI computes, +/// then writes it via the vector repository. Idempotent at the repository +/// level: creating an existing namespace with a matching config is fine; a +/// conflicting embedding config is rejected there and surfaces as an error. +pub async fn create( + State(state): State, + Json(req): Json, +) -> ApiResult> { + let name = req.name.trim(); + if name.is_empty() { + return Err(ApiError::bad_request("name must not be empty")); + } + + let dimensions = req + .embedding_dimensions + .unwrap_or(DEFAULT_EMBEDDING_DIMENSIONS); + if dimensions == 0 { + return Err(ApiError::bad_request( + "embedding_dimensions must be greater than 0", + )); + } + + let (embedding_target, embedding_model) = if req.no_embeddings { + ( + NO_EMBEDDINGS_MODEL.to_string(), + NO_EMBEDDINGS_MODEL.to_string(), + ) + } else { + match req.embedding_target.as_deref().unwrap_or("onnx") { + "onnx" => ( + "onnx".to_string(), + req.embedding_model + .clone() + .unwrap_or_else(|| DEFAULT_ONNX_EMBEDDING_MODEL.to_string()), + ), + "api" => { + let model = req.embedding_model.clone().ok_or_else(|| { + ApiError::bad_request("embedding_model is required with embedding_target=api") + })?; + ("api".to_string(), model) + } + other => { + return Err(ApiError::bad_request(format!( + "unknown embedding_target '{other}' (expected 'onnx' or 'api')" + ))) + } + } + }; + + let db_path = Path::new(state.container.data_dir()).join("codesearch.duckdb"); + let cfg = NamespaceEmbeddingConfig { + embedding_target: embedding_target.clone(), + embedding_model: embedding_model.clone(), + dimensions, + }; + + // The DuckDB write happens on a blocking connection; keep it off the async + // runtime per the project's async rule. + let name_owned = name.to_string(); + tokio::task::spawn_blocking(move || { + DuckdbVectorRepository::create_namespace(&db_path, &name_owned, &cfg) + }) + .await + .map_err(|e| { + ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!("namespace create task failed: {e}"), + ) + })??; + + Ok(Json(json!({ + "created": true, + "namespace": name, + "embedding_target": embedding_target, + "embedding_model": embedding_model, + "embedding_dimensions": dimensions, + }))) +} diff --git a/src/connector/adapter/management/mod.rs b/src/connector/adapter/management/mod.rs index 909c55c..02adcdd 100644 --- a/src/connector/adapter/management/mod.rs +++ b/src/connector/adapter/management/mod.rs @@ -13,8 +13,12 @@ mod copilot_login; mod error; mod handlers; +mod runinfo; mod server; mod streaming; pub use copilot_login::CopilotLoginService; +pub use runinfo::{ + read as read_runinfo, remove as remove_runinfo, write as write_runinfo, ServeRunInfo, +}; pub use server::{routes, run_management_server, AppState}; diff --git a/src/connector/adapter/management/runinfo.rs b/src/connector/adapter/management/runinfo.rs new file mode 100644 index 0000000..0d8f194 --- /dev/null +++ b/src/connector/adapter/management/runinfo.rs @@ -0,0 +1,151 @@ +//! Server run-info file: how a one-shot CLI invocation discovers that a +//! `serve` process is already running against the same data directory. +//! +//! DuckDB is single-writer per file. When `codesearch serve` holds the database +//! open, any CLI subcommand that needs a *write* lock (`create`, `index`, +//! `delete`) can't open it and fails. The fix is to route those commands +//! through the running server's management API instead — but first the CLI has +//! to find it. On startup `serve` writes this small JSON file into the data +//! directory recording the management port and pid; the CLI reads it, probes +//! `/health`, and if the server is live routes the write through it. The file +//! is removed on graceful shutdown; a stale file (server killed) is detected by +//! the health probe failing, so the CLI never trusts it blindly. + +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; + +/// Filename written into the data directory (next to `codesearch.duckdb`). +const RUNINFO_FILE: &str = "serve.json"; + +/// What a running `serve` advertises about itself. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ServeRunInfo { + /// Management API port (`--mgmt-port`). This is the REST surface the CLI + /// routes write commands through. + pub mgmt_port: u16, + /// MCP HTTP port (`--mcp-port`). Recorded for completeness / diagnostics. + pub mcp_port: u16, + /// PID of the `serve` process, for diagnostics and stale-file reasoning. + pub pid: u32, + /// Crate version of the running server, so a mismatched CLI can warn. + pub version: String, +} + +impl ServeRunInfo { + /// The base URL of the management API on loopback. + pub fn mgmt_base_url(&self) -> String { + format!("http://127.0.0.1:{}", self.mgmt_port) + } +} + +/// Absolute path to the run-info file for a given data directory. +pub fn runinfo_path(data_dir: &Path) -> PathBuf { + data_dir.join(RUNINFO_FILE) +} + +/// Write the run-info file. Called once, when `serve` has bound its ports. +/// +/// Best-effort: a failure to write only means the CLI can't auto-detect this +/// server and will fall back to opening the DB directly, so we log and carry on +/// rather than failing the server. +pub fn write(data_dir: &Path, info: &ServeRunInfo) { + let path = runinfo_path(data_dir); + match serde_json::to_vec_pretty(info) { + Ok(bytes) => { + if let Err(e) = std::fs::write(&path, bytes) { + tracing::warn!("could not write serve run-info at {}: {e}", path.display()); + } else { + tracing::info!("wrote serve run-info at {}", path.display()); + } + } + Err(e) => tracing::warn!("could not serialize serve run-info: {e}"), + } +} + +/// Remove the run-info file. Called on graceful shutdown. Best-effort — a +/// leftover file is harmless because readers verify liveness via `/health`. +pub fn remove(data_dir: &Path) { + let path = runinfo_path(data_dir); + if let Err(e) = std::fs::remove_file(&path) { + // Not existing is fine (e.g. write failed at startup). + if e.kind() != std::io::ErrorKind::NotFound { + tracing::warn!("could not remove serve run-info at {}: {e}", path.display()); + } + } +} + +/// Read the run-info file, if present and parseable. `None` means "no server +/// advertised here" — the caller should proceed with direct DB access. +pub fn read(data_dir: &Path) -> Option { + let path = runinfo_path(data_dir); + let bytes = std::fs::read(&path).ok()?; + match serde_json::from_slice(&bytes) { + Ok(info) => Some(info), + Err(e) => { + tracing::warn!( + "ignoring unparseable serve run-info at {}: {e}", + path.display() + ); + None + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sample() -> ServeRunInfo { + ServeRunInfo { + mgmt_port: 8676, + mcp_port: 8677, + pid: 4242, + version: "9.9.9".to_string(), + } + } + + #[test] + fn write_then_read_round_trips() { + let dir = tempfile::tempdir().unwrap(); + write(dir.path(), &sample()); + let got = read(dir.path()).expect("run-info should be readable"); + assert_eq!(got.mgmt_port, 8676); + assert_eq!(got.mcp_port, 8677); + assert_eq!(got.pid, 4242); + assert_eq!(got.version, "9.9.9"); + } + + #[test] + fn read_absent_is_none() { + let dir = tempfile::tempdir().unwrap(); + assert!(read(dir.path()).is_none()); + } + + #[test] + fn read_garbage_is_none() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(runinfo_path(dir.path()), b"not json").unwrap(); + assert!(read(dir.path()).is_none()); + } + + #[test] + fn remove_deletes_the_file() { + let dir = tempfile::tempdir().unwrap(); + write(dir.path(), &sample()); + assert!(read(dir.path()).is_some()); + remove(dir.path()); + assert!(read(dir.path()).is_none()); + } + + #[test] + fn remove_absent_is_ok() { + let dir = tempfile::tempdir().unwrap(); + remove(dir.path()); // must not panic + } + + #[test] + fn mgmt_base_url_is_loopback() { + assert_eq!(sample().mgmt_base_url(), "http://127.0.0.1:8676"); + } +} diff --git a/src/connector/adapter/management/server.rs b/src/connector/adapter/management/server.rs index 6e07b90..fcffbb0 100644 --- a/src/connector/adapter/management/server.rs +++ b/src/connector/adapter/management/server.rs @@ -82,6 +82,7 @@ pub fn routes(state: AppState) -> Router { get(handlers::repositories::get).delete(handlers::repositories::delete), ) .route("/api/stats", get(handlers::repositories::stats)) + .route("/api/namespaces", post(handlers::namespaces::create)) // Search. .route("/api/search", post(handlers::search::search)) // Call-graph queries. @@ -172,6 +173,7 @@ async fn index(State(_state): State) -> Json { { "method": "GET", "path": "/api/repositories/{id}", "description": "one repository + architecture overview" }, { "method": "DELETE", "path": "/api/repositories/{id}", "description": "delete a repository by ID or path" }, { "method": "GET", "path": "/api/stats", "description": "index-wide statistics" }, + { "method": "POST", "path": "/api/namespaces", "description": "create a namespace with a fixed embedding config" }, { "method": "POST", "path": "/api/search", "description": "hybrid semantic + keyword code search" }, { "method": "POST", "path": "/api/impact", "description": "blast radius of changing a symbol" }, { "method": "GET", "path": "/api/context/{symbol}", "description": "callers + callees of a symbol" }, diff --git a/src/lib.rs b/src/lib.rs index f56c654..4867ef6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -27,7 +27,8 @@ pub use cli::{ }; pub use connector::adapter::management::{ - routes as management_routes, run_management_server, AppState as ManagementAppState, + read_runinfo, remove_runinfo, routes as management_routes, run_management_server, + write_runinfo, AppState as ManagementAppState, ServeRunInfo, }; pub use connector::{ diff --git a/src/main.rs b/src/main.rs index 388ef06..4d123e5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -115,6 +115,18 @@ struct Cli { #[arg(long, global = true, value_enum, default_value = "open-ai")] llm_target: LlmTarget, + /// Force direct database access for write commands even if a `serve` + /// process is running. Fails with a DuckDB lock error if the server still + /// holds the write lock — use only when you know the server is stopped. + #[arg(long, global = true)] + local: bool, + + /// Route write commands (create/index/delete) through the management API of + /// a specific running server (e.g. http://127.0.0.1:8676) instead of + /// auto-detecting one. Implies not opening the database directly. + #[arg(long, global = true)] + server: Option, + #[command(subcommand)] command: Commands, } @@ -215,6 +227,63 @@ async fn main() -> Result<()> { let db_path = std::path::Path::new(&data_dir).join("codesearch.duckdb"); + // Write commands (create / index / delete) need a DuckDB write lock. If a + // `serve` process is running against this data-dir it already holds that + // lock, so route the command through its management API instead of failing. + // Read commands don't come here — they open the DB read-only, which is + // allowed alongside the writer. `--local` forces direct access; `--server` + // forces a specific target. On a server-owned dir where the API call fails, + // we error out (rather than silently hitting the lock). + if matches!( + &cli.command, + Commands::Create { .. } | Commands::Index { .. } | Commands::Delete { .. } + ) { + use codesearch::cli::server_client::{resolve_route, RouteTarget, ServerClient}; + let route = resolve_route( + std::path::Path::new(&data_dir), + cli.local, + cli.server.as_deref(), + ) + .await?; + if let RouteTarget::Server(base) = route { + let client = ServerClient::new(base); + let output = match &cli.command { + Commands::Create { + name, + embedding_target, + embedding_model, + embedding_dimensions, + no_embeddings, + } => { + let namespace = name.as_deref().unwrap_or(&cli.namespace); + let target = match embedding_target { + EmbeddingTarget::Onnx => "onnx", + EmbeddingTarget::Api => "api", + }; + client + .create_namespace( + namespace, + target, + embedding_model.as_deref(), + *embedding_dimensions, + *no_embeddings, + ) + .await? + } + Commands::Index { path, name, force } => { + client + .index_repository(path, name.as_deref(), *force) + .await? + } + Commands::Delete { id_or_path } => client.delete_repository(id_or_path).await?, + _ => unreachable!("guarded by the matches! above"), + }; + println!("{output}"); + return Ok(()); + } + // RouteTarget::Local — fall through to the direct-DB paths below. + } + // `create` only writes namespace configuration — handle it before the // container is built so no embedding model is loaded or downloaded. if let Commands::Create { @@ -329,6 +398,10 @@ async fn main() -> Result<()> { | Commands::Tui { .. } ); + // `data_dir` is moved into ContainerConfig below; keep a copy for the serve + // block's run-info file (written next to the DB in this directory). + let data_dir_for_runinfo = data_dir.clone(); + let config = ContainerConfig { data_dir, mock_embeddings: cli.mock_embeddings, @@ -374,6 +447,18 @@ async fn main() -> Result<()> { let container = Arc::new(Container::new(config).await?); + // Advertise this server so one-shot CLI invocations against the same + // data-dir route their write commands through it instead of hitting the + // DuckDB write lock we hold. Removed after the servers exit; a stale + // file (hard kill) is caught by the CLI's /health probe. + let runinfo = codesearch::ServeRunInfo { + mgmt_port: serve_mgmt_port, + mcp_port: serve_mcp_port, + pid: std::process::id(), + version: env!("CARGO_PKG_VERSION").to_string(), + }; + codesearch::write_runinfo(std::path::Path::new(&data_dir_for_runinfo), &runinfo); + let mcp = run_http_server(container.clone(), serve_mcp_port, serve_public); let mgmt = codesearch::run_management_server(container, serve_mgmt_port, serve_public); @@ -383,10 +468,33 @@ async fn main() -> Result<()> { serve_mgmt_port ); - tokio::select! { - res = mcp => res?, - res = mgmt => res?, - } + // The HTTP servers shut down on ctrl-c (SIGINT), but a supervising app + // (e.g. Hoplon) stops the process with SIGTERM. On Unix, watch for + // SIGTERM too so the run-info file is removed on that path rather than + // leaking. A leaked file isn't fatal — the CLI's /health probe rejects a + // stale one — but cleaning up keeps the common case tidy. + #[cfg(unix)] + let result: Result<()> = { + let mut sigterm = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("failed to install SIGTERM handler"); + tokio::select! { + res = mcp => res, + res = mgmt => res, + _ = sigterm.recv() => { + tracing::info!("received SIGTERM; shutting down serve"); + Ok(()) + } + } + }; + #[cfg(not(unix))] + let result: Result<()> = tokio::select! { + res = mcp => res, + res = mgmt => res, + }; + + codesearch::remove_runinfo(std::path::Path::new(&data_dir_for_runinfo)); + result?; return Ok(()); }