nexql_tools/
exec.rs

1// SPDX-License-Identifier: GPL-3.0-only
2// Copyright (C) 2026 NexQL-OSS Team
3
4//! Tool dispatch for catalog (Phase 2) + index (Phase 3) + Phase 4 surfaces.
5
6use std::sync::Arc;
7
8use nexql_index::{
9    BuildDepth, BuildMode, BuildRequest, CatalogDb, Embedder, IndexQueryService, IndexScope,
10    IndexStore, PgCatalogDb, QueryPolicyFilter, SearchOptions, build_index,
11};
12use nexql_policy::{PolicyFilter, SqlDecision, enforce_read_table_policy, select_table_refs, validate_readonly_sql};
13use serde_json::{Value, json};
14
15use crate::cell_json::{redact_pii_in_payload, rows_to_json_array};
16use crate::error::ToolError;
17use crate::export::{ExportFormat, columns_from_rows, rows_to_csv, rows_to_sql_insert};
18use crate::plan::{analyze_deep_plan, build_explain_sql, extract_plan_metrics};
19use crate::registry::ToolName;
20use crate::schema::{ToolSpec, active_tools};
21use crate::session::ToolSession;
22use crate::sql::{self, REPORT_LIMIT_DEFAULT, SLOW_QUERIES_DEFAULT, parse_ref};
23use crate::write::{
24    apply_ddl, create_index_concurrently, edit_row, execute_sql, import_data, run_maintenance,
25    terminate_query,
26};
27
28/// Default hit cap for `search_schema` (matches TS ToolExecutor).
29const SEARCH_SCHEMA_LIMIT: usize = 10;
30
31const NO_INDEX_HINT: &str =
32    "No schema index configured — call the 'rebuild_index' tool to build an index.";
33
34#[derive(Debug, Clone)]
35pub struct ToolOutcome {
36    pub text: String,
37    pub structured: Option<Value>,
38    pub is_error: bool,
39}
40
41impl ToolOutcome {
42    /// Success payload for MCP `structuredContent`.
43    ///
44    /// Cursor (and some other clients) require `structuredContent` to be a JSON
45    /// **object**. Bare arrays are dropped before the model sees them — always
46    /// wrap: `{ "rows": [ ... ] }`.
47    pub fn ok_json(value: Value) -> Self {
48        let value = ensure_structured_object(value);
49        let text = serde_json::to_string_pretty(&value).unwrap_or_else(|_| value.to_string());
50        Self {
51            text,
52            structured: Some(value),
53            is_error: false,
54        }
55    }
56
57    pub fn err(msg: impl Into<String>) -> Self {
58        let message = msg.into();
59        Self {
60            text: message.clone(),
61            structured: Some(json!({ "error": message })),
62            is_error: true,
63        }
64    }
65}
66
67/// Cursor MCP rejects non-object `structuredContent`. Wrap arrays as `{ "rows": … }`.
68fn ensure_structured_object(value: Value) -> Value {
69    match value {
70        Value::Array(rows) => json!({ "rows": rows }),
71        other => other,
72    }
73}
74
75pub struct ToolRouter {
76    session: Arc<ToolSession>,
77    /// Optional override; when `None`, uses `session.index_store`.
78    index_override: Option<Option<IndexStore>>,
79    /// When true and an embedder is set, `search_schema` fuses via RRF.
80    use_semantic: bool,
81    embedder: Option<Arc<dyn Embedder>>,
82    specs: Vec<ToolSpec>,
83    managed_extension: bool,
84}
85
86impl ToolRouter {
87    pub fn new(session: Arc<ToolSession>) -> Self {
88        Self {
89            session,
90            index_override: None,
91            use_semantic: false,
92            embedder: None,
93            specs: active_tools(),
94            managed_extension: false,
95        }
96    }
97
98    /// Build with an explicit index store (or `None` to force the no-index error path).
99    pub fn with_index_store(session: Arc<ToolSession>, store: Option<IndexStore>) -> Self {
100        Self {
101            session,
102            index_override: Some(store),
103            use_semantic: false,
104            embedder: None,
105            specs: active_tools(),
106            managed_extension: false,
107        }
108    }
109
110    /// Enable semantic RRF fusion for `search_schema` (requires embeddings on disk + embedder).
111    pub fn with_semantic(
112        mut self,
113        use_semantic: bool,
114        embedder: Option<Arc<dyn Embedder>>,
115    ) -> Self {
116        self.use_semantic = use_semantic;
117        self.embedder = embedder;
118        self
119    }
120
121    /// Filter active tools by requested `ToolProfile`.
122    pub fn with_profile(mut self, profile: crate::registry::ToolProfile) -> Self {
123        self.specs = crate::schema::tools_for_profile(profile);
124        self
125    }
126
127    /// Exclude setup/profile mutation tools for managed extension hosts.
128    pub fn with_managed_extension(mut self, enabled: bool) -> Self {
129        self.managed_extension = enabled;
130        if enabled {
131            const BLOCKED: &[ToolName] = &[
132                ToolName::SetupConnection,
133                ToolName::SaveProfile,
134                ToolName::TestProfile,
135                ToolName::ExportProfile,
136                ToolName::ImportProfile,
137            ];
138            self.specs.retain(|s| !BLOCKED.contains(&s.name));
139        }
140        self
141    }
142
143    pub fn specs(&self) -> &[ToolSpec] {
144        &self.specs
145    }
146
147    fn index_store(&self) -> Option<&IndexStore> {
148        match &self.index_override {
149            Some(inner) => inner.as_ref(),
150            None => self.session.index_store.as_ref(),
151        }
152    }
153
154    fn query_filter(&self) -> QueryPolicyFilter {
155        policy_to_query_filter(&self.session.filter())
156    }
157
158    pub async fn call(&self, name: &str, args: Value) -> ToolOutcome {
159        let outcome = match self.call_inner(name, args).await {
160            Ok(out) => out,
161            Err(e) => ToolOutcome::err(e.to_string()),
162        };
163        self.tag_outcome_with_context(outcome).await
164    }
165
166    async fn tag_outcome_with_context(&self, mut outcome: ToolOutcome) -> ToolOutcome {
167        let (connection_id, database) = self.session.active_context().await;
168        let access_mode = match self.session.access_mode() {
169            nexql_policy::AccessMode::Read => "read",
170            nexql_policy::AccessMode::Write => "write",
171            nexql_policy::AccessMode::Admin => "admin",
172        };
173        let mut freshness: Option<serde_json::Value> = None;
174        if let Some(store) = self.session.index_store.as_ref() {
175            let base = store.base_dir(&connection_id, &database);
176            if let Ok(Some(manifest)) = store.read_manifest(&base) {
177                let stale = self.session.is_index_stale(&connection_id, &database);
178                let mut freshness_obj = json!({
179                    "indexedAt": manifest.indexed_at,
180                    "schemaFingerprint": manifest.schema_fingerprint,
181                    "stale": stale,
182                });
183                if stale {
184                    freshness_obj["reason"] = json!("schema_changed");
185                }
186                freshness = Some(freshness_obj);
187            } else {
188                freshness = Some(json!({ "stale": true, "reason": "no_index" }));
189            }
190        }
191        if let Some(ref mut structured) = outcome.structured
192            && let Some(obj) = structured.as_object_mut()
193        {
194            if !obj.contains_key("connectionId") {
195                obj.insert("connectionId".into(), json!(connection_id));
196            }
197            if !obj.contains_key("database") {
198                obj.insert("database".into(), json!(database));
199            }
200            if !obj.contains_key("accessMode") {
201                obj.insert("accessMode".into(), json!(access_mode));
202            }
203            if let Some(ref f) = freshness {
204                obj.insert("freshness".into(), f.clone());
205            }
206        }
207        let header = format!(
208            "[context connectionId={connection_id} database={database} accessMode={access_mode}]\n"
209        );
210        if !outcome.text.starts_with("[context ") {
211            outcome.text = format!("{header}{}", outcome.text);
212        }
213        outcome
214    }
215
216    async fn call_inner(&self, name: &str, args: Value) -> Result<ToolOutcome, ToolError> {
217        let tool = ToolName::parse(name).ok_or_else(|| ToolError::Unknown(name.to_string()))?;
218        match tool {
219            ToolName::ListConnections => Ok(self.list_connections()),
220            ToolName::ListDatabases => self.list_databases(&args).await,
221            ToolName::ListSchemas => self.list_schemas().await,
222            ToolName::ListObjects => self.list_objects(&args).await,
223            ToolName::GetCurrentContext => self.get_current_context().await,
224            ToolName::SwitchConnection => self.switch_connection(&args).await,
225            ToolName::RunSelect => self.run_select(&args).await,
226            ToolName::ExplainQuery => self.explain_query(&args).await,
227            ToolName::SearchSchema => self.search_schema(&args).await,
228            ToolName::DescribeObject => self.describe_object(&args).await,
229            ToolName::GetJoinPath => self.get_join_path(&args).await,
230            ToolName::SampleValues => self.sample_values(&args).await,
231            ToolName::GetDdl => self.get_ddl(&args).await,
232            ToolName::TableStats => self.table_stats(&args).await,
233            ToolName::IndexUsage => self.index_usage(&args).await,
234            ToolName::ListRunningQueries => self.list_running_queries().await,
235            ToolName::FindBlockingLocks => self.find_blocking_locks().await,
236            ToolName::SlowQueries => self.slow_queries(&args).await,
237            ToolName::DbHealthCheck => self.db_health_check().await,
238            ToolName::ExplainAnalyze => self.explain_analyze(&args).await,
239            ToolName::AnalyzeQueryPlan => self.analyze_query_plan(&args).await,
240            ToolName::GetIndexStatus => self.get_index_status().await,
241            ToolName::ListExtensions => self.list_extensions().await,
242            ToolName::ServerSettings => self.server_settings().await,
243            ToolName::SuggestIndexes => self.suggest_indexes(&args).await,
244            ToolName::FindUnusedIndexes => self.find_unused_indexes(&args).await,
245            ToolName::BloatReport => self.bloat_report(&args).await,
246            ToolName::FindMissingFks => self.find_missing_fks(&args).await,
247            ToolName::ExportQuery => self.export_query(&args).await,
248            ToolName::ListRoles => self.list_roles(&args).await,
249            ToolName::DbDashboard => self.db_dashboard().await,
250            ToolName::DeepPlanAnalysis => self.deep_plan_analysis(&args).await,
251            ToolName::SchemaDiff => self.schema_diff(&args).await,
252            ToolName::GenerateMigration => self.generate_migration(&args).await,
253            ToolName::ExecuteSql => self.execute_sql_tool(&args).await,
254            ToolName::EditRow => self.edit_row_tool(&args).await,
255            ToolName::ImportData => self.import_data_tool(&args).await,
256            ToolName::ApplyDdl => self.apply_ddl_tool(&args).await,
257            ToolName::CreateIndexConcurrently => self.create_index_concurrently_tool(&args).await,
258            ToolName::RunMaintenance => self.run_maintenance_tool(&args).await,
259            ToolName::TerminateQuery => self.terminate_query_tool(&args).await,
260            ToolName::ResolveTarget => self.resolve_target(&args).await,
261            ToolName::DiscoverTools => self.discover_tools(&args).await,
262            ToolName::AutoTuneQuery => self.auto_tune_query(&args).await,
263            ToolName::CheckDdlSafety => self.check_ddl_safety_tool(&args).await,
264            ToolName::RebuildIndex => self.rebuild_index_tool(&args).await,
265            ToolName::RefreshIndex => self.refresh_index_tool(&args).await,
266            ToolName::RunDoctor => self.run_doctor_tool().await,
267            ToolName::SetupConnection => self.setup_connection_tool(&args).await,
268            ToolName::SaveProfile => self.save_profile_tool(&args).await,
269            ToolName::TestProfile => self.test_profile_tool(&args).await,
270            ToolName::ExportProfile => self.export_profile_tool(&args).await,
271            ToolName::ImportProfile => self.import_profile_tool(&args).await,
272        }
273    }
274
275    fn require_write(&self) -> Result<(), ToolError> {
276        if !self.session.access_mode().allows_writes() {
277            return Err(ToolError::Execution(
278                "write tools require --access-mode write or admin (current session: read)".into(),
279            ));
280        }
281        Ok(())
282    }
283
284    fn require_admin(&self) -> Result<(), ToolError> {
285        if !self.session.access_mode().allows_admin() {
286            return Err(ToolError::Execution(
287                "admin tools require --access-mode admin".into(),
288            ));
289        }
290        Ok(())
291    }
292
293    async fn execute_sql_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
294        self.require_write()?;
295        let sql = args
296            .get("sql")
297            .and_then(|v| v.as_str())
298            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
299        let dry_run = args
300            .get("dry_run")
301            .and_then(|v| v.as_bool())
302            .unwrap_or(false);
303        execute_sql(&self.session, sql, dry_run).await
304    }
305
306    async fn edit_row_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
307        self.require_write()?;
308        edit_row(&self.session, args).await
309    }
310
311    async fn import_data_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
312        self.require_write()?;
313        import_data(&self.session, args).await
314    }
315
316    async fn apply_ddl_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
317        self.require_admin()?;
318        let sql = args
319            .get("sql")
320            .and_then(|v| v.as_str())
321            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
322        let dry_run = args
323            .get("dry_run")
324            .and_then(|v| v.as_bool())
325            .unwrap_or(false);
326        apply_ddl(&self.session, sql, dry_run).await
327    }
328
329    async fn create_index_concurrently_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
330        self.require_admin()?;
331        let sql = args
332            .get("sql")
333            .and_then(|v| v.as_str())
334            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
335        create_index_concurrently(&self.session, sql).await
336    }
337
338    async fn run_maintenance_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
339        self.require_admin()?;
340        run_maintenance(&self.session, args).await
341    }
342
343    async fn terminate_query_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
344        self.require_admin()?;
345        terminate_query(&self.session, args).await
346    }
347
348    /// Autonomously resolve which connection/database matches a free-text `hint` and/or
349    /// `objectHint`, then switch the session context to it.
350    async fn resolve_target(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
351        let hint = args
352            .get("hint")
353            .and_then(|v| v.as_str())
354            .map(str::trim)
355            .filter(|s| !s.is_empty());
356        let object_hint = args
357            .get("objectHint")
358            .and_then(|v| v.as_str())
359            .map(str::trim)
360            .filter(|s| !s.is_empty());
361        if hint.is_none() && object_hint.is_none() {
362            return Err(ToolError::InvalidArgs(
363                "At least one of \"hint\" or \"objectHint\" is required.".into(),
364            ));
365        }
366
367        let connections = self.session.connections();
368        if connections.is_empty() {
369            return Ok(ToolOutcome::err("No connections configured."));
370        }
371
372        #[derive(Clone)]
373        struct Candidate {
374            connection_id: String,
375            database: String,
376        }
377        fn key_of(c: &Candidate) -> String {
378            format!("{}\u{0}{}", c.connection_id, c.database)
379        }
380
381        let indexed: Vec<(String, String)> = self
382            .index_store()
383            .map(|store| store.list_indexed_databases().unwrap_or_default())
384            .unwrap_or_default();
385
386        let mut seen = std::collections::HashSet::new();
387        let mut candidates: Vec<Candidate> = Vec::new();
388        let mut add_candidate = |connection_id: &str, database: &str| {
389            if !connections.iter().any(|c| c.id == connection_id) {
390                return;
391            }
392            let key = format!("{connection_id}\u{0}{database}");
393            if !seen.insert(key) {
394                return;
395            }
396            candidates.push(Candidate {
397                connection_id: connection_id.to_string(),
398                database: database.to_string(),
399            });
400        };
401        for (cid, db) in &indexed {
402            add_candidate(cid, db);
403        }
404        for c in &connections {
405            let db = c.database.clone().unwrap_or_else(|| "postgres".into());
406            add_candidate(&c.id, &db);
407        }
408
409        let mut scored: std::collections::HashMap<String, (Candidate, f64, Vec<String>)> =
410            std::collections::HashMap::new();
411
412        if let Some(hint) = hint {
413            for c in &candidates {
414                let Some(conn) = connections.iter().find(|x| x.id == c.connection_id) else {
415                    continue;
416                };
417                let fields: [(&str, &str); 3] = [
418                    ("connection name", conn.name.as_str()),
419                    ("host", conn.host.as_deref().unwrap_or("")),
420                    ("database", c.database.as_str()),
421                ];
422                let mut best = 0.0f64;
423                let mut best_field = "";
424                for (label, value) in fields {
425                    let s = fuzzy_score(hint, value);
426                    if s > best {
427                        best = s;
428                        best_field = label;
429                    }
430                }
431                if best > 0.0 {
432                    let entry = scored
433                        .entry(key_of(c))
434                        .or_insert_with(|| (c.clone(), 0.0, Vec::new()));
435                    entry.1 += best;
436                    entry
437                        .2
438                        .push(format!("{best_field} matched hint \"{hint}\" ({best:.0})"));
439                }
440            }
441        }
442
443        if let Some(object_hint) = object_hint
444            && let Some(store) = self.index_store()
445        {
446            let filter = self.query_filter();
447            for (cid, db) in &indexed {
448                let svc = IndexQueryService::new(store, cid.clone(), db.clone());
449                if let Ok(hits) = svc.search_schema(
450                    object_hint,
451                    3,
452                    Some(&filter),
453                    SearchOptions {
454                        use_semantic: self.use_semantic,
455                        embedder: self.embedder.as_deref(),
456                    },
457                ) && let Some(top) = hits.first()
458                {
459                    let c = Candidate {
460                        connection_id: cid.clone(),
461                        database: db.clone(),
462                    };
463                    let entry = scored
464                        .entry(key_of(&c))
465                        .or_insert_with(|| (c.clone(), 0.0, Vec::new()));
466                    entry.1 += top.score * 10.0;
467                    entry.2.push(format!(
468                        "schema search for \"{object_hint}\" found {} (score {:.2})",
469                        top.ref_, top.score
470                    ));
471                }
472            }
473        }
474
475        let mut ranked: Vec<(Candidate, f64, Vec<String>)> = scored.into_values().collect();
476        ranked.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
477
478        if ranked.is_empty() {
479            let candidates_json: Vec<Value> = connections
480                .iter()
481                .map(|c| {
482                    json!({
483                        "connectionId": c.id,
484                        "connectionName": c.name,
485                        "database": c.database.clone().unwrap_or_else(|| "postgres".into()),
486                    })
487                })
488                .collect();
489            return Ok(ToolOutcome::ok_json(json!({
490                "ambiguous": true,
491                "message": format!(
492                    "No connection/database matched \"{}\". Choose from the configured connections.",
493                    hint.or(object_hint).unwrap_or_default()
494                ),
495                "candidates": candidates_json
496            })));
497        }
498
499        let winner = &ranked[0];
500        let is_tied = ranked
501            .get(1)
502            .is_some_and(|runner_up| runner_up.1 >= winner.1 * 0.85);
503
504        if is_tied {
505            let threshold = winner.1 * 0.85;
506            let tied: Vec<&(Candidate, f64, Vec<String>)> =
507                ranked.iter().filter(|r| r.1 >= threshold).take(5).collect();
508            let candidates_json: Vec<Value> = tied
509                .iter()
510                .filter_map(|(c, score, evidence)| {
511                    connections
512                        .iter()
513                        .find(|x| x.id == c.connection_id)
514                        .map(|conn| {
515                            json!({
516                                "connectionId": c.connection_id,
517                                "connectionName": conn.name,
518                                "database": c.database,
519                                "score": score,
520                                "evidence": evidence,
521                            })
522                        })
523                })
524                .collect();
525            return Ok(ToolOutcome::ok_json(json!({
526                "ambiguous": true,
527                "message": format!("{} equally-plausible candidates matched.", tied.len()),
528                "candidates": candidates_json
529            })));
530        }
531
532        let (winner_candidate, winner_score, winner_evidence) = winner;
533
534        if let Some(object_hint) = object_hint
535            && let Some(store) = self.index_store()
536        {
537            let filter = self.query_filter();
538            let svc = IndexQueryService::new(
539                store,
540                &winner_candidate.connection_id,
541                &winner_candidate.database,
542            );
543            if let Ok(hits) = svc.search_schema(
544                object_hint,
545                5,
546                Some(&filter),
547                SearchOptions {
548                    use_semantic: self.use_semantic,
549                    embedder: self.embedder.as_deref(),
550                },
551            ) && hits.len() >= 2
552            {
553                let top_score = hits[0].score;
554                let tied: Vec<&nexql_index::RankedHit> = hits
555                    .iter()
556                    .filter(|h| scores_equal(h.score, top_score))
557                    .collect();
558                if tied.len() > 1 {
559                    let candidates_json: Vec<Value> = tied
560                        .iter()
561                        .map(|h| {
562                            json!({
563                                "ref": h.ref_,
564                                "score": h.score,
565                                "kind": h.kind,
566                                "connectionId": winner_candidate.connection_id,
567                                "database": winner_candidate.database,
568                            })
569                        })
570                        .collect();
571                    return Ok(ToolOutcome::ok_json(json!({
572                        "ambiguous": true,
573                        "message": format!(
574                            "{} objects matched \"{object_hint}\" with equal scores — choose explicitly.",
575                            tied.len()
576                        ),
577                        "candidates": candidates_json,
578                    })));
579                }
580            }
581        }
582
583        self.session
584            .switch(
585                &winner_candidate.connection_id,
586                Some(winner_candidate.database.clone()),
587            )
588            .await?;
589        let conn = connections
590            .iter()
591            .find(|x| x.id == winner_candidate.connection_id)
592            .ok_or_else(|| ToolError::Execution("resolved connection vanished".into()))?;
593
594        Ok(ToolOutcome::ok_json(json!({
595            "resolved": true,
596            "connectionId": winner_candidate.connection_id,
597            "connectionName": conn.name,
598            "database": winner_candidate.database,
599            "confidence": winner_score,
600            "evidence": winner_evidence,
601        })))
602    }
603
604    async fn discover_tools(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
605        let query = args
606            .get("query")
607            .and_then(|v| v.as_str())
608            .map(str::to_lowercase);
609        let category = args
610            .get("category")
611            .and_then(|v| v.as_str())
612            .map(str::to_lowercase);
613
614        // Always search the full catalog — meta profile may expose only a subset via tools/list.
615        let all_specs = active_tools();
616        let filtered: Vec<Value> = all_specs
617            .into_iter()
618            .filter(|spec| {
619                if spec.name == ToolName::DiscoverTools {
620                    return false;
621                }
622                if let Some(ref cat) = category {
623                    match cat.as_str() {
624                        "query" if !ToolName::QUERY_PROFILE.contains(&spec.name) => return false,
625                        "dba" if !ToolName::DBA_PROFILE.contains(&spec.name) => return false,
626                        "write" if !ToolName::PHASE9.contains(&spec.name) => return false,
627                        _ => {}
628                    }
629                }
630                if let Some(ref q) = query {
631                    let name_match = spec.name.as_str().contains(q.as_str());
632                    let desc_match = spec.description.to_lowercase().contains(q.as_str());
633                    if !name_match && !desc_match {
634                        return false;
635                    }
636                }
637                true
638            })
639            .map(|spec| {
640                json!({
641                    "name": spec.name.as_str(),
642                    "description": spec.description,
643                    "input_schema": spec.input_schema,
644                })
645            })
646            .collect();
647
648        Ok(ToolOutcome::ok_json(json!({
649            "query": args.get("query"),
650            "category": args.get("category"),
651            "count": filtered.len(),
652            "tools": filtered,
653        })))
654    }
655
656    fn build_tuning_summary(plan_structured: &Option<Value>, suggestions: &Value) -> String {
657        let mut parts = Vec::new();
658        if let Some(structured) = plan_structured
659            && let Some(metrics) = structured.get("metrics")
660        {
661            if let Some(exec_time) = metrics.get("executionTime").and_then(|v| v.as_f64()) {
662                parts.push(format!("Query executed in {:.2}ms.", exec_time));
663            }
664            if let Some(seq_scans) = metrics.get("sequentialScans").and_then(|v| v.as_u64())
665                && seq_scans > 0
666            {
667                parts.push(format!("Found {seq_scans} sequential scan(s)."));
668            }
669        }
670
671        let candidate_count = suggestions
672            .get("high_seq_scan_tables")
673            .and_then(|v| v.as_array())
674            .map(|a| a.len())
675            .unwrap_or(0)
676            + suggestions
677                .get("unindexed_fk_columns")
678                .and_then(|v| v.as_array())
679                .map(|a| a.len())
680                .unwrap_or(0);
681
682        if candidate_count > 0 {
683            parts.push(format!(
684                "{candidate_count} index recommendation(s) identified."
685            ));
686        } else {
687            parts.push("No explicit index candidate recommendations generated.".into());
688        }
689
690        if parts.is_empty() {
691            "Auto-tune evaluation complete. Inspect execution plan and index recommendations."
692                .into()
693        } else {
694            parts.join(" ")
695        }
696    }
697
698    async fn auto_tune_query(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
699        let sql = args
700            .get("sql")
701            .and_then(|v| v.as_str())
702            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
703
704        let deep_plan = self
705            .deep_plan_analysis(&json!({ "sql": sql, "analyze": true }))
706            .await?;
707
708        let suggestions_res = self.suggest_indexes(&json!({ "sql": sql })).await;
709        let (suggestions_data, suggestions_error) = match suggestions_res {
710            Ok(outcome) => (outcome.structured.unwrap_or(json!([])), None),
711            Err(e) => (json!([]), Some(e.to_string())),
712        };
713
714        let summary_text = Self::build_tuning_summary(&deep_plan.structured, &suggestions_data);
715
716        let mut payload = json!({
717            "target_query": sql,
718            "deep_plan_analysis": deep_plan.structured,
719            "index_suggestions": suggestions_data,
720            "tuning_summary": summary_text,
721        });
722
723        if let Some(err) = suggestions_error {
724            payload["suggestions_error"] = json!(err);
725        }
726
727        Ok(ToolOutcome::ok_json(payload))
728    }
729
730    async fn check_ddl_safety_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
731        let ddl = args
732            .get("ddl")
733            .and_then(|v| v.as_str())
734            .ok_or_else(|| ToolError::InvalidArgs("ddl is required".into()))?;
735
736        let report = crate::dba_guard::analyze_ddl_safety(ddl);
737        Ok(ToolOutcome::ok_json(report))
738    }
739
740    async fn rebuild_index_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
741        let store = self
742            .index_store()
743            .ok_or_else(|| ToolError::Execution("Index store unavailable".into()))?;
744        let (connection_id, database) = self.session.active_context().await;
745        let depth_str = args
746            .get("depth")
747            .and_then(|v| v.as_str())
748            .unwrap_or("structure");
749        let depth: BuildDepth = match depth_str.to_lowercase().as_str() {
750            "profiles" | "full" => BuildDepth::Profiles,
751            _ => BuildDepth::Structure,
752        };
753
754        let req = BuildRequest {
755            connection_id: connection_id.clone(),
756            database: database.clone(),
757            scope: IndexScope {
758                included_schemas: vec![],
759                excluded_objects: vec![],
760                pii_excluded_columns: vec![],
761            },
762            depth,
763            build_mode: BuildMode::Guided,
764            environment: "development".into(),
765            embeddings: self.use_semantic,
766        };
767
768        let client = self.session.checkout().await?;
769        let db = PgCatalogDb::new(&client);
770        let manifest = build_index(store, &db, &req, None, None, self.embedder.as_deref())
771            .await
772            .map_err(|e| ToolError::Execution(format!("Index build failed: {e}")))?;
773        self.session
774            .clear_index_stale(&connection_id, &database);
775
776        Ok(ToolOutcome::ok_json(json!({
777            "status": "completed",
778            "connection_id": connection_id,
779            "database": database,
780            "schema_fingerprint": manifest.schema_fingerprint,
781            "counts": manifest.counts,
782            "build_ms": manifest.stats.build_ms,
783        })))
784    }
785
786    async fn refresh_index_tool(&self, _args: &Value) -> Result<ToolOutcome, ToolError> {
787        let store = self
788            .index_store()
789            .ok_or_else(|| ToolError::Execution("Index store unavailable".into()))?;
790        let (connection_id, database) = self.session.active_context().await;
791        let base = store.base_dir(&connection_id, &database);
792        let manifest = store.read_manifest(&base)?.ok_or_else(|| {
793            ToolError::Execution(
794                "No existing index manifest to refresh — call 'rebuild_index'.".into(),
795            )
796        })?;
797
798        let req = BuildRequest {
799            connection_id: connection_id.clone(),
800            database: database.clone(),
801            scope: manifest.scope,
802            depth: manifest.build_depth,
803            build_mode: manifest.build_mode,
804            environment: manifest.environment,
805            embeddings: self.use_semantic,
806        };
807
808        let client = self.session.checkout().await?;
809        let db = PgCatalogDb::new(&client);
810        let new_manifest = build_index(store, &db, &req, None, None, self.embedder.as_deref())
811            .await
812            .map_err(|e| ToolError::Execution(format!("Index refresh failed: {e}")))?;
813        self.session
814            .clear_index_stale(&connection_id, &database);
815
816        Ok(ToolOutcome::ok_json(json!({
817            "status": "refreshed",
818            "connection_id": connection_id,
819            "database": database,
820            "schema_fingerprint": new_manifest.schema_fingerprint,
821            "counts": new_manifest.counts,
822            "build_ms": new_manifest.stats.build_ms,
823        })))
824    }
825
826    async fn run_doctor_tool(&self) -> Result<ToolOutcome, ToolError> {
827        let (connection_id, database) = self.session.active_context().await;
828        let client = self.session.checkout().await?;
829
830        let version: String = client
831            .query_one("SELECT version()", &[])
832            .await
833            .map_err(|e| ToolError::Execution(e.to_string()))?
834            .get(0);
835
836        let is_super: String = client
837            .query_one("SELECT current_setting('is_superuser')", &[])
838            .await
839            .map_err(|e| ToolError::Execution(e.to_string()))?
840            .get(0);
841        let is_superuser = is_super.eq_ignore_ascii_case("on");
842
843        let ro: String = client
844            .query_one("SHOW default_transaction_read_only", &[])
845            .await
846            .map_err(|e| ToolError::Execution(e.to_string()))?
847            .get(0);
848
849        let timeout: String = client
850            .query_one("SHOW statement_timeout", &[])
851            .await
852            .map_err(|e| ToolError::Execution(e.to_string()))?
853            .get(0);
854
855        let pgs_present: bool = match client
856            .query_one(
857                "SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_stat_statements')",
858                &[],
859            )
860            .await
861        {
862            Ok(row) => row.get(0),
863            Err(_) => false,
864        };
865
866        let index_status = if let Some(store) = self.index_store() {
867            let base = store.base_dir(&connection_id, &database);
868            match store.read_manifest(&base) {
869                Ok(Some(m)) => json!({
870                    "present": true,
871                    "indexed_at": m.indexed_at,
872                    "fingerprint": m.schema_fingerprint,
873                    "tables": m.counts.tables,
874                }),
875                _ => json!({ "present": false }),
876            }
877        } else {
878            json!({ "present": false, "reason": "no_index_store" })
879        };
880
881        let recent_errors = read_recent_log_errors();
882
883        Ok(ToolOutcome::ok_json(json!({
884            "status": "ok",
885            "connection_id": connection_id,
886            "database": database,
887            "version": version.split(',').next().unwrap_or(&version),
888            "access_mode": format!("{:?}", self.session.access_mode()),
889            "superuser": is_superuser,
890            "read_only": ro,
891            "statement_timeout": timeout,
892            "pg_stat_statements": pgs_present,
893            "index": index_status,
894            "recent_errors": recent_errors,
895        })))
896    }
897
898    fn register_profile_in_session(
899        &self,
900        name: &str,
901        profile: &nexql_conn::ProfileConfig,
902    ) -> Result<(), ToolError> {
903        self.session.register_profile(
904            name,
905            profile,
906            self.session.access_mode(),
907            self.session.caps(),
908        )
909    }
910
911    async fn setup_connection_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
912        let profile_name = args
913            .get("name")
914            .and_then(|v| v.as_str())
915            .unwrap_or("default");
916
917        let candidates = crate::detect::ConnectionDetector::detect_all(None);
918
919        let url = args.get("url").and_then(|v| v.as_str());
920        let host = args.get("host").and_then(|v| v.as_str());
921        let port = args.get("port").and_then(|v| v.as_u64()).map(|n| n as u16);
922        let dbname = args.get("dbname").and_then(|v| v.as_str());
923        let user = args.get("user").and_then(|v| v.as_str());
924        let password = args.get("password").and_then(|v| v.as_str());
925        let sslmode = args.get("sslmode").and_then(|v| v.as_str());
926
927        let best_cand = candidates
928            .iter()
929            .find(|c| c.is_complete)
930            .or_else(|| candidates.first());
931
932        let res_host = host.or_else(|| best_cand.and_then(|c| c.host.as_deref()));
933        let res_port = port.or_else(|| best_cand.and_then(|c| c.port));
934        let res_dbname = dbname.or_else(|| best_cand.and_then(|c| c.dbname.as_deref()));
935        let res_user = user.or_else(|| best_cand.and_then(|c| c.user.as_deref()));
936        let res_password = password.or_else(|| best_cand.and_then(|c| c.password.as_deref()));
937        let res_url = url.or_else(|| best_cand.and_then(|c| c.url.as_deref()));
938        let res_sslmode = sslmode.or_else(|| best_cand.and_then(|c| c.sslmode.as_deref()));
939
940        if res_url.is_none() && (res_host.is_none() || res_dbname.is_none() || res_user.is_none()) {
941            let missing: Vec<&str> = vec![
942                if res_host.is_none() {
943                    Some("host")
944                } else {
945                    None
946                },
947                if res_dbname.is_none() {
948                    Some("dbname")
949                } else {
950                    None
951                },
952                if res_user.is_none() {
953                    Some("user")
954                } else {
955                    None
956                },
957            ]
958            .into_iter()
959            .flatten()
960            .collect();
961
962            return Ok(ToolOutcome::ok_json(json!({
963                "status": "needs_input",
964                "message": "Insufficient connection details. Please supply missing fields.",
965                "detectedCandidates": candidates.iter().map(|c| c.redacted_json()).collect::<Vec<_>>(),
966                "missingFields": missing
967            })));
968        }
969
970        let params = nexql_conn::ConnectionParams {
971            url: res_url.map(String::from),
972            host: res_host.map(String::from),
973            port: res_port,
974            dbname: res_dbname.map(String::from),
975            user: res_user.map(String::from),
976            password: res_password.map(String::from),
977            sslmode: res_sslmode.map(String::from),
978            ..Default::default()
979        };
980
981        match nexql_conn::test_connection(&params).await {
982            Ok(report) => {
983                let p_config = nexql_conn::ProfileConfig {
984                    url: params.url.clone(),
985                    host: params.host.clone(),
986                    port: params.port,
987                    dbname: params.dbname.clone(),
988                    user: params.user.clone(),
989                    password: params.password.clone(),
990                    sslmode: params.sslmode.clone(),
991                    ..Default::default()
992                };
993
994                let path = nexql_conn::ConfigFile::default_path().ok_or_else(|| {
995                    ToolError::Execution("Could not resolve config directory".into())
996                })?;
997                let mut cfg = nexql_conn::ConfigFile::load_path(&path).unwrap_or_default();
998                cfg.upsert_profile(profile_name, p_config.clone());
999                let backup = cfg
1000                    .save(&path)
1001                    .map_err(|e| ToolError::Execution(e.to_string()))?;
1002                self.register_profile_in_session(profile_name, &p_config)?;
1003
1004                Ok(ToolOutcome::ok_json(json!({
1005                    "status": "configured",
1006                    "profileName": profile_name,
1007                    "serverVersion": report.server_version,
1008                    "isSuperuser": report.is_superuser,
1009                    "latencyMs": report.latency.as_millis(),
1010                    "configPath": path.to_string_lossy().to_string(),
1011                    "backup": backup.map(|b| b.to_string_lossy().to_string()),
1012                    "sessionReloaded": true,
1013                })))
1014            }
1015            Err(e) => Ok(ToolOutcome::ok_json(json!({
1016                "status": "failed",
1017                "error": e.to_string(),
1018                "detectedCandidates": candidates.iter().map(|c| c.redacted_json()).collect::<Vec<_>>()
1019            }))),
1020        }
1021    }
1022
1023    async fn save_profile_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1024        let name = args
1025            .get("name")
1026            .and_then(|v| v.as_str())
1027            .ok_or_else(|| ToolError::InvalidArgs("name parameter is required".into()))?;
1028
1029        let p_config = nexql_conn::ProfileConfig {
1030            url: args.get("url").and_then(|v| v.as_str()).map(String::from),
1031            host: args.get("host").and_then(|v| v.as_str()).map(String::from),
1032            port: args.get("port").and_then(|v| v.as_u64()).map(|n| n as u16),
1033            dbname: args
1034                .get("dbname")
1035                .and_then(|v| v.as_str())
1036                .map(String::from),
1037            user: args.get("user").and_then(|v| v.as_str()).map(String::from),
1038            password: args
1039                .get("password")
1040                .and_then(|v| v.as_str())
1041                .map(String::from),
1042            sslmode: args
1043                .get("sslmode")
1044                .and_then(|v| v.as_str())
1045                .map(String::from),
1046            access_mode: args
1047                .get("access_mode")
1048                .and_then(|v| v.as_str())
1049                .map(String::from),
1050            max_rows: args
1051                .get("max_rows")
1052                .and_then(|v| v.as_u64())
1053                .map(|n| n as u32),
1054            ..Default::default()
1055        };
1056
1057        let path = nexql_conn::ConfigFile::default_path()
1058            .ok_or_else(|| ToolError::Execution("Could not resolve config directory".into()))?;
1059
1060        let mut cfg = nexql_conn::ConfigFile::load_path(&path).unwrap_or_default();
1061        cfg.upsert_profile(name, p_config.clone());
1062        let backup = cfg
1063            .save(&path)
1064            .map_err(|e| ToolError::Execution(e.to_string()))?;
1065        self.register_profile_in_session(name, &p_config)?;
1066
1067        Ok(ToolOutcome::ok_json(json!({
1068            "status": "saved",
1069            "profile": name,
1070            "configPath": path.to_string_lossy().to_string(),
1071            "backup": backup.map(|b| b.to_string_lossy().to_string()),
1072            "sessionReloaded": true,
1073        })))
1074    }
1075
1076    async fn test_profile_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1077        let name = args.get("name").and_then(|v| v.as_str());
1078
1079        let params = if let Some(pname) = name {
1080            let conn = self
1081                .session
1082                .connections()
1083                .into_iter()
1084                .find(|c| c.id == pname)
1085                .ok_or_else(|| ToolError::InvalidArgs(format!("Profile '{pname}' not found")))?;
1086            conn.params.clone()
1087        } else {
1088            nexql_conn::ConnectionParams {
1089                url: args.get("url").and_then(|v| v.as_str()).map(String::from),
1090                host: args.get("host").and_then(|v| v.as_str()).map(String::from),
1091                port: args.get("port").and_then(|v| v.as_u64()).map(|n| n as u16),
1092                dbname: args
1093                    .get("dbname")
1094                    .and_then(|v| v.as_str())
1095                    .map(String::from),
1096                user: args.get("user").and_then(|v| v.as_str()).map(String::from),
1097                password: args
1098                    .get("password")
1099                    .and_then(|v| v.as_str())
1100                    .map(String::from),
1101                sslmode: args
1102                    .get("sslmode")
1103                    .and_then(|v| v.as_str())
1104                    .map(String::from),
1105                ..Default::default()
1106            }
1107        };
1108
1109        match nexql_conn::test_connection(&params).await {
1110            Ok(report) => Ok(ToolOutcome::ok_json(json!({
1111                "success": true,
1112                "serverVersion": report.server_version,
1113                "isSuperuser": report.is_superuser,
1114                "latencyMs": report.latency.as_millis()
1115            }))),
1116            Err(e) => Ok(ToolOutcome::ok_json(json!({
1117                "success": false,
1118                "error": e.to_string()
1119            }))),
1120        }
1121    }
1122
1123    async fn export_profile_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1124        let format = args
1125            .get("format")
1126            .and_then(|v| v.as_str())
1127            .unwrap_or("full");
1128        let path = nexql_conn::ConfigFile::default_path()
1129            .ok_or_else(|| ToolError::Execution("Could not resolve config directory".into()))?;
1130        let cfg = nexql_conn::ConfigFile::load_path(&path).unwrap_or_default();
1131
1132        if format == "project" {
1133            let proj = cfg.export_shareable();
1134            let toml_str =
1135                toml::to_string_pretty(&proj).map_err(|e| ToolError::Execution(e.to_string()))?;
1136            Ok(ToolOutcome::ok_json(json!({
1137                "format": "project",
1138                "filename": ".nexql/config.toml",
1139                "description": "Project policy overlay (no credentials). Use format=full for shareable connection profiles.",
1140                "content": toml_str,
1141            })))
1142        } else {
1143            let sanitized = cfg.export_full_sanitized();
1144            let toml_str = sanitized
1145                .to_toml_string()
1146                .map_err(|e| ToolError::Execution(e.to_string()))?;
1147            Ok(ToolOutcome::ok_json(json!({
1148                "format": "full",
1149                "description": "Full user config with passwords and secrets stripped — suitable for team sharing.",
1150                "profileCount": sanitized.profiles.len(),
1151                "content": toml_str,
1152            })))
1153        }
1154    }
1155
1156    async fn import_profile_tool(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1157        let content = if let Some(c) = args.get("content").and_then(|v| v.as_str()) {
1158            c.to_string()
1159        } else if let Some(p) = args.get("path").and_then(|v| v.as_str()) {
1160            std::fs::read_to_string(p)
1161                .map_err(|e| ToolError::Execution(format!("failed to read file {p}: {e}")))?
1162        } else {
1163            return Err(ToolError::Execution(
1164                "either 'content' or 'path' must be specified".into(),
1165            ));
1166        };
1167
1168        let path = nexql_conn::ConfigFile::default_path()
1169            .ok_or_else(|| ToolError::Execution("Could not resolve config directory".into()))?;
1170        let mut cfg = nexql_conn::ConfigFile::load_path(&path).unwrap_or_default();
1171
1172        let imported: nexql_conn::ConfigFile = toml::from_str(&content)
1173            .map_err(|e| ToolError::Execution(format!("failed to parse TOML content: {e}")))?;
1174
1175        let mut count = 0;
1176        let mut imported_names: Vec<String> = Vec::new();
1177        for (name, prof) in imported.profiles {
1178            cfg.upsert_profile(name.clone(), prof.clone());
1179            self.register_profile_in_session(&name, &prof)?;
1180            imported_names.push(name);
1181            count += 1;
1182        }
1183        if imported.default_profile.is_some() {
1184            cfg.default_profile = imported.default_profile;
1185        }
1186
1187        let backup = cfg
1188            .save(&path)
1189            .map_err(|e| ToolError::Execution(e.to_string()))?;
1190
1191        Ok(ToolOutcome::ok_json(json!({
1192            "status": "imported",
1193            "imported_profiles": count,
1194            "profiles": imported_names,
1195            "configPath": path.to_string_lossy().to_string(),
1196            "backup": backup.map(|b| b.to_string_lossy().to_string()),
1197            "sessionReloaded": true,
1198        })))
1199    }
1200
1201    async fn index_service(&self) -> Result<(&IndexStore, String, String), ToolError> {
1202        let store = self
1203            .index_store()
1204            .ok_or_else(|| ToolError::Execution(NO_INDEX_HINT.into()))?;
1205        let (connection_id, database) = self.session.active_context().await;
1206        let base = store.base_dir(&connection_id, &database);
1207        if store.read_manifest(&base)?.is_none() {
1208            return Err(ToolError::Execution(format!(
1209                "No schema index for database \"{database}\" — call the 'rebuild_index' tool to build an index."
1210            )));
1211        }
1212        Ok((store, connection_id, database))
1213    }
1214
1215    async fn search_schema(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1216        let query = args
1217            .get("query")
1218            .and_then(|v| v.as_str())
1219            .unwrap_or("")
1220            .trim();
1221        if query.is_empty() {
1222            return Ok(ToolOutcome::ok_json(json!([])));
1223        }
1224        let (store, connection_id, database) = self.index_service().await?;
1225        let svc = IndexQueryService::new(store, &connection_id, &database);
1226        let filter = self.query_filter();
1227        let hits = svc.search_schema(
1228            query,
1229            SEARCH_SCHEMA_LIMIT,
1230            Some(&filter),
1231            SearchOptions {
1232                use_semantic: self.use_semantic,
1233                embedder: self.embedder.as_deref(),
1234            },
1235        )?;
1236        let rows: Vec<Value> = hits
1237            .into_iter()
1238            .map(|h| {
1239                json!({
1240                    "ref": h.ref_,
1241                    "score": h.score,
1242                    "kind": h.kind,
1243                })
1244            })
1245            .collect();
1246        Ok(ToolOutcome::ok_json(json!(rows)))
1247    }
1248
1249    async fn describe_object(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1250        let ref_ = args
1251            .get("ref")
1252            .and_then(|v| v.as_str())
1253            .ok_or_else(|| ToolError::InvalidArgs("ref is required".into()))?;
1254        let (store, connection_id, database) = self.index_service().await?;
1255        let svc = IndexQueryService::new(store, &connection_id, &database);
1256        let filter = self.query_filter();
1257        let entry = svc.describe_object(ref_, Some(&filter))?;
1258        let value = serde_json::to_value(entry).map_err(|e| ToolError::Execution(e.to_string()))?;
1259        Ok(ToolOutcome::ok_json(value))
1260    }
1261
1262    async fn get_join_path(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1263        let a = args
1264            .get("a")
1265            .and_then(|v| v.as_str())
1266            .ok_or_else(|| ToolError::InvalidArgs("a is required".into()))?;
1267        let b = args
1268            .get("b")
1269            .and_then(|v| v.as_str())
1270            .ok_or_else(|| ToolError::InvalidArgs("b is required".into()))?;
1271        let (store, connection_id, database) = self.index_service().await?;
1272        let svc = IndexQueryService::new(store, &connection_id, &database);
1273        let path = svc.get_join_path(a, b)?;
1274        let value = serde_json::to_value(path).map_err(|e| ToolError::Execution(e.to_string()))?;
1275        Ok(ToolOutcome::ok_json(value))
1276    }
1277
1278    async fn sample_values(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1279        let ref_ = args
1280            .get("ref")
1281            .and_then(|v| v.as_str())
1282            .ok_or_else(|| ToolError::InvalidArgs("ref is required".into()))?;
1283        let col = args
1284            .get("col")
1285            .and_then(|v| v.as_str())
1286            .ok_or_else(|| ToolError::InvalidArgs("col is required".into()))?;
1287        let (store, connection_id, database) = self.index_service().await?;
1288        let svc = IndexQueryService::new(store, &connection_id, &database);
1289        let filter = self.query_filter();
1290        let result = svc.sample_values(ref_, col, Some(&filter), None)?;
1291
1292        let mut values = result.values;
1293        let mut message = result.message;
1294
1295        if values.is_empty()
1296            && let Ok(client) = self.session.checkout().await
1297        {
1298            let parts: Vec<&str> = ref_.split('.').collect();
1299            let (schema, table) = match parts.as_slice() {
1300                [s, t] => (*s, *t),
1301                _ => ("public", ref_),
1302            };
1303            let safe_schema = schema.replace('"', "\"\"");
1304            let safe_table = table.replace('"', "\"\"");
1305            let safe_col = col.replace('"', "\"\"");
1306            let query = format!(
1307                "SELECT DISTINCT \"{safe_col}\"::text FROM \"{safe_schema}\".\"{safe_table}\" WHERE \"{safe_col}\" IS NOT NULL LIMIT 20"
1308            );
1309            if let Ok(rows) = client.query(&query, &[]).await {
1310                let sampled: Vec<String> = rows
1311                    .iter()
1312                    .filter_map(|r| r.get::<_, Option<String>>(0))
1313                    .collect();
1314                if !sampled.is_empty() {
1315                    values = sampled;
1316                    message = None;
1317                }
1318            }
1319        }
1320
1321        let mut payload = json!({ "values": values });
1322        if let Some(msg) = message {
1323            payload["message"] = json!(msg);
1324        }
1325        Ok(ToolOutcome::ok_json(payload))
1326    }
1327
1328    fn list_connections(&self) -> ToolOutcome {
1329        let rows: Vec<Value> = self
1330            .session
1331            .connections()
1332            .iter()
1333            .map(|c| {
1334                json!({
1335                    "id": c.id,
1336                    "name": c.name,
1337                    "host": c.host,
1338                    "port": c.port,
1339                    "database": c.database,
1340                })
1341            })
1342            .collect();
1343        ToolOutcome::ok_json(json!(rows))
1344    }
1345
1346    async fn list_databases(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1347        let connection_id = args
1348            .get("connectionId")
1349            .and_then(|v| v.as_str())
1350            .ok_or_else(|| ToolError::InvalidArgs("connectionId is required".into()))?;
1351        let conn = self
1352            .session
1353            .connections()
1354            .into_iter()
1355            .find(|c| c.id == connection_id)
1356            .ok_or_else(|| {
1357                ToolError::Execution(format!(
1358                    "Connection not found for ID: {connection_id} — call list_connections"
1359                ))
1360            })?;
1361        // Connect using that profile's params (may differ from active).
1362        let client = {
1363            // Temporarily use active checkout if same id; else one-shot.
1364            if self.session.active_context().await.0 == connection_id {
1365                self.session.checkout().await?
1366            } else {
1367                let pool_opts = self.session.pool_opts();
1368                let pool = nexql_conn::create_pool(&conn.params, &pool_opts).await?;
1369                nexql_conn::checkout_guarded(&pool, &pool_opts).await?
1370            }
1371        };
1372        let rows = client
1373            .query(
1374                "SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname",
1375                &[],
1376            )
1377            .await?;
1378        let names: Vec<String> = rows.iter().map(|r| r.get(0)).collect();
1379        Ok(ToolOutcome::ok_json(json!(names)))
1380    }
1381
1382    async fn list_schemas(&self) -> Result<ToolOutcome, ToolError> {
1383        let client = self.session.checkout().await?;
1384        let rows = client
1385            .query(
1386                r#"
1387                SELECT nspname AS schema_name
1388                FROM pg_namespace
1389                WHERE nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')
1390                  AND nspname NOT LIKE 'pg_%'
1391                ORDER BY nspname
1392                "#,
1393                &[],
1394            )
1395            .await?;
1396        let out: Vec<Value> = rows
1397            .iter()
1398            .filter(|r| {
1399                let name: String = r.get(0);
1400                self.session.filter().allows_schema(&name)
1401            })
1402            .map(|r| json!({ "schema_name": r.get::<_, String>(0) }))
1403            .collect();
1404        Ok(ToolOutcome::ok_json(json!(out)))
1405    }
1406
1407    async fn list_objects(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1408        let schema = args
1409            .get("schema")
1410            .and_then(|v| v.as_str())
1411            .unwrap_or("public");
1412        if !schema
1413            .chars()
1414            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
1415        {
1416            return Err(ToolError::InvalidArgs(
1417                "Invalid or missing schema name format".into(),
1418            ));
1419        }
1420        if !self.session.filter().allows_schema(schema) {
1421            return Ok(ToolOutcome::ok_json(json!([])));
1422        }
1423        let kind = args.get("kind").and_then(|v| v.as_str());
1424        let mut queries = Vec::new();
1425        let push_rel = |queries: &mut Vec<String>, relkinds: &[&str], label: &str| {
1426            let kinds = relkinds
1427                .iter()
1428                .map(|k| format!("'{k}'"))
1429                .collect::<Vec<_>>()
1430                .join(",");
1431            queries.push(format!(
1432                r#"
1433                SELECT n.nspname AS schema, c.relname AS name, '{label}' AS kind,
1434                       d.description AS comment
1435                FROM pg_class c
1436                JOIN pg_namespace n ON n.oid = c.relnamespace
1437                LEFT JOIN pg_description d ON d.objoid = c.oid AND d.objsubid = 0
1438                WHERE n.nspname = $1 AND c.relkind IN ({kinds})
1439                "#
1440            ));
1441        };
1442        if kind.is_none() || kind == Some("table") {
1443            push_rel(&mut queries, &["r", "f", "p"], "table");
1444        }
1445        if kind.is_none() || kind == Some("view") {
1446            push_rel(&mut queries, &["v"], "view");
1447        }
1448        if kind.is_none() || kind == Some("matview") {
1449            push_rel(&mut queries, &["m"], "matview");
1450        }
1451        if queries.is_empty() {
1452            return Ok(ToolOutcome::ok_json(json!([])));
1453        }
1454        let sql = queries.join("\nUNION ALL\n") + "\nORDER BY kind, name";
1455        let client = self.session.checkout().await?;
1456        let rows = client.query(&sql, &[&schema]).await?;
1457        let out: Vec<Value> = rows
1458            .iter()
1459            .filter(|r| {
1460                let s: String = r.get("schema");
1461                let name: String = r.get("name");
1462                self.session.filter().allows_table(&s, &name)
1463            })
1464            .map(|r| {
1465                json!({
1466                    "schema": r.get::<_, String>("schema"),
1467                    "name": r.get::<_, String>("name"),
1468                    "kind": r.get::<_, String>("kind"),
1469                    "comment": r.get::<_, Option<String>>("comment"),
1470                })
1471            })
1472            .collect();
1473        Ok(ToolOutcome::ok_json(json!(out)))
1474    }
1475
1476    async fn get_current_context(&self) -> Result<ToolOutcome, ToolError> {
1477        let (connection_id, database) = self.session.active_context().await;
1478        let conn = self
1479            .session
1480            .connections()
1481            .into_iter()
1482            .find(|c| c.id == connection_id);
1483        Ok(ToolOutcome::ok_json(json!({
1484            "connectionId": connection_id,
1485            "connectionName": conn.as_ref().map(|c| c.name.clone()).unwrap_or_else(|| "Unknown".into()),
1486            "database": database,
1487            "host": conn.as_ref().and_then(|c| c.host.clone()),
1488            "port": conn.as_ref().and_then(|c| c.port),
1489            "access_mode": match self.session.access_mode() {
1490                nexql_policy::AccessMode::Read => "read",
1491                nexql_policy::AccessMode::Write => "write",
1492                nexql_policy::AccessMode::Admin => "admin",
1493            },
1494        })))
1495    }
1496
1497    async fn switch_connection(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1498        let connection_id = args
1499            .get("connectionId")
1500            .and_then(|v| v.as_str())
1501            .ok_or_else(|| ToolError::InvalidArgs("connectionId is required".into()))?;
1502        let database = args
1503            .get("database")
1504            .and_then(|v| v.as_str())
1505            .map(str::to_owned);
1506        self.session.switch(connection_id, database).await?;
1507        self.get_current_context().await
1508    }
1509
1510    async fn run_select(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1511        let sql = args
1512            .get("sql")
1513            .and_then(|v| v.as_str())
1514            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
1515        match validate_readonly_sql(sql)? {
1516            SqlDecision::Allow => {}
1517            SqlDecision::Reject => {
1518                return Err(ToolError::Execution(
1519                    "Security Error: Only read-only SELECT, WITH, or EXPLAIN statements are permitted."
1520                        .into(),
1521                ));
1522            }
1523        }
1524        enforce_read_table_policy(&self.session.filter(), sql)?;
1525        let trimmed = sql.trim().to_ascii_lowercase();
1526        if trimmed.starts_with("explain") {
1527            return self.run_select_internal(sql, None).await;
1528        }
1529        let max_rows = self.session.caps().max_rows;
1530        self.run_select_internal(sql, Some(max_rows)).await
1531    }
1532
1533    async fn explain_query(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1534        let sql = args
1535            .get("sql")
1536            .and_then(|v| v.as_str())
1537            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
1538        match validate_readonly_sql(sql)? {
1539            SqlDecision::Allow => {}
1540            SqlDecision::Reject => {
1541                return Err(ToolError::Execution(
1542                    "Security Error: Only SELECT, WITH, or EXPLAIN statements can be analyzed."
1543                        .into(),
1544                ));
1545            }
1546        }
1547        enforce_read_table_policy(&self.session.filter(), sql)?;
1548        let clean = if sql.trim().to_ascii_lowercase().starts_with("explain") {
1549            sql.to_string()
1550        } else {
1551            format!("EXPLAIN {sql}")
1552        };
1553        // Re-validate EXPLAIN wrapper
1554        if validate_readonly_sql(&clean)? == SqlDecision::Reject {
1555            return Err(ToolError::Execution(
1556                "Security Error: EXPLAIN target is not read-only.".into(),
1557            ));
1558        }
1559        self.run_select_internal(&clean, None).await
1560    }
1561
1562    async fn get_ddl(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1563        let ref_ = args
1564            .get("ref")
1565            .and_then(|v| v.as_str())
1566            .ok_or_else(|| ToolError::InvalidArgs("ref is required".into()))?;
1567        let (schema, name) = parse_ref(ref_).map_err(ToolError::InvalidArgs)?;
1568        let kind = args.get("kind").and_then(|v| v.as_str()).unwrap_or("table");
1569        let reg = sql::regclass_literal(&schema, &name);
1570        let client = self.session.checkout().await?;
1571
1572        match kind {
1573            "view" | "matview" => {
1574                let sql = format!("SELECT pg_get_viewdef({reg}, true) AS definition");
1575                let rows = client.query(&sql, &[]).await?;
1576                Ok(ToolOutcome::ok_json(rows_to_json(&rows)))
1577            }
1578            "function" => {
1579                let sql = format!(
1580                    r#"SELECT p.proname AS name, pg_get_functiondef(p.oid) AS definition
1581                       FROM pg_proc p
1582                       JOIN pg_namespace n ON n.oid = p.pronamespace
1583                       WHERE n.nspname = '{schema}' AND p.proname = '{name}'"#
1584                );
1585                let rows = client.query(&sql, &[]).await?;
1586                Ok(ToolOutcome::ok_json(rows_to_json(&rows)))
1587            }
1588            "index" => {
1589                let sql = format!("SELECT pg_get_indexdef({reg}) AS definition");
1590                let rows = client.query(&sql, &[]).await?;
1591                Ok(ToolOutcome::ok_json(rows_to_json(&rows)))
1592            }
1593            "table" => {
1594                let columns = client
1595                    .query(&sql::column_details(&schema, &name), &[])
1596                    .await?;
1597                let constraints = client
1598                    .query(
1599                        &format!(
1600                            r#"SELECT conname AS name, pg_get_constraintdef(oid) AS definition
1601                               FROM pg_constraint WHERE conrelid = {reg} ORDER BY conname"#
1602                        ),
1603                        &[],
1604                    )
1605                    .await?;
1606                let indexes = client
1607                    .query(
1608                        &format!(
1609                            r#"SELECT indexname AS name, indexdef AS definition
1610                               FROM pg_indexes
1611                               WHERE schemaname = '{schema}' AND tablename = '{name}'
1612                               ORDER BY indexname"#
1613                        ),
1614                        &[],
1615                    )
1616                    .await?;
1617                Ok(ToolOutcome::ok_json(json!({
1618                    "table": format!("{schema}.{name}"),
1619                    "columns": rows_to_json(&columns),
1620                    "constraints": rows_to_json(&constraints),
1621                    "indexes": rows_to_json(&indexes),
1622                })))
1623            }
1624            other => Err(ToolError::InvalidArgs(format!(
1625                "Unsupported DDL kind \"{other}\". Use table, view, matview, function, or index."
1626            ))),
1627        }
1628    }
1629
1630    async fn table_stats(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1631        let ref_ = args
1632            .get("ref")
1633            .and_then(|v| v.as_str())
1634            .ok_or_else(|| ToolError::InvalidArgs("ref is required".into()))?;
1635        let (schema, name) = parse_ref(ref_).map_err(ToolError::InvalidArgs)?;
1636        let client = self.session.checkout().await?;
1637        let stats = client.query(&sql::table_stats(&schema, &name), &[]).await?;
1638        let activity = client
1639            .query(&sql::table_activity(&schema, &name), &[])
1640            .await?;
1641        let columns = client
1642            .query(&sql::column_stats(&schema, &name), &[])
1643            .await?;
1644        let size = rows_to_json(&stats)
1645            .as_array()
1646            .and_then(|a| a.first())
1647            .cloned()
1648            .unwrap_or(Value::Null);
1649        let activity = rows_to_json(&activity)
1650            .as_array()
1651            .and_then(|a| a.first())
1652            .cloned()
1653            .unwrap_or(Value::Null);
1654        Ok(ToolOutcome::ok_json(json!({
1655            "size": size,
1656            "activity": activity,
1657            "columns": rows_to_json(&columns),
1658        })))
1659    }
1660
1661    async fn index_usage(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1662        let ref_ = args
1663            .get("ref")
1664            .and_then(|v| v.as_str())
1665            .ok_or_else(|| ToolError::InvalidArgs("ref is required".into()))?;
1666        let (schema, name) = parse_ref(ref_).map_err(ToolError::InvalidArgs)?;
1667        let client = self.session.checkout().await?;
1668        let rows = client.query(&sql::index_usage(&schema, &name), &[]).await?;
1669        Ok(ToolOutcome::ok_json(rows_to_json(&rows)))
1670    }
1671
1672    async fn list_running_queries(&self) -> Result<ToolOutcome, ToolError> {
1673        let client = self.session.checkout().await?;
1674        let rows = client.query(sql::running_queries(), &[]).await?;
1675        Ok(ToolOutcome::ok_json(rows_to_json(&rows)))
1676    }
1677
1678    async fn find_blocking_locks(&self) -> Result<ToolOutcome, ToolError> {
1679        let client = self.session.checkout().await?;
1680        let rows = client.query(sql::blocking_locks(), &[]).await?;
1681        let values = rows_to_json(&rows);
1682        if values.as_array().map(|a| a.is_empty()).unwrap_or(true) {
1683            return Ok(ToolOutcome::ok_json(json!({
1684                "message": "No blocking locks found.",
1685                "locks": [],
1686            })));
1687        }
1688        Ok(ToolOutcome::ok_json(values))
1689    }
1690
1691    async fn slow_queries(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1692        let limit = args
1693            .get("limit")
1694            .and_then(|v| v.as_u64())
1695            .map(|n| n as u32)
1696            .unwrap_or(SLOW_QUERIES_DEFAULT);
1697        let client = self.session.checkout().await?;
1698        match client.query(&sql::slow_queries(limit), &[]).await {
1699            Ok(rows) => Ok(ToolOutcome::ok_json(rows_to_json(&rows))),
1700            Err(e) => {
1701                if let Some(message) = sql::map_stat_statements_error(&e) {
1702                    Ok(ToolOutcome::ok_json(json!({
1703                        "error": message,
1704                        "hint": message,
1705                    })))
1706                } else {
1707                    Err(ToolError::Postgres(e))
1708                }
1709            }
1710        }
1711    }
1712
1713    async fn db_health_check(&self) -> Result<ToolOutcome, ToolError> {
1714        let client = self.session.checkout().await?;
1715        let sections: &[(&str, &str)] = &[
1716            ("overview", sql::database_stats()),
1717            ("cache", sql::cache_hit_ratio()),
1718            ("dead_tuples", sql::database_maintenance_stats()),
1719            ("connection_states", sql::connection_states()),
1720            ("blocking_locks", sql::blocking_locks()),
1721        ];
1722        let mut report = serde_json::Map::new();
1723        for (key, q) in sections {
1724            match client.query(*q, &[]).await {
1725                Ok(rows) => {
1726                    report.insert((*key).into(), rows_to_json(&rows));
1727                }
1728                Err(e) => {
1729                    report.insert((*key).into(), json!({ "error": e.to_string() }));
1730                }
1731            }
1732        }
1733        let lock_count = report
1734            .get("blocking_locks")
1735            .and_then(|v| v.as_array())
1736            .map(|a| a.len() as u64);
1737        report.insert("blocking_lock_count".into(), json!(lock_count));
1738        Ok(ToolOutcome::ok_json(Value::Object(report)))
1739    }
1740
1741    async fn explain_analyze(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1742        let sql = args
1743            .get("sql")
1744            .and_then(|v| v.as_str())
1745            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
1746        require_select_or_with(&self.session.filter(), sql)?;
1747        let explain = build_explain_sql(sql, true);
1748        self.run_explain_in_transaction(&explain).await
1749    }
1750
1751    async fn analyze_query_plan(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1752        let sql = args
1753            .get("sql")
1754            .and_then(|v| v.as_str())
1755            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
1756        require_select_or_with(&self.session.filter(), sql)?;
1757        let analyze = args
1758            .get("analyze")
1759            .and_then(|v| v.as_bool())
1760            .unwrap_or(false);
1761        let explain = build_explain_sql(sql, analyze);
1762        let outcome = self.run_explain_in_transaction(&explain).await?;
1763        let rows = outcome.structured.unwrap_or(Value::Null);
1764        let row_array = rows
1765            .get("rows")
1766            .and_then(|v| v.as_array())
1767            .or_else(|| rows.as_array());
1768        let plan = row_array
1769            .and_then(|a| a.first())
1770            .and_then(|r| r.get("QUERY PLAN"))
1771            .cloned()
1772            .unwrap_or(Value::Null);
1773        let metrics = extract_plan_metrics(&plan).or_else(|| extract_plan_metrics(&rows));
1774        let recommendations = metrics
1775            .as_ref()
1776            .and_then(|m| m.get("recommendations"))
1777            .cloned()
1778            .unwrap_or_else(|| json!([]));
1779        Ok(ToolOutcome::ok_json(json!({
1780            "metrics": metrics,
1781            "recommendations": recommendations,
1782            "plan": plan,
1783        })))
1784    }
1785
1786    /// EXPLAIN ANALYZE executes the query — always wrap in READ ONLY + ROLLBACK.
1787    async fn run_explain_in_transaction(
1788        &self,
1789        explain_sql: &str,
1790    ) -> Result<ToolOutcome, ToolError> {
1791        let client = self.session.checkout().await?;
1792        client
1793            .batch_execute("SET statement_timeout = '30s'")
1794            .await?;
1795        client.batch_execute("BEGIN").await?;
1796        let result = async {
1797            client.batch_execute("SET TRANSACTION READ ONLY").await?;
1798            let rows = client.query(explain_sql, &[]).await?;
1799            Ok::<_, ToolError>(rows_to_json(&rows))
1800        }
1801        .await;
1802        // Always roll back — belt-and-braces on top of default_transaction_read_only.
1803        let _ = client.batch_execute("ROLLBACK").await;
1804        match result {
1805            Ok(values) => Ok(ToolOutcome::ok_json(values)),
1806            Err(e) => Err(e),
1807        }
1808    }
1809
1810    async fn get_index_status(&self) -> Result<ToolOutcome, ToolError> {
1811        let (store, connection_id, database) = self.index_service().await?;
1812        let base = store.base_dir(&connection_id, &database);
1813        let Some(manifest) = store.read_manifest(&base)? else {
1814            return Err(ToolError::Execution(format!(
1815                "No schema index for database \"{database}\" — run `nexql-mcp index build`."
1816            )));
1817        };
1818
1819        let mut live_fingerprint: Option<String> = None;
1820        let mut drift: Option<bool> = None;
1821        if let Ok(client) = self.session.checkout().await {
1822            let db = PgCatalogDb::new(&client);
1823            if let Ok(fp) = db.schema_fingerprint().await {
1824                drift = Some(fp != manifest.schema_fingerprint);
1825                live_fingerprint = Some(fp);
1826            }
1827        }
1828
1829        Ok(ToolOutcome::ok_json(json!({
1830            "connectionId": manifest.connection_id,
1831            "database": manifest.database,
1832            "indexedAt": manifest.indexed_at,
1833            "fingerprint": manifest.schema_fingerprint,
1834            "liveFingerprint": live_fingerprint,
1835            "drift": drift,
1836            "pgVersion": manifest.pg_version,
1837            "counts": {
1838                "tables": manifest.counts.tables,
1839                "views": manifest.counts.views,
1840                "functions": manifest.counts.functions,
1841                "enums": manifest.counts.enums,
1842            },
1843            "buildMs": manifest.stats.build_ms,
1844            "warnings": manifest.stats.warnings,
1845        })))
1846    }
1847
1848    async fn list_extensions(&self) -> Result<ToolOutcome, ToolError> {
1849        let client = self.session.checkout().await?;
1850        let rows = client.query(sql::list_extensions(), &[]).await?;
1851        Ok(ToolOutcome::ok_json(rows_to_json(&rows)))
1852    }
1853
1854    async fn server_settings(&self) -> Result<ToolOutcome, ToolError> {
1855        let client = self.session.checkout().await?;
1856        let rows = client.query(sql::server_settings(), &[]).await?;
1857        Ok(ToolOutcome::ok_json(rows_to_json(&rows)))
1858    }
1859
1860    async fn suggest_indexes(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1861        let limit = args
1862            .get("limit")
1863            .and_then(|v| v.as_u64())
1864            .map(|n| n as u32)
1865            .unwrap_or(REPORT_LIMIT_DEFAULT);
1866        let client = self.session.checkout().await?;
1867        let mut query_errors = serde_json::Map::new();
1868
1869        let high_seq_json = match client
1870            .query(&sql::high_seq_scan_tables(limit), &[])
1871            .await
1872        {
1873            Ok(rows) => rows_to_json(&rows),
1874            Err(e) => {
1875                query_errors.insert(
1876                    "high_seq_scan_tables".into(),
1877                    json!(nexql_conn::format_postgres_error(&e)),
1878                );
1879                Value::Null
1880            }
1881        };
1882
1883        let unindexed_json = match client
1884            .query(&sql::unindexed_fk_columns(limit), &[])
1885            .await
1886        {
1887            Ok(rows) => rows_to_json(&rows),
1888            Err(e) => {
1889                query_errors.insert(
1890                    "unindexed_fk_columns".into(),
1891                    json!(nexql_conn::format_postgres_error(&e)),
1892                );
1893                Value::Null
1894            }
1895        };
1896
1897        let mut pg_stat_available = false;
1898        let mut slow_queries = Value::Null;
1899        let mut pg_stat_note: Option<String> = None;
1900        match client.query(&sql::slow_queries(limit.min(10)), &[]).await {
1901            Ok(rows) => {
1902                pg_stat_available = true;
1903                slow_queries = rows_to_json(&rows);
1904            }
1905            Err(e) => {
1906                if let Some(message) = sql::map_stat_statements_error(&e) {
1907                    pg_stat_note = Some(message);
1908                } else {
1909                    query_errors.insert(
1910                        "slow_queries".into(),
1911                        json!(nexql_conn::format_postgres_error(&e)),
1912                    );
1913                }
1914            }
1915        }
1916
1917        let mut plan_heuristics = Value::Null;
1918        if let Some(sql_text) = args.get("sql").and_then(|v| v.as_str()) {
1919            require_select_or_with(&self.session.filter(), sql_text)?;
1920            let explain = build_explain_sql(sql_text, false);
1921            match self.run_explain_in_transaction(&explain).await {
1922                Ok(outcome) => {
1923                    let rows = outcome.structured.unwrap_or(Value::Null);
1924                    let plan = rows
1925                        .as_array()
1926                        .and_then(|a| a.first())
1927                        .and_then(|r| r.get("QUERY PLAN"))
1928                        .cloned()
1929                        .unwrap_or(Value::Null);
1930                    let metrics =
1931                        extract_plan_metrics(&plan).or_else(|| extract_plan_metrics(&rows));
1932                    plan_heuristics = json!({
1933                        "metrics": metrics,
1934                        "hint": "Use analyze_query_plan with analyze=true for actual timings before creating indexes.",
1935                    });
1936                }
1937                Err(e) => {
1938                    query_errors.insert("plan_heuristics".into(), json!(e.to_string()));
1939                }
1940            }
1941        }
1942
1943        let has_candidates = high_seq_json
1944            .as_array()
1945            .map(|a| !a.is_empty())
1946            .unwrap_or(false)
1947            || unindexed_json
1948                .as_array()
1949                .map(|a| !a.is_empty())
1950                .unwrap_or(false)
1951            || plan_heuristics != Value::Null;
1952
1953        let mut payload = if !has_candidates && !pg_stat_available {
1954            json!({
1955                "suggestions": [],
1956                "message": "No index suggestions yet. Either table stats show healthy index use, or there is not enough scan history. Enable pg_stat_statements and/or pass a sql argument for EXPLAIN plan heuristics.",
1957                "hint": pg_stat_note,
1958            })
1959        } else if !has_candidates {
1960            json!({
1961                "high_seq_scan_tables": high_seq_json,
1962                "unindexed_fk_columns": unindexed_json,
1963                "slow_queries": slow_queries,
1964                "plan_heuristics": plan_heuristics,
1965                "message": "No strong index candidates from sequential-scan or unindexed-FK heuristics. Review slow_queries / pass sql for plan-level advice.",
1966                "hint": "CREATE INDEX CONCURRENTLY after validating with EXPLAIN (ANALYZE, BUFFERS).",
1967            })
1968        } else {
1969            json!({
1970                "high_seq_scan_tables": high_seq_json,
1971                "unindexed_fk_columns": unindexed_json,
1972                "slow_queries": slow_queries,
1973                "plan_heuristics": plan_heuristics,
1974                "pg_stat_statements": pg_stat_available,
1975                "hint": pg_stat_note.unwrap_or_else(|| {
1976                    "Validate candidates with analyze_query_plan / EXPLAIN before CREATE INDEX CONCURRENTLY.".into()
1977                }),
1978            })
1979        };
1980
1981        if !query_errors.is_empty() {
1982            payload["query_errors"] = Value::Object(query_errors);
1983        }
1984
1985        Ok(ToolOutcome::ok_json(payload))
1986    }
1987
1988    async fn find_unused_indexes(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
1989        let limit = args
1990            .get("limit")
1991            .and_then(|v| v.as_u64())
1992            .map(|n| n as u32)
1993            .unwrap_or(REPORT_LIMIT_DEFAULT);
1994        let client = self.session.checkout().await?;
1995        let rows = client.query(&sql::find_unused_indexes(limit), &[]).await?;
1996        let indexes = rows_to_json(&rows);
1997        if indexes.as_array().map(|a| a.is_empty()).unwrap_or(true) {
1998            return Ok(ToolOutcome::ok_json(json!({
1999                "indexes": [],
2000                "message": "No unused non-constraint indexes found (idx_scan = 0). Note: pg_stat_reset / server restart clears scan counts — treat never-scanned indexes cautiously on fresh stats.",
2001            })));
2002        }
2003        Ok(ToolOutcome::ok_json(json!({
2004            "indexes": indexes,
2005            "hint": "Prefer DROP INDEX CONCURRENTLY after confirming the workload (and that stats are mature).",
2006        })))
2007    }
2008
2009    async fn bloat_report(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
2010        let limit = args
2011            .get("limit")
2012            .and_then(|v| v.as_u64())
2013            .map(|n| n as u32)
2014            .unwrap_or(REPORT_LIMIT_DEFAULT);
2015        let client = self.session.checkout().await?;
2016        let rows = client.query(&sql::bloat_report(limit), &[]).await?;
2017        let tables = rows_to_json(&rows);
2018        if tables.as_array().map(|a| a.is_empty()).unwrap_or(true) {
2019            return Ok(ToolOutcome::ok_json(json!({
2020                "tables": [],
2021                "method": "dead_tuple_ratio",
2022                "message": "No tables with significant dead-tuple pressure (>1000 dead tuples). This is a simplified estimate from pg_stat_user_tables, not physical page bloat.",
2023            })));
2024        }
2025        Ok(ToolOutcome::ok_json(json!({
2026            "tables": tables,
2027            "method": "dead_tuple_ratio",
2028            "note": "Approximate bloat via n_dead_tup / (n_live_tup + n_dead_tup). Not a physical page-bloat estimate (pgstattuple / check_postgres). Consider VACUUM / VACUUM FULL only after confirming impact.",
2029            "hint": "VACUUM ANALYZE on high bloat_pct tables; investigate autovacuum settings if last_autovacuum is stale.",
2030        })))
2031    }
2032
2033    async fn find_missing_fks(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
2034        let limit = args
2035            .get("limit")
2036            .and_then(|v| v.as_u64())
2037            .map(|n| n as u32)
2038            .unwrap_or(REPORT_LIMIT_DEFAULT);
2039        let capped = limit.clamp(1, sql::REPORT_LIMIT_MAX) as usize;
2040
2041        // Prefer schema-index join-graph inferred edges when an index exists.
2042        if let Ok((store, connection_id, database)) = self.index_service().await {
2043            let base = store.base_dir(&connection_id, &database);
2044            if let Ok(Some(manifest)) = store.read_manifest(&base)
2045                && let Ok(Some(graph)) = store.read_join_graph(&base, &manifest)
2046            {
2047                let candidates: Vec<Value> = graph
2048                    .edges
2049                    .into_iter()
2050                    .filter(|e| e.inferred == Some(true) && e.disabled != Some(true))
2051                    .take(capped)
2052                    .map(|e| {
2053                        let cols: Vec<Value> = e
2054                            .cols
2055                            .iter()
2056                            .map(|(a, b)| json!({ "from": a, "to": b }))
2057                            .collect();
2058                        json!({
2059                            "from_table": e.from,
2060                            "to_table": e.to,
2061                            "via": e.via,
2062                            "columns": cols,
2063                            "detection": "join_graph_inferred",
2064                        })
2065                    })
2066                    .collect();
2067                if !candidates.is_empty() {
2068                    return Ok(ToolOutcome::ok_json(json!({
2069                        "candidates": candidates,
2070                        "source": "join_graph",
2071                        "hint": "These edges were inferred by naming convention and have no declared FK. Review before ALTER TABLE … ADD FOREIGN KEY.",
2072                    })));
2073                }
2074            }
2075        }
2076
2077        let client = self.session.checkout().await?;
2078        let rows = client
2079            .query(&sql::find_missing_fks_catalog(limit), &[])
2080            .await?;
2081        let candidates = rows_to_json(&rows);
2082        if candidates.as_array().map(|a| a.is_empty()).unwrap_or(true) {
2083            return Ok(ToolOutcome::ok_json(json!({
2084                "candidates": [],
2085                "source": "catalog",
2086                "message": "No missing FK candidates found via join-graph inferred edges or *_id naming against single-column PKs.",
2087            })));
2088        }
2089        Ok(ToolOutcome::ok_json(json!({
2090            "candidates": candidates,
2091            "source": "catalog",
2092            "hint": "Naming-inferred only — verify referential integrity and nullability before adding constraints. Run `nexql-mcp index build` for join-graph inferred edges.",
2093        })))
2094    }
2095
2096    async fn list_roles(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
2097        let client = self.session.checkout().await?;
2098        let role = args
2099            .get("role")
2100            .and_then(|v| v.as_str())
2101            .map(str::trim)
2102            .filter(|s| !s.is_empty());
2103
2104        let Some(role_name) = role else {
2105            let rows = client.query(sql::list_roles(), &[]).await?;
2106            return Ok(ToolOutcome::ok_json(rows_to_json(&rows)));
2107        };
2108
2109        let details = client.query(sql::role_details(), &[&role_name]).await?;
2110        if details.is_empty() {
2111            return Err(ToolError::Execution(format!(
2112                "Role \"{role_name}\" not found"
2113            )));
2114        }
2115        let member_of = client.query(sql::role_member_of(), &[&role_name]).await?;
2116        let has_members = client.query(sql::role_has_members(), &[&role_name]).await?;
2117        let privileges = client
2118            .query(sql::role_table_privileges(), &[&role_name])
2119            .await?;
2120
2121        Ok(ToolOutcome::ok_json(json!({
2122            "role": rows_to_json(&details).as_array().and_then(|a| a.first().cloned()).unwrap_or(Value::Null),
2123            "member_of": rows_to_json(&member_of),
2124            "has_members": rows_to_json(&has_members),
2125            "table_privileges": rows_to_json(&privileges),
2126        })))
2127    }
2128
2129    async fn export_query(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
2130        let sql = args
2131            .get("sql")
2132            .and_then(|v| v.as_str())
2133            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
2134        require_select_or_with(&self.session.filter(), sql)?;
2135
2136        let format = args
2137            .get("format")
2138            .and_then(|v| v.as_str())
2139            .map(|s| {
2140                ExportFormat::parse(s).ok_or_else(|| {
2141                    ToolError::InvalidArgs(format!(
2142                        "Unsupported format \"{s}\". Use csv, json, or sqlinsert."
2143                    ))
2144                })
2145            })
2146            .transpose()?
2147            .unwrap_or(ExportFormat::Csv);
2148
2149        let table_target = match args.get("table").and_then(|v| v.as_str()) {
2150            Some(t) if !t.trim().is_empty() => Some(parse_ref(t).map_err(ToolError::InvalidArgs)?),
2151            _ => None,
2152        };
2153
2154        if format == ExportFormat::SqlInsert && table_target.is_none() {
2155            return Err(ToolError::InvalidArgs(
2156                "table (schema.name) is required when format=sqlinsert".into(),
2157            ));
2158        }
2159
2160        let max_rows = self.session.caps().max_rows;
2161        let outcome = self.run_select_internal(sql, Some(max_rows)).await?;
2162        if outcome.is_error {
2163            return Ok(outcome);
2164        }
2165
2166        let structured = outcome.structured.unwrap_or(Value::Null);
2167        let rows_val = structured
2168            .get("rows")
2169            .cloned()
2170            .or_else(|| structured.get("data").and_then(|d| d.get("rows").cloned()))
2171            .unwrap_or(Value::Array(vec![]));
2172        let rows = rows_val.as_array().cloned().unwrap_or_default();
2173        let columns = columns_from_rows(&rows);
2174        let truncated = structured
2175            .get("truncated")
2176            .and_then(|v| v.as_bool())
2177            .unwrap_or(false);
2178
2179        let payload = match format {
2180            ExportFormat::Json => json!({
2181                "format": format.as_str(),
2182                "rowCount": rows.len(),
2183                "truncated": truncated,
2184                "columns": columns,
2185                "rows": rows,
2186            }),
2187            ExportFormat::Csv => {
2188                let content = rows_to_csv(&rows, &columns);
2189                let caps = self.session.caps();
2190                let (char_trunc, content) = caps.truncate_chars(&content);
2191                json!({
2192                    "format": format.as_str(),
2193                    "rowCount": rows.len(),
2194                    "truncated": truncated || char_trunc,
2195                    "columns": columns,
2196                    "content": content,
2197                })
2198            }
2199            ExportFormat::SqlInsert => {
2200                let (schema, table) = table_target.expect("checked above");
2201                let content = rows_to_sql_insert(&rows, &columns, &schema, &table);
2202                let caps = self.session.caps();
2203                let (char_trunc, content) = caps.truncate_chars(&content);
2204                json!({
2205                    "format": format.as_str(),
2206                    "rowCount": rows.len(),
2207                    "truncated": truncated || char_trunc,
2208                    "table": format!("{schema}.{table}"),
2209                    "columns": columns,
2210                    "content": content,
2211                })
2212            }
2213        };
2214
2215        Ok(ToolOutcome::ok_json(payload))
2216    }
2217
2218    async fn db_dashboard(&self) -> Result<ToolOutcome, ToolError> {
2219        let client = self.session.checkout().await?;
2220        let sections: &[(&str, &str)] = &[
2221            ("db_info", sql::dashboard_db_info()),
2222            ("connection_states", sql::connection_states()),
2223            ("top_tables", sql::dashboard_top_tables()),
2224            ("object_counts", sql::dashboard_object_counts()),
2225            ("active_queries", sql::dashboard_active_queries()),
2226            ("blocking_locks", sql::blocking_locks()),
2227            ("max_connections", sql::dashboard_max_connections()),
2228            ("extension_count", sql::dashboard_extension_count()),
2229            ("cache", sql::cache_hit_ratio()),
2230        ];
2231        let mut report = serde_json::Map::new();
2232        for (key, q) in sections {
2233            match client.query(*q, &[]).await {
2234                Ok(rows) => {
2235                    report.insert((*key).into(), rows_to_json(&rows));
2236                }
2237                Err(e) => {
2238                    report.insert((*key).into(), json!({ "error": e.to_string() }));
2239                }
2240            }
2241        }
2242
2243        // Normalize single-row sections to objects for agents.
2244        for key in ["db_info", "object_counts", "extension_count", "cache"] {
2245            if let Some(Value::Array(arr)) = report.get(key).cloned()
2246                && arr.len() == 1
2247            {
2248                report.insert(key.into(), arr.into_iter().next().unwrap());
2249            }
2250        }
2251        if let Some(Value::Array(arr)) = report.get("max_connections").cloned()
2252            && let Some(row) = arr.first()
2253        {
2254            report.insert(
2255                "max_connections".into(),
2256                row.get("max_connections")
2257                    .cloned()
2258                    .unwrap_or_else(|| row.clone()),
2259            );
2260        }
2261
2262        Ok(ToolOutcome::ok_json(Value::Object(report)))
2263    }
2264
2265    async fn deep_plan_analysis(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
2266        let sql = args
2267            .get("sql")
2268            .and_then(|v| v.as_str())
2269            .ok_or_else(|| ToolError::InvalidArgs("sql is required".into()))?;
2270        require_select_or_with(&self.session.filter(), sql)?;
2271        let analyze = args
2272            .get("analyze")
2273            .and_then(|v| v.as_bool())
2274            .unwrap_or(true);
2275        let explain = build_explain_sql(sql, analyze);
2276        let outcome = self.run_explain_in_transaction(&explain).await?;
2277        let rows = outcome.structured.unwrap_or(Value::Null);
2278        let row_array = rows
2279            .get("rows")
2280            .and_then(|v| v.as_array())
2281            .or_else(|| rows.as_array());
2282        let plan = row_array
2283            .and_then(|a| a.first())
2284            .and_then(|r| r.get("QUERY PLAN"))
2285            .cloned()
2286            .unwrap_or(Value::Null);
2287        let deep = analyze_deep_plan(&plan, sql)
2288            .or_else(|| analyze_deep_plan(&rows, sql))
2289            .ok_or_else(|| {
2290                ToolError::Execution("Could not parse EXPLAIN JSON plan for deep analysis".into())
2291            })?;
2292        let metrics = extract_plan_metrics(&plan).or_else(|| extract_plan_metrics(&rows));
2293        Ok(ToolOutcome::ok_json(json!({
2294            "deep": deep,
2295            "metrics": metrics,
2296            "plan": plan,
2297            "analyzed": analyze,
2298        })))
2299    }
2300
2301    async fn schema_diff(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
2302        let source_schema = args
2303            .get("sourceSchema")
2304            .and_then(|v| v.as_str())
2305            .ok_or_else(|| ToolError::InvalidArgs("sourceSchema is required".into()))?;
2306        let target_schema = args
2307            .get("targetSchema")
2308            .and_then(|v| v.as_str())
2309            .ok_or_else(|| ToolError::InvalidArgs("targetSchema is required".into()))?;
2310        crate::schema_diff::require_safe_schema(source_schema)?;
2311        crate::schema_diff::require_safe_schema(target_schema)?;
2312
2313        let client = self.session.checkout().await?;
2314        let source = crate::schema_diff::load_schema_snapshot(&client, source_schema).await?;
2315        let target = crate::schema_diff::load_schema_snapshot(&client, target_schema).await?;
2316        let diffs = crate::schema_diff::compute_schema_diff(&source, &target);
2317        let changed = diffs
2318            .iter()
2319            .filter(|d| d.status != crate::schema_diff::DiffStatus::Unchanged)
2320            .count();
2321        Ok(ToolOutcome::ok_json(json!({
2322            "sourceSchema": source_schema,
2323            "targetSchema": target_schema,
2324            "tableCount": diffs.len(),
2325            "changedCount": changed,
2326            "diffs": crate::schema_diff::diffs_to_json(&diffs),
2327        })))
2328    }
2329
2330    async fn generate_migration(&self, args: &Value) -> Result<ToolOutcome, ToolError> {
2331        let source_schema = args
2332            .get("sourceSchema")
2333            .and_then(|v| v.as_str())
2334            .ok_or_else(|| ToolError::InvalidArgs("sourceSchema is required".into()))?;
2335        let target_schema = args
2336            .get("targetSchema")
2337            .and_then(|v| v.as_str())
2338            .ok_or_else(|| ToolError::InvalidArgs("targetSchema is required".into()))?;
2339        crate::schema_diff::require_safe_schema(source_schema)?;
2340        crate::schema_diff::require_safe_schema(target_schema)?;
2341
2342        let client = self.session.checkout().await?;
2343        let source = crate::schema_diff::load_schema_snapshot(&client, source_schema).await?;
2344        let target = crate::schema_diff::load_schema_snapshot(&client, target_schema).await?;
2345        let diffs = crate::schema_diff::compute_schema_diff(&source, &target);
2346        let statements =
2347            crate::schema_diff::build_migration_statements(source_schema, target_schema, &diffs);
2348        let sql = if statements.is_empty() {
2349            format!("-- No differences between {source_schema} and {target_schema}")
2350        } else {
2351            statements.join("\n\n")
2352        };
2353        Ok(ToolOutcome::ok_json(json!({
2354            "sourceSchema": source_schema,
2355            "targetSchema": target_schema,
2356            "statementCount": statements.len(),
2357            "sql": sql,
2358            "hint": "Read-only: review and run via execute_sql / apply_ddl only with --access-mode write|admin. Destructive drops are commented out.",
2359        })))
2360    }
2361
2362    async fn run_select_internal(
2363        &self,
2364        sql: &str,
2365        max_rows: Option<u32>,
2366    ) -> Result<ToolOutcome, ToolError> {
2367        let client = self.session.checkout().await?;
2368        let Some(max_rows) = max_rows else {
2369            let rows = client.query(sql, &[]).await?;
2370            let values = rows_to_json(&rows);
2371            let payload = self.apply_pii_redaction(sql, ensure_structured_object(values));
2372            let text = serde_json::to_string_pretty(&payload)
2373                .map_err(|e| ToolError::Execution(e.to_string()))?;
2374            let caps = self.session.caps();
2375            let (trunc, text) = caps.truncate_chars(&text);
2376            let structured = if trunc {
2377                json!({ "truncated_chars": true, "data": payload })
2378            } else {
2379                payload
2380            };
2381            return Ok(ToolOutcome {
2382                text: text.to_string(),
2383                structured: Some(structured),
2384                is_error: false,
2385            });
2386        };
2387
2388        let cleaned = sql.trim().trim_end_matches(';').trim();
2389        let wrapped = format!(
2390            "SELECT * FROM ({cleaned}) AS nexql_limited LIMIT {}",
2391            max_rows + 1
2392        );
2393        let rows = client.query(&wrapped, &[]).await.map_err(|e| {
2394            ToolError::Execution(format!(
2395                "Failed to execute row-limited query (refusing unbounded fallback): {}",
2396                nexql_conn::format_postgres_error(&e)
2397            ))
2398        })?;
2399        let truncated = rows.len() as u32 > max_rows;
2400        let keep = if truncated {
2401            &rows[..max_rows as usize]
2402        } else {
2403            &rows[..]
2404        };
2405        let values = rows_to_json(keep);
2406        // Always `{ "rows": [...] }` — truncation flags are extra fields on the object.
2407        let mut payload = self.apply_pii_redaction(sql, ensure_structured_object(values));
2408        if truncated
2409            && let Some(obj) = payload.as_object_mut()
2410        {
2411            obj.insert("truncated".into(), json!(true));
2412            obj.insert("maxRows".into(), json!(max_rows));
2413        }
2414        let text = serde_json::to_string_pretty(&payload)
2415            .map_err(|e| ToolError::Execution(e.to_string()))?;
2416        let caps = self.session.caps();
2417        let (char_trunc, text) = caps.truncate_chars(&text);
2418        let structured = if char_trunc {
2419            json!({ "truncated_chars": true, "data": payload })
2420        } else {
2421            payload
2422        };
2423        Ok(ToolOutcome {
2424            text: text.to_string(),
2425            structured: Some(structured),
2426            is_error: false,
2427        })
2428    }
2429
2430    fn apply_pii_redaction(&self, sql: &str, mut payload: Value) -> Value {
2431        let filter = self.session.filter();
2432        if filter.pii_columns.is_empty() {
2433            return payload;
2434        }
2435        let Ok(tables) = select_table_refs(sql) else {
2436            return payload;
2437        };
2438        let (redacted, cols) = redact_pii_in_payload(payload, &filter.pii_columns, &tables);
2439        payload = redacted;
2440        if !cols.is_empty()
2441            && let Some(obj) = payload.as_object_mut()
2442        {
2443            obj.insert("piiRedactedColumns".into(), json!(cols));
2444        }
2445        payload
2446    }
2447}
2448
2449/// Lowercase + collapse to alphanumeric-separated-by-single-spaces, for `fuzzy_score`.
2450fn normalize_for_match(s: &str) -> String {
2451    let mut out = String::new();
2452    let mut last_was_sep = true;
2453    for ch in s.to_lowercase().chars() {
2454        if ch.is_ascii_alphanumeric() {
2455            out.push(ch);
2456            last_was_sep = false;
2457        } else if !last_was_sep {
2458            out.push(' ');
2459            last_was_sep = true;
2460        }
2461    }
2462    out.trim().to_string()
2463}
2464
2465/// Cheap fuzzy match, 0-100: exact > substring > token overlap.
2466fn fuzzy_score(hint: &str, candidate: &str) -> f64 {
2467    let h = normalize_for_match(hint);
2468    let c = normalize_for_match(candidate);
2469    if h.is_empty() || c.is_empty() {
2470        return 0.0;
2471    }
2472    if h == c {
2473        return 100.0;
2474    }
2475    if c.contains(&h) || h.contains(&c) {
2476        return 75.0;
2477    }
2478    let h_tokens: std::collections::HashSet<&str> =
2479        h.split(' ').filter(|s| !s.is_empty()).collect();
2480    let c_tokens: std::collections::HashSet<&str> =
2481        c.split(' ').filter(|s| !s.is_empty()).collect();
2482    let overlap = h_tokens.intersection(&c_tokens).count();
2483    if overlap == 0 {
2484        return 0.0;
2485    }
2486    (overlap as f64 / h_tokens.len().max(c_tokens.len()) as f64) * 60.0
2487}
2488
2489fn policy_to_query_filter(filter: &PolicyFilter) -> QueryPolicyFilter {
2490    QueryPolicyFilter {
2491        allow_schemas: filter.allow_schemas.clone(),
2492        deny_schemas: filter.deny_schemas.clone(),
2493        deny_tables: filter.deny_tables.clone(),
2494        pii_columns: filter.pii_columns.clone(),
2495    }
2496}
2497
2498fn require_select_or_with(filter: &PolicyFilter, sql: &str) -> Result<(), ToolError> {
2499    match validate_readonly_sql(sql)? {
2500        SqlDecision::Allow => {}
2501        SqlDecision::Reject => {
2502            return Err(ToolError::Execution(
2503                "Security Error: Only SELECT or WITH statements can be analyzed.".into(),
2504            ));
2505        }
2506    }
2507    enforce_read_table_policy(filter, sql)?;
2508    let trimmed = sql.trim().to_ascii_lowercase();
2509    if !(trimmed.starts_with("select") || trimmed.starts_with("with")) {
2510        return Err(ToolError::Execution(
2511            "Security Error: Only SELECT or WITH statements can be analyzed.".into(),
2512        ));
2513    }
2514    Ok(())
2515}
2516
2517fn rows_to_json(rows: &[tokio_postgres::Row]) -> Value {
2518    rows_to_json_array(rows)
2519}
2520
2521fn scores_equal(a: f64, b: f64) -> bool {
2522    (a - b).abs() <= f64::EPSILON * a.abs().max(b.abs()).max(1.0)
2523}
2524
2525fn read_recent_log_errors() -> Vec<String> {
2526    let path = std::env::var("NEXQL_MCP_LOG")
2527        .map(std::path::PathBuf::from)
2528        .ok()
2529        .or_else(|| {
2530            std::env::var_os("HOME").map(|h| {
2531                std::path::PathBuf::from(h)
2532                    .join(".config")
2533                    .join("nexql-mcp")
2534                    .join("logs")
2535                    .join("nexql-mcp.log")
2536            })
2537        });
2538
2539    let Some(log_path) = path else {
2540        return Vec::new();
2541    };
2542
2543    let Ok(content) = std::fs::read_to_string(&log_path) else {
2544        return Vec::new();
2545    };
2546
2547    content
2548        .lines()
2549        .rev()
2550        .take(50)
2551        .filter(|line| {
2552            line.contains("ERROR")
2553                || line.contains("WARN")
2554                || line.contains("failed")
2555                || line.contains("Error")
2556        })
2557        .map(String::from)
2558        .collect()
2559}
2560
2561#[cfg(test)]
2562mod tests {
2563    use super::*;
2564    use crate::plan::build_explain_sql;
2565    use nexql_policy::PolicyFilter;
2566    use serde_json::json;
2567
2568    use crate::session::{ConnectionInfo, ConnectionPolicy, ToolSession};
2569    use nexql_policy::{AccessMode, PolicyCaps};
2570
2571    fn test_conn() -> ConnectionInfo {
2572        ConnectionInfo {
2573            id: "conn-1".into(),
2574            name: "conn-1".into(),
2575            host: Some("127.0.0.1".into()),
2576            port: Some(5432),
2577            database: Some("appdb".into()),
2578            params: Default::default(),
2579            policy: ConnectionPolicy {
2580                access_mode: AccessMode::Read,
2581                caps: PolicyCaps::default(),
2582                filter: PolicyFilter::default(),
2583                environment: None,
2584            },
2585        }
2586    }
2587
2588    #[test]
2589    fn scores_equal_treats_near_duplicates_as_tied() {
2590        let s = 3.295836866004329_f64;
2591        assert!(super::scores_equal(s, s));
2592        assert!(super::scores_equal(s, s + f64::EPSILON));
2593    }
2594
2595    #[test]
2596    fn policy_maps_one_to_one() {
2597        let f = PolicyFilter {
2598            allow_schemas: vec!["public".into()],
2599            deny_schemas: vec!["pgboss".into()],
2600            deny_tables: vec!["auth.*".into()],
2601            pii_columns: vec!["public.users.ssn".into()],
2602        };
2603        let q = policy_to_query_filter(&f);
2604        assert_eq!(q.allow_schemas, f.allow_schemas);
2605        assert_eq!(q.deny_schemas, f.deny_schemas);
2606        assert_eq!(q.deny_tables, f.deny_tables);
2607        assert_eq!(q.pii_columns, f.pii_columns);
2608    }
2609
2610    #[test]
2611    fn ok_json_wraps_arrays_for_cursor_structured_content() {
2612        let out = ToolOutcome::ok_json(json!([{ "id": 1 }, { "id": 2 }]));
2613        assert!(!out.is_error);
2614        let s = out.structured.as_ref().unwrap();
2615        assert!(s.is_object(), "structuredContent must be object, got {s}");
2616        assert_eq!(s["rows"].as_array().unwrap().len(), 2);
2617        assert!(out.text.contains("\"rows\""));
2618    }
2619
2620    #[test]
2621    fn ok_json_leaves_objects_unchanged() {
2622        let out = ToolOutcome::ok_json(json!({ "kind": "table", "name": "orders" }));
2623        let s = out.structured.as_ref().unwrap();
2624        assert_eq!(s["kind"], "table");
2625        assert!(s.get("rows").is_none());
2626    }
2627
2628    #[test]
2629    fn router_specs_include_phase4_and_phase9() {
2630        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2631        let router = ToolRouter::with_index_store(session, None);
2632        assert_eq!(router.specs().len(), ToolName::ACTIVE.len());
2633        let names: Vec<_> = router.specs().iter().map(|s| s.name.as_str()).collect();
2634        assert!(names.contains(&"search_schema"));
2635        assert!(names.contains(&"get_ddl"));
2636        assert!(names.contains(&"explain_analyze"));
2637        assert!(names.contains(&"get_index_status"));
2638        assert!(names.contains(&"list_extensions"));
2639        assert!(names.contains(&"server_settings"));
2640        assert!(names.contains(&"suggest_indexes"));
2641        assert!(names.contains(&"find_unused_indexes"));
2642        assert!(names.contains(&"bloat_report"));
2643        assert!(names.contains(&"find_missing_fks"));
2644        assert!(names.contains(&"export_query"));
2645        assert!(names.contains(&"list_roles"));
2646        assert!(names.contains(&"db_dashboard"));
2647        assert!(names.contains(&"deep_plan_analysis"));
2648        assert!(names.contains(&"execute_sql"));
2649        assert!(names.contains(&"edit_row"));
2650        assert!(names.contains(&"import_data"));
2651        assert!(names.contains(&"apply_ddl"));
2652        assert!(names.contains(&"create_index_concurrently"));
2653        assert!(names.contains(&"run_maintenance"));
2654        assert!(names.contains(&"terminate_query"));
2655    }
2656
2657    #[tokio::test]
2658    async fn write_tools_refuse_read_mode() {
2659        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2660        let router = ToolRouter::with_index_store(session, None);
2661        for tool in [
2662            "execute_sql",
2663            "edit_row",
2664            "import_data",
2665            "apply_ddl",
2666            "create_index_concurrently",
2667            "run_maintenance",
2668            "terminate_query",
2669        ] {
2670            let out = router
2671                .call(tool, json!({ "sql": "SELECT 1", "table": "public.t", "rows": [], "action": "insert", "values": {}, "pid": 1 }))
2672                .await;
2673            assert!(out.is_error, "{tool}: {}", out.text);
2674            assert!(
2675                out.text.contains("write") || out.text.contains("admin"),
2676                "{tool}: {}",
2677                out.text
2678            );
2679        }
2680    }
2681
2682    #[tokio::test]
2683    async fn table_stats_rejects_injection_ref() {
2684        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2685        let router = ToolRouter::with_index_store(session, None);
2686        let out = router
2687            .call("table_stats", json!({ "ref": "public.users; DROP" }))
2688            .await;
2689        assert!(out.is_error, "{}", out.text);
2690        assert!(
2691            out.text.contains("Invalid object reference") || out.text.contains("invalid arguments"),
2692            "expected ref validation error, got: {}",
2693            out.text
2694        );
2695    }
2696
2697    #[test]
2698    fn explain_transaction_path_builds_readonly_sequence() {
2699        // Documented contract: BEGIN → SET TRANSACTION READ ONLY → EXPLAIN → ROLLBACK
2700        let explain = build_explain_sql("SELECT 1", true);
2701        assert!(explain.starts_with("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)"));
2702        assert!(!explain.to_ascii_lowercase().contains("commit"));
2703        let steps = ["BEGIN", "SET TRANSACTION READ ONLY", &explain, "ROLLBACK"];
2704        assert_eq!(steps.len(), 4);
2705        assert_eq!(steps[0], "BEGIN");
2706        assert_eq!(steps[1], "SET TRANSACTION READ ONLY");
2707        assert_eq!(steps[3], "ROLLBACK");
2708    }
2709
2710    #[tokio::test]
2711    async fn missing_index_returns_actionable_error() {
2712        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2713        let router = ToolRouter::with_index_store(session, None);
2714        let out = router
2715            .call("search_schema", json!({ "query": "users" }))
2716            .await;
2717        assert!(out.is_error, "{}", out.text);
2718        assert!(
2719            out.text.contains("rebuild_index"),
2720            "expected actionable hint, got: {}",
2721            out.text
2722        );
2723    }
2724
2725    #[tokio::test]
2726    async fn empty_index_dir_returns_build_hint() {
2727        let tmp = tempfile::TempDir::new().unwrap();
2728        let store = IndexStore::new(tmp.path());
2729        let session = ToolSession::for_tests(
2730            vec![test_conn()],
2731            PolicyFilter::default(),
2732            Some(IndexStore::new(tmp.path())),
2733        );
2734        let router = ToolRouter::with_index_store(session, Some(store));
2735        let out = router
2736            .call("describe_object", json!({ "ref": "public.users" }))
2737            .await;
2738        assert!(out.is_error, "{}", out.text);
2739        assert!(
2740            out.text.contains("rebuild_index"),
2741            "expected build hint, got: {}",
2742            out.text
2743        );
2744    }
2745
2746    #[tokio::test]
2747    async fn outcome_tagged_with_connection_id_and_database() {
2748        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2749        let router = ToolRouter::new(session);
2750        let out = router.call("list_connections", json!({})).await;
2751        let structured = out.structured.expect("structured outcome");
2752        assert_eq!(
2753            structured.get("connectionId").and_then(|v| v.as_str()),
2754            Some("conn-1")
2755        );
2756        assert_eq!(
2757            structured.get("database").and_then(|v| v.as_str()),
2758            Some("appdb")
2759        );
2760    }
2761
2762    #[tokio::test]
2763    async fn setup_connection_returns_needs_input_when_incomplete() {
2764        unsafe {
2765            std::env::remove_var("DATABASE_URL");
2766            std::env::remove_var("POSTGRES_URL");
2767            std::env::remove_var("PGHOST");
2768        }
2769        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2770        let router = ToolRouter::new(session);
2771        let out = router.call("setup_connection", json!({})).await;
2772        let structured = out.structured.expect("structured outcome");
2773        assert!(structured.get("status").is_some());
2774    }
2775
2776    #[tokio::test]
2777    async fn save_profile_persists_config() {
2778        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2779        let router = ToolRouter::new(session);
2780        let temp_dir = tempfile::tempdir().unwrap();
2781        let cfg_path = temp_dir.path().join("config.toml");
2782        unsafe {
2783            std::env::set_var("NEXQL_MCP_CONFIG", &cfg_path);
2784        }
2785
2786        let out = router
2787            .call(
2788                "save_profile",
2789                json!({
2790                    "name": "staging",
2791                    "host": "127.0.0.1",
2792                    "port": 5432,
2793                    "dbname": "stage_db",
2794                    "user": "stage_user"
2795                }),
2796            )
2797            .await;
2798
2799        let structured = out.structured.expect("structured outcome");
2800        assert_eq!(
2801            structured.get("status").and_then(|v| v.as_str()),
2802            Some("saved")
2803        );
2804        assert_eq!(
2805            structured.get("profile").and_then(|v| v.as_str()),
2806            Some("staging")
2807        );
2808    }
2809
2810    #[tokio::test]
2811    async fn check_ddl_safety_tool_dispatches_ast_report() {
2812        let session = ToolSession::for_tests(vec![test_conn()], PolicyFilter::default(), None);
2813        let router = ToolRouter::new(session);
2814        let out = router
2815            .call(
2816                "check_ddl_safety",
2817                json!({ "ddl": "CREATE INDEX idx_col ON users(col);" }),
2818            )
2819            .await;
2820        let structured = out.structured.expect("structured outcome");
2821        assert_eq!(
2822            structured.get("overall_risk").and_then(|v| v.as_str()),
2823            Some("CRITICAL")
2824        );
2825    }
2826}
nexql_tools/exec.rs

nexql_tools/
exec.rs
