Skip to main content

nusy_codegraph/
nats_service.rs

1//! CodeGraph NATS service — exposes code graph tools via NatsServiceBuilder.
2//!
3//! Provides 8 handlers over the `codegraph.cmd` subject prefix, making the
4//! code graph queryable and modifiable via NATS request/reply:
5//!
6//! | Command      | Description                                    |
7//! |-------------|------------------------------------------------|
8//! | `search`    | Search nodes by name/body pattern              |
9//! | `read`      | Read a specific node by ID                     |
10//! | `deps`      | Callers/callees for a node                     |
11//! | `replace`   | Update a node's body (requires rationale)      |
12//! | `query`     | Structured filter query                        |
13//! | `build`     | Run cargo build                                |
14//! | `test`      | Run cargo test                                 |
15//! | `tools`     | Return tool schemas for MCP bridge discovery   |
16//!
17//! Subject prefix: `codegraph.cmd`
18//! Binary: `nusy-codegraph-service`
19
20use crate::mcp_tools::{NodeUpdate, QueryFilter, codegraph_query_objects, codegraph_update_object};
21use crate::schema::{CodeNode, CodeNodeKind};
22use crate::search::{CodeSearch, callees, callers, search_nodes};
23use arrow::array::{Array, RecordBatch};
24use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
25use serde::{Deserialize, Serialize};
26use std::fs::File;
27use std::path::{Path, PathBuf};
28use std::process::Command;
29
30// ─── Request/Response types ─────────────────────────────────────────────────
31
32#[derive(Deserialize)]
33pub struct SearchRequest {
34    pub query: String,
35    #[serde(default)]
36    pub kind: Option<String>,
37    #[serde(default)]
38    pub limit: Option<usize>,
39}
40
41#[derive(Serialize)]
42pub struct NodeSummary {
43    pub id: String,
44    pub name: String,
45    pub kind: String,
46    pub signature: Option<String>,
47    pub file_path: Option<String>,
48    pub loc: Option<i32>,
49}
50
51impl NodeSummary {
52    fn from_node(n: &CodeNode) -> Self {
53        NodeSummary {
54            id: n.id.clone(),
55            name: n.name.clone(),
56            kind: n.kind.as_str().to_string(),
57            signature: n.signature.clone(),
58            file_path: n.file_path.clone(),
59            loc: n.loc,
60        }
61    }
62}
63
64#[derive(Serialize)]
65pub struct SearchResponse {
66    pub nodes: Vec<NodeSummary>,
67    pub total_matched: usize,
68    pub total_scanned: usize,
69}
70
71#[derive(Deserialize)]
72pub struct ReadRequest {
73    pub id: String,
74}
75
76#[derive(Serialize)]
77pub struct ReadResponse {
78    pub node: Option<CodeNodeFull>,
79    pub found: bool,
80}
81
82#[derive(Serialize)]
83pub struct CodeNodeFull {
84    pub id: String,
85    pub name: String,
86    pub kind: String,
87    pub parent_id: Option<String>,
88    pub signature: Option<String>,
89    pub docstring: Option<String>,
90    pub body: Option<String>,
91    pub body_hash: Option<String>,
92    pub loc: Option<i32>,
93    pub complexity: Option<i32>,
94    pub file_path: Option<String>,
95    pub start_line: Option<u32>,
96    pub end_line: Option<u32>,
97}
98
99impl CodeNodeFull {
100    fn from_node(n: CodeNode) -> Self {
101        CodeNodeFull {
102            id: n.id,
103            name: n.name,
104            kind: n.kind.as_str().to_string(),
105            parent_id: n.parent_id,
106            signature: n.signature,
107            docstring: n.docstring,
108            body: n.body,
109            body_hash: n.body_hash,
110            loc: n.loc,
111            complexity: n.cyclomatic_complexity,
112            file_path: n.file_path,
113            start_line: n.start_line,
114            end_line: n.end_line,
115        }
116    }
117}
118
119#[derive(Deserialize)]
120pub struct DepsRequest {
121    pub id: String,
122    /// "callers", "callees", or "both" (default)
123    #[serde(default = "deps_default_direction")]
124    pub direction: String,
125    #[serde(default)]
126    pub limit: Option<usize>,
127}
128
129fn deps_default_direction() -> String {
130    "both".to_string()
131}
132
133#[derive(Serialize)]
134pub struct DepsResponse {
135    pub id: String,
136    pub callers: Vec<NodeSummary>,
137    pub callees: Vec<NodeSummary>,
138}
139
140#[derive(Deserialize)]
141pub struct ReplaceRequest {
142    pub id: String,
143    pub new_body: String,
144    pub rationale: String,
145}
146
147#[derive(Serialize)]
148pub struct ReplaceResponse {
149    pub success: bool,
150    pub id: String,
151    pub new_body_hash: Option<String>,
152    pub rationale: String,
153    pub error: Option<String>,
154}
155
156#[derive(Deserialize, Default)]
157pub struct QueryRequest {
158    #[serde(default)]
159    pub kind: Option<String>,
160    #[serde(default)]
161    pub name_contains: Option<String>,
162    #[serde(default)]
163    pub min_loc: Option<i32>,
164    #[serde(default)]
165    pub limit: Option<usize>,
166}
167
168#[derive(Serialize)]
169pub struct QueryResponse {
170    pub nodes: Vec<NodeSummary>,
171    pub total_matched: usize,
172    pub total_scanned: usize,
173}
174
175#[derive(Deserialize, Default)]
176pub struct BuildRequest {
177    #[serde(default)]
178    pub crate_name: Option<String>,
179}
180
181#[derive(Serialize)]
182pub struct BuildResponse {
183    pub success: bool,
184    pub stdout: String,
185    pub stderr: String,
186    pub command: String,
187}
188
189#[derive(Deserialize, Default)]
190pub struct TestRequest {
191    #[serde(default)]
192    pub crate_name: Option<String>,
193    #[serde(default)]
194    pub filter: Option<String>,
195}
196
197#[derive(Serialize)]
198pub struct TestResponse {
199    pub success: bool,
200    pub stdout: String,
201    pub stderr: String,
202    pub command: String,
203}
204
205/// MCP tool schema for bridge discovery.
206#[derive(Serialize)]
207pub struct ToolSchema {
208    pub name: String,
209    pub description: String,
210    pub input_schema: serde_json::Value,
211}
212
213#[derive(Serialize)]
214pub struct ToolsResponse {
215    pub service: String,
216    pub tools: Vec<ToolSchema>,
217}
218
219// ─── State ──────────────────────────────────────────────────────────────────
220
221/// Mutable state for the CodeGraph NATS service.
222pub struct CodeGraphState {
223    /// All code nodes as a single Arrow RecordBatch.
224    pub nodes: RecordBatch,
225    /// All code edges as a single Arrow RecordBatch.
226    pub edges: RecordBatch,
227    /// Workspace root for running cargo commands.
228    pub workspace_root: PathBuf,
229    /// Directory where nodes.parquet / edges.parquet live.
230    pub graph_dir: PathBuf,
231    /// Absolute path to cargo binary (resolved at startup, not runtime).
232    pub cargo_path: PathBuf,
233    /// EX-3184: Optional NATS sync publisher — publishes updates for cross-agent sync.
234    /// None when running without `--nats-url` (offline mode).
235    pub sync_publisher: Option<crate::nats_sync::CodeGraphPublisher>,
236}
237
238/// Resolve the cargo binary path, checking common locations.
239///
240/// launchd services run with a minimal PATH that excludes `~/.cargo/bin`,
241/// so we resolve the absolute path at startup rather than relying on PATH.
242pub fn resolve_cargo_path(explicit: Option<&Path>) -> PathBuf {
243    if let Some(p) = explicit {
244        return p.to_path_buf();
245    }
246    // Try $HOME/.cargo/bin/cargo
247    if let Ok(home) = std::env::var("HOME") {
248        let candidate = PathBuf::from(format!("{home}/.cargo/bin/cargo"));
249        if candidate.exists() {
250            return candidate;
251        }
252    }
253    // Try well-known macOS paths
254    for path in &[
255        "/Users/hankh19/.cargo/bin/cargo",
256        "/opt/homebrew/bin/cargo",
257        "/usr/local/bin/cargo",
258    ] {
259        let candidate = PathBuf::from(path);
260        if candidate.exists() {
261            return candidate;
262        }
263    }
264    // Fall back to bare name (relies on PATH)
265    PathBuf::from("cargo")
266}
267
268impl CodeGraphState {
269    /// Load from Parquet files in `graph_dir`, falling back to empty batches.
270    pub fn load(graph_dir: &Path, workspace_root: &Path, cargo_path: Option<&Path>) -> Self {
271        let nodes_path = graph_dir.join("nodes.parquet");
272        let edges_path = graph_dir.join("edges.parquet");
273        let nodes = load_parquet_or_empty(&nodes_path, &crate::schema::code_nodes_schema());
274        let edges = load_parquet_or_empty(&edges_path, &crate::schema::code_edges_schema());
275        CodeGraphState {
276            nodes,
277            edges,
278            workspace_root: workspace_root.to_path_buf(),
279            graph_dir: graph_dir.to_path_buf(),
280            cargo_path: resolve_cargo_path(cargo_path),
281            sync_publisher: None,
282        }
283    }
284}
285
286fn load_parquet_or_empty(path: &Path, schema: &arrow::datatypes::Schema) -> RecordBatch {
287    let empty = || RecordBatch::new_empty(std::sync::Arc::new(schema.clone()));
288    if !path.exists() {
289        return empty();
290    }
291    let file = match File::open(path) {
292        Ok(f) => f,
293        Err(_) => return empty(),
294    };
295    let reader = match ParquetRecordBatchReaderBuilder::try_new(file).and_then(|b| b.build()) {
296        Ok(r) => r,
297        Err(_) => return empty(),
298    };
299    // Collect ALL row groups — Parquet splits into batches of ~1024 rows.
300    // Previously only read the first batch, silently dropping the rest.
301    let batches: Vec<RecordBatch> = reader.filter_map(|r| r.ok()).collect();
302    if batches.is_empty() {
303        return empty();
304    }
305    arrow::compute::concat_batches(&batches[0].schema(), &batches).unwrap_or_else(|_| empty())
306}
307
308// ─── Handlers ───────────────────────────────────────────────────────────────
309
310pub fn handle_search(payload: &[u8], state: &mut CodeGraphState) -> Vec<u8> {
311    let req: SearchRequest = match serde_json::from_slice(payload) {
312        Ok(r) => r,
313        Err(e) => return noesis_ship::service::error_response(&format!("invalid JSON: {e}"), 400),
314    };
315    let kind = req.kind.as_deref().and_then(CodeNodeKind::parse);
316    let search = CodeSearch {
317        name_pattern: Some(req.query.clone()),
318        body_pattern: Some(req.query.clone()),
319        kind,
320        limit: req.limit.or(Some(20)),
321        ..Default::default()
322    };
323    let result = search_nodes(&state.nodes, &search);
324    let nodes: Vec<NodeSummary> = result.nodes.iter().map(NodeSummary::from_node).collect();
325    let total_matched = nodes.len();
326    noesis_ship::service::serialize_response(&SearchResponse {
327        total_scanned: result.total_scanned,
328        total_matched,
329        nodes,
330    })
331}
332
333pub fn handle_read(payload: &[u8], state: &mut CodeGraphState) -> Vec<u8> {
334    let req: ReadRequest = match serde_json::from_slice(payload) {
335        Ok(r) => r,
336        Err(e) => return noesis_ship::service::error_response(&format!("invalid JSON: {e}"), 400),
337    };
338    use crate::schema::node_col;
339    use arrow::array::StringArray;
340    let idx = state
341        .nodes
342        .column(node_col::ID)
343        .as_any()
344        .downcast_ref::<StringArray>()
345        .and_then(|ids| (0..ids.len()).find(|&i| ids.value(i) == req.id.as_str()));
346    let node = idx.and_then(|i| extract_node_at_index(&state.nodes, i));
347    let found = node.is_some();
348    noesis_ship::service::serialize_response(&ReadResponse {
349        found,
350        node: node.map(CodeNodeFull::from_node),
351    })
352}
353
354pub fn handle_deps(payload: &[u8], state: &mut CodeGraphState) -> Vec<u8> {
355    let req: DepsRequest = match serde_json::from_slice(payload) {
356        Ok(r) => r,
357        Err(e) => return noesis_ship::service::error_response(&format!("invalid JSON: {e}"), 400),
358    };
359    let limit = req.limit.unwrap_or(50);
360    let caller_nodes = if req.direction == "callers" || req.direction == "both" {
361        callers(&req.id, &state.nodes, &state.edges)
362            .into_iter()
363            .take(limit)
364            .map(|n| NodeSummary::from_node(&n))
365            .collect()
366    } else {
367        Vec::new()
368    };
369    let callee_nodes = if req.direction == "callees" || req.direction == "both" {
370        callees(&req.id, &state.nodes, &state.edges)
371            .into_iter()
372            .take(limit)
373            .map(|n| NodeSummary::from_node(&n))
374            .collect()
375    } else {
376        Vec::new()
377    };
378    noesis_ship::service::serialize_response(&DepsResponse {
379        id: req.id,
380        callers: caller_nodes,
381        callees: callee_nodes,
382    })
383}
384
385pub fn handle_replace(payload: &[u8], state: &mut CodeGraphState) -> Vec<u8> {
386    let req: ReplaceRequest = match serde_json::from_slice(payload) {
387        Ok(r) => r,
388        Err(e) => return noesis_ship::service::error_response(&format!("invalid JSON: {e}"), 400),
389    };
390    if req.rationale.trim().is_empty() {
391        return noesis_ship::service::error_response("rationale is required for code_replace", 400);
392    }
393    let update = NodeUpdate {
394        body: Some(req.new_body.clone()),
395        signature: None,
396        docstring: None,
397        body_hash: None,
398        loc: None,
399        cyclomatic_complexity: None,
400        coverage_pct: None,
401    };
402    match codegraph_update_object(&state.nodes, &req.id, &update) {
403        Ok(new_batch) => {
404            use sha2::{Digest, Sha256};
405            let mut hasher = Sha256::new();
406            hasher.update(req.new_body.as_bytes());
407            let hash = format!("{:x}", hasher.finalize());
408            let new_body_hash = Some(hash[..16].to_string());
409
410            state.nodes = new_batch;
411            let graph_dir = state.graph_dir.clone();
412            if let Err(e) =
413                crate::ingest_pipeline::write_graph_parquet(&state.nodes, &state.edges, &graph_dir)
414            {
415                tracing::warn!(error = %e, "failed to persist updated graph");
416            }
417
418            // EX-3184: Publish update to NATS for cross-agent sync.
419            if let Some(ref publisher) = state.sync_publisher {
420                let hash_str = new_body_hash.as_deref().unwrap_or("");
421                let update =
422                    publisher.make_update(&req.id, &req.new_body, hash_str, &req.rationale);
423                // Fire-and-forget — don't block the response on NATS publish.
424                let publisher_clone = state.sync_publisher.as_ref().unwrap().client_ref().clone();
425                let update_bytes = serde_json::to_vec(&update).unwrap_or_default();
426                tokio::spawn(async move {
427                    if let Err(e) = publisher_clone
428                        .publish(crate::nats_sync::UPDATES_SUBJECT, update_bytes.into())
429                        .await
430                    {
431                        tracing::warn!(error = %e, "failed to publish graph sync update");
432                    }
433                });
434            }
435
436            noesis_ship::service::serialize_response(&ReplaceResponse {
437                success: true,
438                id: req.id,
439                new_body_hash,
440                rationale: req.rationale,
441                error: None,
442            })
443        }
444        Err(e) => noesis_ship::service::serialize_response(&ReplaceResponse {
445            success: false,
446            id: req.id,
447            new_body_hash: None,
448            rationale: req.rationale,
449            error: Some(e.to_string()),
450        }),
451    }
452}
453
454pub fn handle_query(payload: &[u8], state: &mut CodeGraphState) -> Vec<u8> {
455    let req: QueryRequest = match serde_json::from_slice(payload) {
456        Ok(r) => r,
457        Err(e) => return noesis_ship::service::error_response(&format!("invalid JSON: {e}"), 400),
458    };
459    let filter = QueryFilter {
460        kind: req.kind,
461        name_contains: req.name_contains,
462        parent_id: None,
463        min_loc: req.min_loc,
464        min_complexity: None,
465        max_coverage: None,
466        limit: req.limit.or(Some(50)),
467    };
468    match codegraph_query_objects(&state.nodes, &filter) {
469        Ok(result) => {
470            let nodes: Vec<NodeSummary> = result.nodes.iter().map(NodeSummary::from_node).collect();
471            noesis_ship::service::serialize_response(&QueryResponse {
472                total_matched: result.total_matched,
473                total_scanned: result.total_scanned,
474                nodes,
475            })
476        }
477        Err(e) => noesis_ship::service::error_response(&e.to_string(), 500),
478    }
479}
480
481/// VY-3495: Set RUSTC env var to nusy-rustc for graph-native compilation.
482///
483/// If nusy-rustc is installed and a graph exists, cargo will read source
484/// from Arrow CodeNode bodies instead of .rs files on disk.
485fn inject_graph_rustc_env(cmd: &mut Command, state: &CodeGraphState) {
486    // Look for nusy-rustc in common locations
487    let nusy_rustc = which_nusy_rustc();
488    if let Some(rustc_path) = nusy_rustc {
489        cmd.env("RUSTC", &rustc_path);
490        cmd.env("NUSY_GRAPH_DIR", &state.graph_dir);
491        cmd.env("NUSY_WORKSPACE", &state.workspace_root);
492        // Use nightly toolchain for rustc_private
493        cmd.env("RUSTUP_TOOLCHAIN", "nightly");
494        tracing::info!(
495            "graph-native build: RUSTC={} NUSY_GRAPH_DIR={}",
496            rustc_path.display(),
497            state.graph_dir.display()
498        );
499    } else {
500        tracing::debug!("nusy-rustc not found, using standard cargo build");
501    }
502}
503
504/// Find the nusy-rustc binary.
505fn which_nusy_rustc() -> Option<PathBuf> {
506    // Check $HOME/.cargo/bin first (installed via cargo install)
507    if let Ok(home) = std::env::var("HOME") {
508        let path = PathBuf::from(home).join(".cargo/bin/nusy-rustc");
509        if path.exists() {
510            return Some(path);
511        }
512    }
513    // Check workspace target directory
514    let workspace_target = PathBuf::from("target/debug/nusy-rustc");
515    if workspace_target.exists() {
516        return Some(workspace_target);
517    }
518    let workspace_release = PathBuf::from("target/release/nusy-rustc");
519    if workspace_release.exists() {
520        return Some(workspace_release);
521    }
522    // Check PATH
523    if let Ok(output) = Command::new("which").arg("nusy-rustc").output() {
524        if output.status.success() {
525            let path = String::from_utf8_lossy(&output.stdout).trim().to_string();
526            if !path.is_empty() {
527                return Some(PathBuf::from(path));
528            }
529        }
530    }
531    None
532}
533
534pub fn handle_build(payload: &[u8], state: &mut CodeGraphState) -> Vec<u8> {
535    let req: BuildRequest = serde_json::from_slice(payload).unwrap_or_default();
536    let mut cmd = Command::new(&state.cargo_path);
537    cmd.arg("build").current_dir(&state.workspace_root);
538    // VY-3495: Graph-native compilation via nusy-rustc FileLoader
539    inject_graph_rustc_env(&mut cmd, state);
540    if let Some(ref c) = req.crate_name {
541        cmd.args(["--package", c]);
542    }
543    let command_str = format!(
544        "cargo build{}",
545        req.crate_name
546            .as_deref()
547            .map(|c| format!(" --package {c}"))
548            .unwrap_or_default()
549    );
550    match cmd.output() {
551        Ok(out) => noesis_ship::service::serialize_response(&BuildResponse {
552            success: out.status.success(),
553            stdout: String::from_utf8_lossy(&out.stdout).to_string(),
554            stderr: String::from_utf8_lossy(&out.stderr).to_string(),
555            command: command_str,
556        }),
557        Err(e) => noesis_ship::service::error_response(&format!("cargo error: {e}"), 500),
558    }
559}
560
561pub fn handle_test(payload: &[u8], state: &mut CodeGraphState) -> Vec<u8> {
562    let req: TestRequest = serde_json::from_slice(payload).unwrap_or_default();
563    let mut cmd = Command::new(&state.cargo_path);
564    cmd.arg("test").current_dir(&state.workspace_root);
565    // VY-3495: Graph-native compilation via nusy-rustc FileLoader
566    inject_graph_rustc_env(&mut cmd, state);
567    if let Some(ref c) = req.crate_name {
568        cmd.args(["--package", c]);
569    }
570    if let Some(ref f) = req.filter {
571        cmd.arg(f);
572    }
573    let command_str = format!(
574        "cargo test{}{}",
575        req.crate_name
576            .as_deref()
577            .map(|c| format!(" --package {c}"))
578            .unwrap_or_default(),
579        req.filter
580            .as_deref()
581            .map(|f| format!(" {f}"))
582            .unwrap_or_default()
583    );
584    match cmd.output() {
585        Ok(out) => noesis_ship::service::serialize_response(&TestResponse {
586            success: out.status.success(),
587            stdout: String::from_utf8_lossy(&out.stdout).to_string(),
588            stderr: String::from_utf8_lossy(&out.stderr).to_string(),
589            command: command_str,
590        }),
591        Err(e) => noesis_ship::service::error_response(&format!("cargo error: {e}"), 500),
592    }
593}
594
595/// Return MCP tool schemas — called by the MCP bridge on `tools/list`.
596pub fn handle_tools(_payload: &[u8], _state: &mut CodeGraphState) -> Vec<u8> {
597    let tools = vec![
598        ToolSchema {
599            name: "code_search".to_string(),
600            description: "Search code nodes by name, body pattern, or kind".to_string(),
601            input_schema: serde_json::json!({
602                "type": "object",
603                "properties": {
604                    "query": {"type": "string", "description": "Substring match on node name and body"},
605                    "kind": {"type": "string", "description": "Optional: function|class|module|file|..."},
606                    "limit": {"type": "integer", "description": "Max results (default 20)"}
607                },
608                "required": ["query"]
609            }),
610        },
611        ToolSchema {
612            name: "code_read".to_string(),
613            description: "Read a code node's full body, signature, and metadata by ID".to_string(),
614            input_schema: serde_json::json!({
615                "type": "object",
616                "properties": {
617                    "id": {"type": "string", "description": "Node ID (e.g. crate::module::fn_name)"}
618                },
619                "required": ["id"]
620            }),
621        },
622        ToolSchema {
623            name: "code_dependencies".to_string(),
624            description: "Find callers and/or callees of a code node".to_string(),
625            input_schema: serde_json::json!({
626                "type": "object",
627                "properties": {
628                    "id": {"type": "string"},
629                    "direction": {
630                        "type": "string",
631                        "enum": ["callers", "callees", "both"],
632                        "description": "Default: both"
633                    },
634                    "limit": {"type": "integer", "description": "Max per direction (default 50)"}
635                },
636                "required": ["id"]
637            }),
638        },
639        ToolSchema {
640            name: "code_replace".to_string(),
641            description: "Replace a code node's body. Always requires user approval.".to_string(),
642            input_schema: serde_json::json!({
643                "type": "object",
644                "properties": {
645                    "id": {"type": "string"},
646                    "new_body": {"type": "string"},
647                    "rationale": {"type": "string", "description": "Required: reason for this change"}
648                },
649                "required": ["id", "new_body", "rationale"]
650            }),
651        },
652        ToolSchema {
653            name: "code_query".to_string(),
654            description: "Structured filter query across all code nodes".to_string(),
655            input_schema: serde_json::json!({
656                "type": "object",
657                "properties": {
658                    "kind": {"type": "string"},
659                    "name_contains": {"type": "string"},
660                    "min_loc": {"type": "integer"},
661                    "limit": {"type": "integer"}
662                }
663            }),
664        },
665        ToolSchema {
666            name: "code_build".to_string(),
667            description: "Run cargo build on the workspace or a specific crate".to_string(),
668            input_schema: serde_json::json!({
669                "type": "object",
670                "properties": {
671                    "crate_name": {"type": "string", "description": "Optional: specific crate"}
672                }
673            }),
674        },
675        ToolSchema {
676            name: "code_test".to_string(),
677            description: "Run cargo test on the workspace or a specific crate".to_string(),
678            input_schema: serde_json::json!({
679                "type": "object",
680                "properties": {
681                    "crate_name": {"type": "string"},
682                    "filter": {"type": "string", "description": "Optional test name filter"}
683                }
684            }),
685        },
686    ];
687    noesis_ship::service::serialize_response(&ToolsResponse {
688        service: "codegraph".to_string(),
689        tools,
690    })
691}
692
693// ─── Internal: extract CodeNode at row index ─────────────────────────────────
694
695fn extract_node_at_index(batch: &RecordBatch, idx: usize) -> Option<CodeNode> {
696    use crate::schema::node_col;
697    use arrow::array::{
698        Float64Array, Int32Array, LargeStringArray, StringArray, UInt32Array, UInt64Array,
699    };
700    use arrow::datatypes::Int32Type;
701
702    if idx >= batch.num_rows() {
703        return None;
704    }
705
706    let get_str = |col: usize| -> Option<String> {
707        batch
708            .column(col)
709            .as_any()
710            .downcast_ref::<StringArray>()
711            .and_then(|a| {
712                if a.is_null(idx) {
713                    None
714                } else {
715                    Some(a.value(idx).to_string())
716                }
717            })
718    };
719
720    let get_large_str = |col: usize| -> Option<String> {
721        batch
722            .column(col)
723            .as_any()
724            .downcast_ref::<LargeStringArray>()
725            .and_then(|a| {
726                if a.is_null(idx) {
727                    None
728                } else {
729                    Some(a.value(idx).to_string())
730                }
731            })
732    };
733
734    let get_i32 = |col: usize| -> Option<i32> {
735        batch
736            .column(col)
737            .as_any()
738            .downcast_ref::<Int32Array>()
739            .and_then(|a| {
740                if a.is_null(idx) {
741                    None
742                } else {
743                    Some(a.value(idx))
744                }
745            })
746    };
747
748    let get_f64 = |col: usize| -> Option<f64> {
749        batch
750            .column(col)
751            .as_any()
752            .downcast_ref::<Float64Array>()
753            .and_then(|a| {
754                if a.is_null(idx) {
755                    None
756                } else {
757                    Some(a.value(idx))
758                }
759            })
760    };
761
762    let get_u32 = |col: usize| -> Option<u32> {
763        batch
764            .column(col)
765            .as_any()
766            .downcast_ref::<UInt32Array>()
767            .and_then(|a| {
768                if a.is_null(idx) {
769                    None
770                } else {
771                    Some(a.value(idx))
772                }
773            })
774    };
775
776    let get_u64 = |col: usize| -> Option<u64> {
777        batch
778            .column(col)
779            .as_any()
780            .downcast_ref::<UInt64Array>()
781            .and_then(|a| {
782                if a.is_null(idx) {
783                    None
784                } else {
785                    Some(a.value(idx))
786                }
787            })
788    };
789
790    let id = get_str(node_col::ID)?;
791
792    // KIND is a DictionaryArray<Int32Type, StringArray>
793    let kind_str = {
794        use arrow::array::DictionaryArray;
795        if let Some(dict) = batch
796            .column(node_col::KIND)
797            .as_any()
798            .downcast_ref::<DictionaryArray<Int32Type>>()
799        {
800            let key = dict.keys().value(idx) as usize;
801            dict.values()
802                .as_any()
803                .downcast_ref::<StringArray>()
804                .map(|v| v.value(key).to_string())
805                .unwrap_or_default()
806        } else {
807            get_str(node_col::KIND).unwrap_or_default()
808        }
809    };
810    let kind = CodeNodeKind::parse(&kind_str).unwrap_or(CodeNodeKind::File);
811
812    Some(CodeNode {
813        id,
814        kind,
815        parent_id: get_str(node_col::PARENT_ID),
816        name: get_str(node_col::NAME).unwrap_or_default(),
817        signature: get_str(node_col::SIGNATURE),
818        docstring: get_str(node_col::DOCSTRING),
819        body_hash: get_str(node_col::BODY_HASH),
820        body: get_large_str(node_col::BODY),
821        loc: get_i32(node_col::LOC),
822        cyclomatic_complexity: get_i32(node_col::CYCLOMATIC_COMPLEXITY),
823        coverage_pct: get_f64(node_col::COVERAGE_PCT),
824        last_modified: None,
825        start_line: get_u32(node_col::START_LINE),
826        end_line: get_u32(node_col::END_LINE),
827        start_col: get_u32(node_col::START_COL),
828        end_col: get_u32(node_col::END_COL),
829        file_path: get_str(node_col::FILE_PATH),
830        byte_offset: get_u64(node_col::BYTE_OFFSET),
831    })
832}
833
834// ─── Tests ──────────────────────────────────────────────────────────────────
835
836#[cfg(test)]
837mod tests {
838    use super::*;
839    use crate::schema::{
840        CodeEdge, CodeNode, CodeNodeKind, build_code_edges_batch, build_code_nodes_batch,
841    };
842
843    fn make_state() -> CodeGraphState {
844        let nodes = vec![
845            CodeNode {
846                id: "crate::search::search_nodes".to_string(),
847                kind: CodeNodeKind::Function,
848                parent_id: Some("crate::search".to_string()),
849                name: "search_nodes".to_string(),
850                signature: Some(
851                    "pub fn search_nodes(batch: &RecordBatch, query: &CodeSearch) -> SearchResult"
852                        .to_string(),
853                ),
854                body: Some("// finds matching nodes".to_string()),
855                loc: Some(42),
856                ..Default::default()
857            },
858            CodeNode {
859                id: "crate::ingest::ingest_files".to_string(),
860                kind: CodeNodeKind::Function,
861                parent_id: Some("crate::ingest".to_string()),
862                name: "ingest_files".to_string(),
863                signature: Some(
864                    "pub fn ingest_files(root: &Path, files: &[PathBuf]) -> Result<IngestResult>"
865                        .to_string(),
866                ),
867                body: Some("// ingest source files".to_string()),
868                loc: Some(100),
869                ..Default::default()
870            },
871        ];
872        let edges: Vec<CodeEdge> = vec![];
873        let nodes_batch = build_code_nodes_batch(&nodes).expect("nodes batch");
874        let edges_batch = build_code_edges_batch(&edges).expect("edges batch");
875        CodeGraphState {
876            nodes: nodes_batch,
877            edges: edges_batch,
878            workspace_root: PathBuf::from("."),
879            graph_dir: PathBuf::from("."),
880            cargo_path: PathBuf::from("cargo"),
881            sync_publisher: None,
882        }
883    }
884
885    #[test]
886    fn search_by_name_returns_match() {
887        let mut state = make_state();
888        let payload = serde_json::to_vec(&serde_json::json!({"query": "ingest"})).unwrap();
889        let response = handle_search(&payload, &mut state);
890        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
891        assert!(parsed["total_matched"].as_u64().unwrap() >= 1);
892        let names: Vec<&str> = parsed["nodes"]
893            .as_array()
894            .unwrap()
895            .iter()
896            .map(|n| n["name"].as_str().unwrap())
897            .collect();
898        assert!(names.contains(&"ingest_files"));
899    }
900
901    #[test]
902    fn search_no_match_returns_empty() {
903        let mut state = make_state();
904        let payload =
905            serde_json::to_vec(&serde_json::json!({"query": "zzz_nonexistent_xyz"})).unwrap();
906        let response = handle_search(&payload, &mut state);
907        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
908        assert_eq!(parsed["total_matched"], 0);
909        assert_eq!(parsed["nodes"].as_array().unwrap().len(), 0);
910    }
911
912    #[test]
913    fn search_invalid_json_returns_400() {
914        let mut state = make_state();
915        let response = handle_search(b"not json {{", &mut state);
916        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
917        assert_eq!(parsed["code"], 400);
918    }
919
920    #[test]
921    fn read_existing_node_returns_body() {
922        let mut state = make_state();
923        let payload =
924            serde_json::to_vec(&serde_json::json!({"id": "crate::search::search_nodes"})).unwrap();
925        let response = handle_read(&payload, &mut state);
926        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
927        assert_eq!(parsed["found"], true);
928        assert_eq!(parsed["node"]["name"], "search_nodes");
929        assert_eq!(parsed["node"]["loc"], 42);
930    }
931
932    #[test]
933    fn read_missing_node_returns_not_found() {
934        let mut state = make_state();
935        let payload = serde_json::to_vec(&serde_json::json!({"id": "nonexistent::id"})).unwrap();
936        let response = handle_read(&payload, &mut state);
937        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
938        assert_eq!(parsed["found"], false);
939        assert!(parsed["node"].is_null());
940    }
941
942    #[test]
943    fn query_by_kind_returns_functions() {
944        let mut state = make_state();
945        let payload = serde_json::to_vec(&serde_json::json!({
946            "kind": "function",
947            "limit": 10
948        }))
949        .unwrap();
950        let response = handle_query(&payload, &mut state);
951        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
952        assert!(parsed["total_matched"].as_u64().unwrap() >= 1);
953    }
954
955    #[test]
956    fn replace_empty_rationale_returns_400() {
957        let mut state = make_state();
958        let payload = serde_json::to_vec(&serde_json::json!({
959            "id": "crate::search::search_nodes",
960            "new_body": "fn new_body() {}",
961            "rationale": "  "
962        }))
963        .unwrap();
964        let response = handle_replace(&payload, &mut state);
965        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
966        assert_eq!(parsed["code"], 400);
967    }
968
969    #[test]
970    fn tools_returns_seven_schemas() {
971        let mut state = make_state();
972        let response = handle_tools(b"{}", &mut state);
973        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
974        let tools = parsed["tools"].as_array().unwrap();
975        assert_eq!(tools.len(), 7);
976        let names: Vec<&str> = tools.iter().map(|t| t["name"].as_str().unwrap()).collect();
977        for expected in &[
978            "code_search",
979            "code_read",
980            "code_dependencies",
981            "code_replace",
982            "code_query",
983            "code_build",
984            "code_test",
985        ] {
986            assert!(names.contains(expected), "missing tool: {expected}");
987        }
988    }
989
990    #[test]
991    fn deps_invalid_json_returns_400() {
992        let mut state = make_state();
993        let response = handle_deps(b"bad", &mut state);
994        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
995        assert_eq!(parsed["code"], 400);
996    }
997
998    /// Contract test: every MCP tool name in handle_tools() must map to a
999    /// handler registered in service.rs. The bridge strips the `code_` prefix
1000    /// to get the NATS command suffix (e.g., `code_search` → `search`,
1001    /// `code_dependencies` → `dependencies`). This test verifies that every
1002    /// suffix matches a handler name, preventing the `deps` vs `dependencies`
1003    /// mismatch that caused the original 404 bug.
1004    #[test]
1005    fn tool_names_match_handler_commands() {
1006        // These are the handler command names registered in service.rs via
1007        // NatsServiceBuilder::handler(). Keep this in sync with service.rs.
1008        let registered_handlers: Vec<&str> = vec![
1009            "search",
1010            "read",
1011            "dependencies",
1012            "replace",
1013            "query",
1014            "build",
1015            "test",
1016            "tools",
1017        ];
1018
1019        let mut state = make_state();
1020        let response = handle_tools(b"{}", &mut state);
1021        let parsed: serde_json::Value = serde_json::from_slice(&response).unwrap();
1022        let tools = parsed["tools"].as_array().unwrap();
1023
1024        for tool in tools {
1025            let name = tool["name"].as_str().unwrap();
1026            // Bridge routing: strip "code_" prefix to get NATS command suffix
1027            let suffix = name
1028                .strip_prefix("code_")
1029                .unwrap_or_else(|| panic!("tool name '{name}' missing 'code_' prefix"));
1030            assert!(
1031                registered_handlers.contains(&suffix),
1032                "MCP tool '{name}' maps to NATS command '{suffix}', \
1033                 but no handler is registered for '{suffix}'. \
1034                 Registered handlers: {registered_handlers:?}"
1035            );
1036        }
1037    }
1038
1039    #[test]
1040    fn resolve_cargo_path_returns_existing_binary() {
1041        use super::resolve_cargo_path;
1042        // With no explicit path, should resolve to something
1043        let resolved = resolve_cargo_path(None);
1044        // On any dev machine with Rust installed, this should be a real path
1045        assert!(!resolved.as_os_str().is_empty());
1046    }
1047
1048    #[test]
1049    fn resolve_cargo_path_explicit_overrides() {
1050        use super::resolve_cargo_path;
1051        let explicit = std::path::Path::new("/usr/local/fake/cargo");
1052        let resolved = resolve_cargo_path(Some(explicit));
1053        assert_eq!(resolved, PathBuf::from("/usr/local/fake/cargo"));
1054    }
1055}