//! WS5 exploration-elimination shortcuts.
//!
//! Two families live here:
//!
//! - **Real graph queries** over the already-built edge graph — `find_circular_imports`
//!   and `find_coupling_hotspots`. No analyze-time precompute (ADR-030): each is a
//!   cheap read over `edges`. Edge-derived, so results declare a confidence tier
//!   (ADR-028), default `>= resolved`.
//! - **Categorisation/churn shortcuts** (Task 4) — added alongside, each reading
//!   an existing signal (categorisation tag / git churn) and returning an honest
//!   empty result with a missing-signal note where the signal is absent.

use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};

use serde_json::{Value, json};

use loomweave_core::{EdgeConfidence, McpErrorCode};
use loomweave_storage::{
    call_edges_targeting, entities_by_churn, entities_targeted_by_unresolved_call_sites,
    entity_by_id, resolve_entity_ref,
};

use crate::ParamError;
use crate::ServerState;
use crate::catalogue::{Page, RawScope, finalize_entity_page, missing_signal};
use crate::{
    entity_json, flatten_storage_envelope_result, optional_bool, optional_confidence,
    optional_non_empty_string, required_str, success_envelope, tool_error_envelope,
};

/// Scan bound on edges materialised for a graph query.
const EDGE_SCAN_CAP: usize = 500_000;
/// Scan bound on entities materialised for the dead-code candidate set.
const ENTITY_SCAN_CAP: usize = 500_000;
const EDGE_SCAN_ORDER_BY: &str = "ORDER BY kind, from_id, to_id, confidence, \
     COALESCE(source_byte_start, -1), COALESCE(source_byte_end, -1)";

/// Categorisation tags whose union is the reachability root set for
/// `find_dead_code` — entities "called from outside" the codebase. Tag-emitting
/// plugins populate these; the empty-root guard protects indexes with no root
/// tags from a flood of false positives.
///
/// The trailing `wardline:*` entries are Wardline-derived trust boundaries
/// (clarion-bf496d55d1, §4.2): the Python plugin emits `wardline:external_boundary`
/// / `wardline:trusted` from the on-disk Wardline vocabulary descriptor
/// (`@external_boundary` / `@trusted` decorators) into `entity_tags` at analyze
/// time — a developer-annotated, higher-confidence "called from outside the
/// static graph" signal than the structural heuristics. They map onto the
/// existing entry-point / exported-api root classes (`external_boundary` →
/// entry point, `trusted` → exported API); for dead-code reachability only the
/// union matters, so both simply join the root set, reading the same single
/// `entity_tags` signal under the same host validation discipline as every
/// other tag. Enrich-only: with no Wardline descriptor no `wardline:*` tag is
/// emitted and the root set is byte-identical to before. Stale facts cannot
/// resurrect a deleted entity as a root — `entity_tags` rows cascade-delete
/// with their entity, and roots join only live `entities`.
const DEAD_CODE_ROOT_TAGS: &[&str] = &[
    "entry-point",
    "http-route",
    "test",
    "data-model",
    "cli-command",
    "exported-api",
    "wardline:external_boundary",
    "wardline:trusted",
];

/// Tags that force an entity to be treated as live regardless of static
/// reachability — dynamic-dispatch / reflection barriers. Better to under-report
/// dead code than to call a reflectively-reached function dead (fail toward
/// "live").
const DEAD_CODE_BARRIER_TAGS: &[&str] = &["dynamic-dispatch", "reflection"];

/// Tags excluding an entity from dead-code candidacy even when unreached —
/// framework-magic entry kinds (decorated handlers, plugin hooks) whose callers
/// are invisible to static analysis.
const DEAD_CODE_EXCLUDED_TAGS: &[&str] = &["framework-handler", "plugin-hook"];

/// Entity KINDS that are not code and therefore can never be "dead code":
/// core `file` anchors (a config sidecar like `.env.example` is not dead
/// code — dogfood-4 B2, weft-3fb0f5dfc7), the synthetic project anchor,
/// subsystems (groupings, not code), and guidance sheets. Exclusion is by
/// kind, not plugin, so a plugin-emitted non-code kind is covered too.
const DEAD_CODE_NON_CODE_KINDS: &[&str] = &["file", "project", "subsystem", "guidance"];

/// Runtime import predicate used by graph shortcuts. Missing or malformed
/// properties fail toward inclusion; explicit `type_only=true` or
/// `scope="function"` marks an import as non-module-runtime evidence.
const RUNTIME_IMPORT_EDGE_SQL: &str = "\
    (properties IS NULL \
     OR json_valid(properties) = 0 \
     OR (COALESCE(json_extract(properties, '$.type_only'), 0) != 1 \
         AND COALESCE(json_extract(properties, '$.scope'), 'module') = 'module'))";

/// Rule id for an emitted dead-code candidate (ADR-017 `LMWV-FACT-*` namespace).
const DEAD_CODE_RULE_ID: &str = "LMWV-FACT-DEAD-CODE-CANDIDATE";
/// Heuristic confidence for a dead-code candidate — never presented as certain.
const DEAD_CODE_CONFIDENCE: f64 = 0.6;
const HOTSPOTS_PAGE_DEFAULT: usize = 20;
const HOTSPOTS_PAGE_MAX: usize = 200;
const CYCLES_PAGE_DEFAULT: usize = 50;
const CYCLES_PAGE_MAX: usize = 200;

/// Reachability-root sourcing mode for `find_dead_code` (clarion-663aca16aa,
/// A5). `Explicit` (the default) takes roots verbatim from emitted root tags and
/// keeps the empty-root honest-empty guard + per-plugin missing-root exclusion.
/// `Auto` derives roots from the same emitted tags but declares the lower,
/// derived confidence and relaxes the per-plugin exclusion (a rootless plugin is
/// surveyed against the globally-derived root set rather than withheld).
#[derive(Clone, Copy, PartialEq, Eq)]
enum RootsMode {
    Explicit,
    Auto,
}

impl RootsMode {
    fn parse(arguments: &serde_json::Map<String, Value>) -> std::result::Result<Self, ParamError> {
        match optional_non_empty_string(arguments, "roots")?.as_deref() {
            None | Some("explicit") => Ok(RootsMode::Explicit),
            Some("auto") => Ok(RootsMode::Auto),
            Some(_) => Err(ParamError::new("roots must be one of explicit, auto")),
        }
    }

    fn as_str(self) -> &'static str {
        match self {
            RootsMode::Explicit => "explicit",
            RootsMode::Auto => "auto",
        }
    }
}

/// The confidence strings included at or below a requested tier (the tier is a
/// ceiling: `resolved` → resolved only; `inferred` → all).
fn allowed_confidences(max: EdgeConfidence) -> &'static [&'static str] {
    match max {
        EdgeConfidence::Resolved => &["resolved"],
        EdgeConfidence::Ambiguous => &["resolved", "ambiguous"],
        EdgeConfidence::Inferred => &["resolved", "ambiguous", "inferred"],
    }
}

fn confidence_in_clause(max: EdgeConfidence) -> String {
    allowed_confidences(max)
        .iter()
        .map(|c| format!("'{c}'"))
        .collect::<Vec<_>>()
        .join(", ")
}

impl ServerState {
    /// `find_circular_imports(scope?, confidence?, limit?)` — import cycles in the
    /// module import graph (`imports` edges). Each cycle is a strongly-connected
    /// component of size > 1 (or a self-import). Edge-derived: default confidence
    /// `resolved`; the tier is echoed. Scope restricts to cycles whose members are
    /// all in scope. Bounded; each member carries its `sei`.
    pub(crate) async fn tool_find_circular_imports(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let scope = RawScope::parse(arguments)?;
        let confidence = optional_confidence(arguments)?;
        let page = Page::parse(arguments, CYCLES_PAGE_DEFAULT, CYCLES_PAGE_MAX)?;
        let project_root = self.project_root.clone();
        let result = self
            .readers
            .with_reader(move |conn| {
                let filter = scope.resolve(conn)?;
                let (in_scope, scope_truncated) = filter.in_scope_ids(conn, &project_root)?;

                // Build the import adjacency, restricted to in-scope endpoints.
                let (mut adjacency, scan_truncated) =
                    import_adjacency_for_cycles(conn, confidence, EDGE_SCAN_CAP)?;
                if let Some(in_scope) = &in_scope {
                    adjacency = adjacency
                        .into_iter()
                        .filter_map(|(from, tos)| {
                            if !in_scope.contains(&from) {
                                return None;
                            }
                            let tos: Vec<String> =
                                tos.into_iter().filter(|to| in_scope.contains(to)).collect();
                            (!tos.is_empty()).then_some((from, tos))
                        })
                        .collect();
                }

                let cycles = strongly_connected_cycles(&adjacency);
                let total = cycles.len();
                let returned: Vec<Vec<String>> = cycles
                    .into_iter()
                    .skip(page.offset)
                    .take(page.limit)
                    .collect();
                let returned_count = returned.len();
                let truncated = page.offset.saturating_add(returned_count) < total;

                let cycles_json: Vec<Value> = returned
                    .iter()
                    .map(|members| {
                        let entities: Vec<Value> = members
                            .iter()
                            .map(|id| match entity_by_id(conn, id) {
                                Ok(Some(entity)) => entity_json(conn, &entity),
                                _ => json!({ "id": id, "sei": Value::Null }),
                            })
                            .collect();
                        json!({ "length": members.len(), "members": entities })
                    })
                    .collect();

                Ok(success_envelope(json!({
                    "cycles": cycles_json,
                    "confidence": confidence.as_str(),
                    "page": {
                        "total": total,
                        "offset": page.offset,
                        "limit": page.limit,
                        "returned": returned_count,
                        "truncated": truncated,
                    },
                    "scope_truncated": scope_truncated,
                    "scan_truncated": scan_truncated,
                })))
            })
            .await;
        Ok(flatten_storage_envelope_result(result))
    }

    /// `find_coupling_hotspots(limit?, scope?, confidence?)` — entities ranked by
    /// coupling (distinct fan-in + fan-out over all edges) within scope.
    /// Edge-derived: default confidence `resolved`; the tier is echoed. Bounded;
    /// each entity carries its `sei`.
    pub(crate) async fn tool_find_coupling_hotspots(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let scope = RawScope::parse(arguments)?;
        let confidence = optional_confidence(arguments)?;
        let app_only = optional_bool(arguments, "app_only")?.unwrap_or(false);
        let page = Page::parse(arguments, HOTSPOTS_PAGE_DEFAULT, HOTSPOTS_PAGE_MAX)?;
        let project_root = self.project_root.clone();
        let result = self
            .readers
            .with_reader(move |conn| {
                let filter = scope.resolve(conn)?;
                let (in_scope, scope_truncated) = filter.in_scope_ids(conn, &project_root)?;
                let in_clause = confidence_in_clause(confidence);
                // app_only: drop test-tagged / non-first-party entities both as
                // hotspot rows AND from the fan-in/out edge counts, so a hub's
                // coupling reflects only first-party app callers/callees.
                let app_excluded = if app_only {
                    non_app_entity_ids(conn)?
                } else {
                    HashSet::new()
                };

                let coupling = coupling_counts(conn, &in_clause, &app_excluded)?;

                let mut ranked: Vec<(String, i64, i64)> = coupling
                    .into_iter()
                    .filter(|(id, _)| in_scope.as_ref().is_none_or(|ids| ids.contains(id)))
                    .map(|(id, (fan_in, fan_out))| (id, fan_in, fan_out))
                    .collect();
                // Rank by total coupling desc, ties by id for determinism.
                ranked.sort_by(|a, b| (b.1 + b.2).cmp(&(a.1 + a.2)).then_with(|| a.0.cmp(&b.0)));

                let total = ranked.len();
                let returned: Vec<(String, i64, i64)> = ranked
                    .into_iter()
                    .skip(page.offset)
                    .take(page.limit)
                    .collect();
                let returned_count = returned.len();
                let truncated = page.offset.saturating_add(returned_count) < total;

                let hotspots: Vec<Value> = returned
                    .iter()
                    .map(|(id, fan_in, fan_out)| {
                        let entity = match entity_by_id(conn, id) {
                            Ok(Some(entity)) => entity_json(conn, &entity),
                            _ => json!({ "id": id, "sei": Value::Null }),
                        };
                        json!({
                            "entity": entity,
                            "fan_in": fan_in,
                            "fan_out": fan_out,
                            "coupling": fan_in + fan_out,
                        })
                    })
                    .collect();

                Ok(success_envelope(json!({
                    "hotspots": hotspots,
                    "confidence": confidence.as_str(),
                    "app_only": app_only,
                    "page": {
                        "total": total,
                        "offset": page.offset,
                        "limit": page.limit,
                        "returned": returned_count,
                        "truncated": truncated,
                    },
                    "scope_truncated": scope_truncated,
                })))
            })
            .await;
        Ok(flatten_storage_envelope_result(result))
    }

    /// `find_dead_code(scope?, limit?, offset?)` — entities not reachable from
    /// the root set (entry points ∪ exported API ∪ tests ∪ HTTP routes ∪ CLI
    /// commands ∪ data models) over the call+import graph.
    ///
    /// **Conservative by construction (fail toward "live").** Reachability
    /// counts *all* edge confidence tiers (resolved ∪ ambiguous ∪ inferred):
    /// including more edges keeps more entities live, so the tool under-reports
    /// rather than over-reports dead code. Hence no `confidence` argument — a
    /// ceiling would only make *more* code look dead, the harmful direction.
    /// Dynamic-dispatch / reflection barrier tags force their entities live, and
    /// framework-magic kinds are excluded from candidacy.
    ///
    /// **Empty-root guard.** When no root categorisation is emitted the root set
    /// is empty; the naive computation would then flag the *entire* codebase as
    /// dead. Instead the tool returns an honest signal-unavailable result with
    /// zero candidates — never a false positive. Results are heuristic
    /// (`LMWV-FACT-DEAD-CODE-CANDIDATE`, confidence < 1), bounded, SEI-carrying.
    pub(crate) async fn tool_find_dead_code(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let scope = RawScope::parse(arguments)?;
        let page = Page::parse(arguments, SHORTCUT_PAGE_DEFAULT, SHORTCUT_PAGE_MAX)?;
        let roots_mode = RootsMode::parse(arguments)?;
        let app_only = optional_bool(arguments, "app_only")?.unwrap_or(false);
        let project_root = self.project_root.clone();
        let result = self
            .readers
            .with_reader(move |conn| {
                let filter = scope.resolve(conn)?;
                let (in_scope, scope_truncated) = filter.in_scope_ids(conn, &project_root)?;

                // Roots = "called from outside" categorisations. Both modes seed
                // from the same emitted tags; `auto` only changes the honesty
                // reporting and relaxes the per-plugin missing-root exclusion.
                let roots = ids_with_any_tag(conn, DEAD_CODE_ROOT_TAGS)?;
                if roots.is_empty() {
                    // No root tags (in either mode): honest-empty, never a
                    // whole-corpus false positive.
                    return Ok(dead_code_no_roots_envelope(&page, scope_truncated));
                }

                // Forward BFS over call+import edges across ALL confidence tiers
                // (fail toward live). Reachability is whole-graph: an in-scope
                // entity reached via an out-of-scope caller must not be flagged.
                let (adjacency, scan_truncated) = call_import_adjacency(conn)?;
                let barriers = ids_with_any_tag(conn, DEAD_CODE_BARRIER_TAGS)?;
                let mut live: HashSet<String> = roots;
                live.extend(barriers);
                let reachable = forward_reachable(&adjacency, live);

                let excluded = ids_with_any_tag(conn, DEAD_CODE_EXCLUDED_TAGS)?;
                // Fail toward live: an entity whose name matches an unresolved
                // call site is a plausible callee and must NOT be flagged dead.
                // The Rust resolver emits no `calls` edge for `x.method()` /
                // `Type::assoc()` (no type inference), so those callees are
                // invisible to static reachability — without this they would be
                // false-flagged dead. Language-agnostic: a fully-resolving
                // plugin (pyright-backed Python) leaves this set empty.
                let unresolved_targets = entities_targeted_by_unresolved_call_sites(conn)?;
                // app_only: drop test-tagged / non-first-party entities from the
                // survey (read-only, computed on query). Empty set when off.
                let app_excluded = if app_only {
                    non_app_entity_ids(conn)?
                } else {
                    HashSet::new()
                };
                let CandidateSet {
                    mut candidates,
                    excluded_by_plugin,
                    blocked_reason_by_id,
                    entity_scan_truncated,
                } = dead_code_candidate_set(
                    conn,
                    &reachable,
                    &excluded,
                    in_scope.as_ref(),
                    roots_mode,
                    &app_excluded,
                )?;

                // Count the unresolved-call-site shield separately so the
                // disclosure is exact, then remove the shielded candidates.
                let unresolved_call_site_suppressed = candidates
                    .iter()
                    .filter(|id| unresolved_targets.contains(*id))
                    .count();
                candidates.retain(|id| !unresolved_targets.contains(id));

                let (withheld_count, withheld_reasons) =
                    withhold_blocked_candidates(&mut candidates, &blocked_reason_by_id);

                let total = candidates.len();
                let returned: Vec<String> = candidates
                    .into_iter()
                    .skip(page.offset)
                    .take(page.limit)
                    .collect();
                let returned_count = returned.len();
                let truncated = page.offset.saturating_add(returned_count) < total;

                // Per-row payload is the entity alone; the constant facets
                // (rule/kind/confidence/reason) are stated ONCE in the
                // top-level `finding` block instead of verbatim on every row
                // of every page (weft-3fb0f5dfc7 / dogfood-4 B2).
                let dead_code: Vec<Value> = returned
                    .iter()
                    .map(|id| {
                        let entity = match entity_by_id(conn, id) {
                            Ok(Some(entity)) => entity_json(conn, &entity),
                            _ => json!({ "id": id, "sei": Value::Null }),
                        };
                        json!({ "entity": entity })
                    })
                    .collect();

                Ok(success_envelope(json!({
                    // Lead summary: counts by kind + a confidence verdict, so the
                    // headline is self-qualifying (the agent-first "number of
                    // things in various kinds → state the breakdown up front"
                    // convention; reference implementation).
                    "summary": dead_code_summary(
                        total,
                        reachable.len(),
                        excluded.len(),
                        unresolved_call_site_suppressed,
                        withheld_count,
                        roots_mode,
                    ),
                    "app_only": app_only,
                    "dead_code": dead_code,
                    // The constant facets every candidate shares, hoisted.
                    "finding": {
                        "rule_id": DEAD_CODE_RULE_ID,
                        "kind": "fact",
                        "confidence": DEAD_CODE_CONFIDENCE,
                        "confidence_basis": "heuristic",
                        "reason": "unreachable from the reachability root set over call+import \
                                   edges across all confidence tiers; static reachability \
                                   cannot prove dynamic or reflective reach",
                    },
                    "excluded": {
                        "non_code_kinds": DEAD_CODE_NON_CODE_KINDS,
                        "plugins_without_roots": plugins_without_roots_json(excluded_by_plugin),
                    },
                    // Aggregate-level actionability without identity
                    // disclosure: how many candidate rows were withheld as
                    // briefing-blocked, why, and how to recover them.
                    "withheld": {
                        "count": withheld_count,
                        "reasons": withheld_reasons.iter().collect::<Vec<_>>(),
                        "recovery": withheld_reasons
                            .iter()
                            .next()
                            .map(|reason| crate::briefing_block_remediation(reason)),
                    },
                    "page": {
                        "total": total,
                        "offset": page.offset,
                        "limit": page.limit,
                        "returned": returned_count,
                        "truncated": truncated,
                    },
                    "scope_truncated": scope_truncated,
                    "scan_truncated": scan_truncated || entity_scan_truncated,
                    "unresolved_call_site_suppressed": unresolved_call_site_suppressed,
                })))
            })
            .await;
        Ok(flatten_storage_envelope_result(result))
    }
}

const SHORTCUT_PAGE_DEFAULT: usize = 50;
const SHORTCUT_PAGE_MAX: usize = 200;
const CHURN_SCAN_CAP: usize = 50_000;

/// The honest signal-unavailable envelope for `find_dead_code` when no
/// reachability root tags exist — zero candidates, never a whole-corpus false
/// positive. Identical in both `explicit` and `auto` modes: `auto` cannot
/// fabricate roots from an empty tag set.
fn dead_code_no_roots_envelope(page: &Page, scope_truncated: bool) -> Value {
    success_envelope(json!({
        "dead_code": [],
        "page": {
            "total": 0, "offset": page.offset, "limit": page.limit,
            "returned": 0, "truncated": false,
        },
        "scope_truncated": scope_truncated,
        "scan_truncated": false,
        "signal": missing_signal(
            "entity_tags",
            "this index has no reachability root tags (entry-point / http-route / \
             test / data-model / cli-command / exported-api), so dead code cannot \
             be determined — this is NOT a guarantee there is no dead code",
        ),
    }))
}

/// Build the lead `summary` block for `find_dead_code` — the agent-first "state
/// the breakdown up front" convention: counts by kind plus a confidence verdict,
/// so the headline is self-qualifying and no agent is misled by a raw candidate
/// count. A LOW confidence verdict (with a recruiting advisory) fires when an
/// implausibly large share of analysed entities is unreachable — a sign the
/// reachability roots do not cover the corpus (e.g. a library exercised only by
/// tests) rather than that the code is genuinely dead.
fn dead_code_summary(
    dead_candidates: usize,
    reachable: usize,
    plugins_without_roots: usize,
    shielded_by_unresolved_calls: usize,
    withheld_secret: usize,
    roots_mode: RootsMode,
) -> Value {
    let analysed = reachable.saturating_add(dead_candidates);
    let dead_pct = dead_candidates
        .saturating_mul(100)
        .checked_div(analysed)
        .unwrap_or(0);
    let low_confidence = dead_pct > 25;
    let advisory = low_confidence.then(|| {
        format!(
            "{dead_pct}% of analysed entities ({dead_candidates}/{analysed}) are unreachable \
             from the configured reachability roots — implausibly high, so the roots likely do \
             not cover this corpus (e.g. a library exercised only by tests). Treat these \
             candidates as LOW CONFIDENCE: configure entry-point roots (tag cli-command / \
             entry-point / test / exported-api / data-model entities) before relying on \
             dead-code detection."
        )
    });
    json!({
        "dead_candidates": dead_candidates,
        "reachable": reachable,
        "not_analysed": {
            "plugins_without_roots": plugins_without_roots,
            "shielded_by_unresolved_calls": shielded_by_unresolved_calls,
            "withheld_secret": withheld_secret,
        },
        "confidence": if low_confidence { "low" } else { "moderate" },
        "advisory": advisory,
        // How the reachability roots were sourced. `auto` derives them from
        // emitted tags, so it declares the lower, derived confidence.
        "roots_mode": roots_mode.as_str(),
        "roots_confidence": match roots_mode {
            RootsMode::Auto => json!("derived"),
            RootsMode::Explicit => Value::Null,
        },
    })
}

impl ServerState {
    /// `find_entry_points(scope?)` — entities tagged as entry points. Reads the
    /// `entry-point` categorisation tag; honest-empty when no plugin emits it.
    pub(crate) async fn tool_find_entry_points(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        self.categorisation_shortcut(
            arguments,
            "entry-point",
            "no entity is tagged as an entry point in this index (honest-empty, not a guaranteed \
             absence of entry points)",
        )
        .await
    }

    /// `find_http_routes(scope?)` — entities tagged as HTTP routes (honest-empty
    /// when the `http-route` tag is not emitted).
    pub(crate) async fn tool_find_http_routes(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        self.categorisation_shortcut(
            arguments,
            "http-route",
            "no entity is tagged as an HTTP route in this index",
        )
        .await
    }

    /// `find_data_models(scope?)` — entities tagged as data models (honest-empty
    /// when the `data-model` tag is not emitted).
    pub(crate) async fn tool_find_data_models(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        self.categorisation_shortcut(
            arguments,
            "data-model",
            "no entity is tagged as a data model in this index",
        )
        .await
    }

    /// `find_tests(scope?)` — entities tagged as tests (honest-empty when the
    /// `test` tag is not emitted).
    pub(crate) async fn tool_find_tests(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        self.categorisation_shortcut(
            arguments,
            "test",
            "no entity is tagged as a test in this index",
        )
        .await
    }

    /// `find_deprecations(scope?)` — entities tagged deprecated (honest-empty
    /// when the `deprecated` tag is not emitted).
    pub(crate) async fn tool_find_deprecations(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        self.categorisation_shortcut(
            arguments,
            "deprecated",
            "no entity is tagged as deprecated in this index",
        )
        .await
    }

    /// `find_todos(scope?)` — entities tagged with a TODO marker (honest-empty
    /// when the `todo` tag is not emitted).
    pub(crate) async fn tool_find_todos(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        self.categorisation_shortcut(
            arguments,
            "todo",
            "no entity is tagged with a TODO/FIXME marker in this index",
        )
        .await
    }

    /// Shared body for the categorisation shortcuts: parse scope/page, then run
    /// the canonical tag through [`ServerState::tag_facet`].
    async fn categorisation_shortcut(
        &self,
        arguments: &serde_json::Map<String, Value>,
        tag: &'static str,
        missing_reason: &'static str,
    ) -> std::result::Result<Value, ParamError> {
        let scope = RawScope::parse(arguments)?;
        let page = Page::parse(arguments, SHORTCUT_PAGE_DEFAULT, SHORTCUT_PAGE_MAX)?;
        Ok(self
            .tag_facet(tag.to_owned(), missing_reason, scope, page)
            .await)
    }

    /// `what_tests_this(id)` — the test entities that exercise an entity: its
    /// callers that carry the `test` categorisation tag. Honest-empty when test
    /// categorisation is not emitted (so an empty result is never read as "this
    /// is untested"). Stateless, bounded, SEI-carrying.
    pub(crate) async fn tool_what_tests_this(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let entity_id = required_str(arguments, "id")?.to_owned();
        let page = Page::parse(arguments, SHORTCUT_PAGE_DEFAULT, SHORTCUT_PAGE_MAX)?;
        let result = self
            .readers
            .with_reader(move |conn| {
                let Some(entity) = resolve_entity_ref(conn, &entity_id)? else {
                    return Ok(tool_error_envelope(
                        McpErrorCode::EntityNotFound,
                        &format!("entity {entity_id} was not found"),
                        false,
                    ));
                };

                // Callers tagged `test`. Resolved-tier callers only.
                let callers = call_edges_targeting(conn, &entity.id, EdgeConfidence::Resolved)?;
                let caller_ids: HashSet<String> =
                    callers.into_iter().map(|edge| edge.from_id).collect();

                // One query for the test-tagged subset, then materialise only
                // those rows (avoids a per-caller tag probe + lookup).
                let test_ids = test_tagged_subset(conn, &caller_ids)?;
                let mut test_callers: Vec<Value> = Vec::new();
                for caller_id in &test_ids {
                    if let Some(caller) = entity_by_id(conn, caller_id)? {
                        test_callers.push(entity_json(conn, &caller));
                    }
                }
                test_callers.sort_by(|a, b| a["id"].as_str().cmp(&b["id"].as_str()));

                let total = test_callers.len();
                let returned: Vec<Value> = test_callers
                    .into_iter()
                    .skip(page.offset)
                    .take(page.limit)
                    .collect();
                let returned_count = returned.len();
                let truncated = page.offset.saturating_add(returned_count) < total;

                let mut response = json!({
                    "entity": entity_json(conn, &entity),
                    "tests": returned,
                    "page": {
                        "total": total,
                        "offset": page.offset,
                        "limit": page.limit,
                        "returned": returned_count,
                        "truncated": truncated,
                    },
                });
                if total == 0
                    && let Some(object) = response.as_object_mut()
                {
                    object.insert(
                        "signal".to_owned(),
                        missing_signal(
                            "entity_tags",
                            "no test-tagged caller found in this index, so this is not a guarantee \
                             the entity is untested",
                        ),
                    );
                }
                Ok(success_envelope(response))
            })
            .await;
        Ok(flatten_storage_envelope_result(result))
    }

    /// `high_churn(limit?, scope?)` — entities ranked by `git_churn_count`
    /// descending. The analyze pipeline does not populate churn in v1.0, so this
    /// is honest-empty in practice (the missing signal is surfaced); the query is
    /// real, so it lights up if churn is ever populated. Bounded, SEI-carrying.
    pub(crate) async fn tool_high_churn(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let scope = RawScope::parse(arguments)?;
        let page = Page::parse(arguments, SHORTCUT_PAGE_DEFAULT, SHORTCUT_PAGE_MAX)?;
        let project_root = self.project_root.clone();
        let result = self
            .readers
            .with_reader(move |conn| {
                let filter = scope.resolve(conn)?;
                let (rows, scan_truncated) = entities_by_churn(conn, CHURN_SCAN_CAP)?;
                // Keep churn alongside; finalize over the entity rows, then graft
                // the churn count onto each returned entity.
                let churn_by_id: std::collections::HashMap<String, i64> =
                    rows.iter().map(|(e, c)| (e.id.clone(), *c)).collect();
                let entities: Vec<_> = rows.into_iter().map(|(e, _)| e).collect();
                let mut response = finalize_entity_page(
                    conn,
                    &project_root,
                    entities,
                    &filter,
                    page,
                    scan_truncated,
                );
                if let Some(list) = response["entities"].as_array() {
                    let grafted: Vec<Value> = list
                        .iter()
                        .map(|entity| {
                            let mut entity = entity.clone();
                            if let Some(object) = entity.as_object_mut()
                                && let Some(id) = object.get("id").and_then(Value::as_str)
                                && let Some(churn) = churn_by_id.get(id)
                            {
                                object.insert("git_churn_count".to_owned(), json!(churn));
                            }
                            entity
                        })
                        .collect();
                    if let Some(object) = response.as_object_mut() {
                        object.insert("entities".to_owned(), Value::Array(grafted));
                    }
                }
                if response["page"]["total"] == json!(0)
                    && let Some(object) = response.as_object_mut()
                {
                    object.insert(
                        "signal".to_owned(),
                        missing_signal(
                            "git_churn_count",
                            "no entity carries git churn; the analyze pipeline does not populate \
                             git_churn_count in v1.0",
                        ),
                    );
                }
                Ok(success_envelope(response))
            })
            .await;
        Ok(flatten_storage_envelope_result(result))
    }

    /// `recently_changed(since?, scope?)` — entities changed since a timestamp.
    /// Loomweave does not index a per-entity git change timestamp in v1.0, so this
    /// is an honest no-op: it returns an empty set with a missing-signal note
    /// pointing at `index_diff` for repo-level freshness. The args are accepted
    /// for forward-compatibility. Never fabricates a change set.
    // Honest no-op: no storage read, but kept `async` for the uniform tool
    // dispatch interface (every `tool_*` is awaited in `handle_tool_call`).
    #[allow(clippy::unused_async)]
    pub(crate) async fn tool_recently_changed(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        // Validate args so a malformed call still errors honestly.
        let _ = RawScope::parse(arguments)?;
        let since = match arguments.get("since") {
            None | Some(Value::Null) => None,
            Some(Value::String(value)) => Some(value.clone()),
            Some(_) => return Err(ParamError::new("since must be an ISO-8601 string or null")),
        };
        Ok(success_envelope(json!({
            "entities": [],
            "since": since,
            "page": { "total": 0, "offset": 0, "limit": 0, "returned": 0, "truncated": false },
            "signal": missing_signal(
                "git_change_time",
                "Loomweave does not index a per-entity git change timestamp in v1.0; use index_diff \
                 for repo-level freshness (HEAD vs last analyze)"
            ),
        })))
    }
}

/// Of the given entity ids, those carrying the `test` categorisation tag.
/// One chunked `IN` query rather than a probe per id.
fn test_tagged_subset(
    conn: &rusqlite::Connection,
    ids: &HashSet<String>,
) -> loomweave_storage::Result<HashSet<String>> {
    let mut out = HashSet::new();
    let all: Vec<&String> = ids.iter().collect();
    for chunk in all.chunks(500) {
        let placeholders = std::iter::repeat_n("?", chunk.len())
            .collect::<Vec<_>>()
            .join(", ");
        let sql = format!(
            "SELECT entity_id FROM entity_tags WHERE tag = 'test' AND entity_id IN ({placeholders})"
        );
        let mut stmt = conn.prepare(&sql)?;
        let mut rows = stmt.query(rusqlite::params_from_iter(chunk.iter()))?;
        while let Some(row) = rows.next()? {
            out.insert(row.get::<_, String>(0)?);
        }
    }
    Ok(out)
}

/// Entity ids that are NOT first-party application code, for the opt-in
/// `app_only` filter (clarion-663aca16aa, A5): an entity is excluded when it is
/// (a) `test`-tagged or (b) owned by the `core` plugin (Loomweave's synthetic
/// file/project/subsystem/guidance anchors — never app code; `tour.*`-style
/// non-first-party fixtures land here too). Pure read over `entity_tags` +
/// `entities`; no summary dependency, no re-analysis. A third-party plugin
/// blocklist can extend this later without touching callers.
fn non_app_entity_ids(conn: &rusqlite::Connection) -> loomweave_storage::Result<HashSet<String>> {
    let mut out = ids_with_any_tag(conn, &["test"])?;
    let mut stmt = conn.prepare("SELECT id FROM entities WHERE plugin_id = 'core'")?;
    let mut rows = stmt.query([])?;
    while let Some(row) = rows.next()? {
        out.insert(row.get::<_, String>(0)?);
    }
    Ok(out)
}

/// Entity ids carrying any of `tags` (one `IN` query, distinct ids).
fn ids_with_any_tag(
    conn: &rusqlite::Connection,
    tags: &[&str],
) -> loomweave_storage::Result<HashSet<String>> {
    if tags.is_empty() {
        return Ok(HashSet::new());
    }
    let placeholders = std::iter::repeat_n("?", tags.len())
        .collect::<Vec<_>>()
        .join(", ");
    let sql = format!("SELECT DISTINCT entity_id FROM entity_tags WHERE tag IN ({placeholders})");
    let mut stmt = conn.prepare(&sql)?;
    let mut rows = stmt.query(rusqlite::params_from_iter(tags.iter()))?;
    let mut out = HashSet::new();
    while let Some(row) = rows.next()? {
        out.insert(row.get::<_, String>(0)?);
    }
    Ok(out)
}

/// Withhold briefing-blocked candidates from the dead-code row set
/// (weft-3fb0f5dfc7, dogfood-4 B2 + PM ruling): an all-null stub row is
/// unactionable noise, and an identity-bearing row would breach the
/// stub-collapse non-disclosure invariant (clarion-307668e2be) — so neither
/// appears. The exclusion is reported ONCE, in aggregate; counted only over
/// rows that would otherwise have been candidates, so the count is exact.
fn withhold_blocked_candidates(
    candidates: &mut Vec<String>,
    blocked_reason_by_id: &HashMap<String, String>,
) -> (usize, BTreeSet<String>) {
    let mut reasons: BTreeSet<String> = BTreeSet::new();
    let mut count = 0usize;
    candidates.retain(|id| match blocked_reason_by_id.get(id) {
        Some(reason) => {
            count += 1;
            reasons.insert(reason.clone());
            false
        }
        None => true,
    });
    (count, reasons)
}

/// Render the per-plugin missing-root-coverage exclusions as the in-band
/// scope marker.
fn plugins_without_roots_json(excluded_by_plugin: BTreeMap<String, usize>) -> Vec<Value> {
    excluded_by_plugin
        .into_iter()
        .map(|(plugin, count)| {
            json!({
                "plugin": plugin,
                "entities_excluded": count,
                "reason": format!(
                    "plugin '{plugin}' emitted no reachability root tags \
                     (entry-point / exported-api / …), so its entities cannot \
                     be honestly surveyed — excluded rather than false-flagged \
                     dead"
                ),
            })
        })
        .collect()
}

/// The filtered dead-code candidate id set plus the in-band exclusion
/// bookkeeping (`weft-3fb0f5dfc7`).
struct CandidateSet {
    /// Sorted candidate ids: code-kind entities of root-covered plugins that
    /// are unreachable, not barrier/framework-excluded, and in scope.
    candidates: Vec<String>,
    /// Per-plugin counts of entities excluded for missing root coverage.
    excluded_by_plugin: BTreeMap<String, usize>,
    /// Briefing-block reason per surviving candidate id (withheld later).
    blocked_reason_by_id: HashMap<String, String>,
    entity_scan_truncated: bool,
}

/// Build the dead-code candidate set with the survey-honesty exclusions
/// (weft-3fb0f5dfc7 / dogfood-4 B2):
///
/// - non-code KINDS (file anchors, the project anchor, subsystems, guidance)
///   are never dead-CODE candidates;
/// - a plugin that emitted NO reachability root tags (the Rust plugin today —
///   binary/lib roots unsupported) would have its ENTIRE entity set
///   false-flagged dead, so its entities are excluded and counted for the
///   in-band scope marker instead — a wrong answer is worse than an honest
///   scope statement (the dogfooded `specimen-rs/src/main.rs` false positive).
fn dead_code_candidate_set(
    conn: &rusqlite::Connection,
    reachable: &HashSet<String>,
    excluded: &HashSet<String>,
    in_scope: Option<&HashSet<String>>,
    roots_mode: RootsMode,
    app_excluded: &HashSet<String>,
) -> loomweave_storage::Result<CandidateSet> {
    let (all_rows, entity_scan_truncated) = all_entity_rows(conn)?;
    // In `auto` mode roots are derived globally from emitted tags, so a plugin
    // without root tags of its own is still surveyed against that set rather
    // than withheld — the per-plugin missing-root exclusion only applies to the
    // verbatim `explicit` mode.
    let plugins_with_roots = match roots_mode {
        RootsMode::Explicit => Some(plugins_with_root_tags(conn)?),
        RootsMode::Auto => None,
    };
    let mut excluded_by_plugin: BTreeMap<String, usize> = BTreeMap::new();
    let mut blocked_reason_by_id: HashMap<String, String> = HashMap::new();

    let mut candidates: Vec<String> = all_rows
        .into_iter()
        .filter(|row| !DEAD_CODE_NON_CODE_KINDS.contains(&row.kind.as_str()))
        .filter(|row| match &plugins_with_roots {
            None => true,
            Some(plugins_with_roots) => {
                if plugins_with_roots.contains(&row.plugin_id) {
                    true
                } else {
                    *excluded_by_plugin.entry(row.plugin_id.clone()).or_default() += 1;
                    false
                }
            }
        })
        .map(|row| {
            if let Some(reason) = row.briefing_blocked {
                blocked_reason_by_id.insert(row.id.clone(), reason);
            }
            row.id
        })
        .filter(|id| !reachable.contains(id))
        .filter(|id| !excluded.contains(id))
        .filter(|id| !app_excluded.contains(id))
        .filter(|id| in_scope.is_none_or(|ids| ids.contains(id)))
        .collect();
    candidates.sort();

    Ok(CandidateSet {
        candidates,
        excluded_by_plugin,
        blocked_reason_by_id,
        entity_scan_truncated,
    })
}

/// A dead-code candidate row: just enough to apply the kind / plugin-coverage
/// exclusions without materialising full entities.
struct CandidateRow {
    id: String,
    kind: String,
    plugin_id: String,
    /// The briefing-block reason (the `briefing_blocked` generated column,
    /// migration 0002) — `Some` for a secret-scan-withheld entity.
    briefing_blocked: Option<String>,
}

/// All entity (id, kind, plugin, block-reason) rows, bounded by
/// [`ENTITY_SCAN_CAP`]. Returns `(rows, truncated)`.
fn all_entity_rows(
    conn: &rusqlite::Connection,
) -> loomweave_storage::Result<(Vec<CandidateRow>, bool)> {
    let cap = i64::try_from(ENTITY_SCAN_CAP.saturating_add(1)).unwrap_or(i64::MAX);
    let mut stmt = conn.prepare(
        "SELECT id, kind, plugin_id, briefing_blocked FROM entities ORDER BY id LIMIT ?1",
    )?;
    let mut rows = stmt.query(rusqlite::params![cap])?;
    let mut out = Vec::new();
    let mut truncated = false;
    while let Some(row) = rows.next()? {
        if out.len() >= ENTITY_SCAN_CAP {
            truncated = true;
            break;
        }
        out.push(CandidateRow {
            id: row.get::<_, String>(0)?,
            kind: row.get::<_, String>(1)?,
            plugin_id: row.get::<_, String>(2)?,
            briefing_blocked: row.get::<_, Option<String>>(3)?,
        });
    }
    Ok((out, truncated))
}

/// The plugins that own at least one ROOT-tagged entity — i.e. the plugins
/// whose entity sets the dead-code survey can answer honestly. One query over
/// `entity_tags ⋈ entities`.
fn plugins_with_root_tags(
    conn: &rusqlite::Connection,
) -> loomweave_storage::Result<HashSet<String>> {
    let placeholders = std::iter::repeat_n("?", DEAD_CODE_ROOT_TAGS.len())
        .collect::<Vec<_>>()
        .join(", ");
    let sql = format!(
        "SELECT DISTINCT e.plugin_id FROM entity_tags t \
         JOIN entities e ON e.id = t.entity_id \
         WHERE t.tag IN ({placeholders})"
    );
    let mut stmt = conn.prepare(&sql)?;
    let mut rows = stmt.query(rusqlite::params_from_iter(DEAD_CODE_ROOT_TAGS.iter()))?;
    let mut out = HashSet::new();
    while let Some(row) = rows.next()? {
        out.insert(row.get::<_, String>(0)?);
    }
    Ok(out)
}

/// The call+import adjacency over *all* confidence tiers (the conservative,
/// fail-toward-live choice for reachability). Bounded by [`EDGE_SCAN_CAP`];
/// returns `(adjacency, scan_truncated)`.
fn call_import_adjacency(
    conn: &rusqlite::Connection,
) -> loomweave_storage::Result<(HashMap<String, Vec<String>>, bool)> {
    call_import_adjacency_with_cap(conn, EDGE_SCAN_CAP)
}

/// Per-entity `(fan_in, fan_out)` distinct-neighbour coupling over the call +
/// runtime-import edge graph (spec §3.3). Structural edges (`contains`,
/// `in_subsystem`, `guides`, `emits_finding`) are excluded so the ranking
/// reflects actionable coupling, not containment fan-out.
///
/// When `app_excluded` is non-empty (the opt-in `app_only` filter), an edge is
/// dropped entirely if EITHER endpoint is excluded — so a test-tagged caller
/// contributes to neither its own fan-out nor the hub's fan-in, and the
/// excluded entity never surfaces as a hotspot row. The distinct counting is
/// done in Rust (rather than `COUNT(DISTINCT …)` in SQL) so the endpoint
/// exclusion composes cleanly; when `app_excluded` is empty this is identical
/// to the previous aggregate.
fn coupling_counts(
    conn: &rusqlite::Connection,
    in_clause: &str,
    app_excluded: &HashSet<String>,
) -> loomweave_storage::Result<HashMap<String, (i64, i64)>> {
    let kinds = format!("(kind = 'calls' OR (kind = 'imports' AND {RUNTIME_IMPORT_EDGE_SQL}))");
    let sql = format!(
        "SELECT DISTINCT from_id, to_id FROM edges \
         WHERE {kinds} AND confidence IN ({in_clause})"
    );
    let mut out_neighbours: HashMap<String, HashSet<String>> = HashMap::new();
    let mut in_neighbours: HashMap<String, HashSet<String>> = HashMap::new();
    let mut stmt = conn.prepare(&sql)?;
    let mut rows = stmt.query([])?;
    while let Some(row) = rows.next()? {
        let from: String = row.get(0)?;
        let to: String = row.get(1)?;
        if app_excluded.contains(&from) || app_excluded.contains(&to) {
            continue;
        }
        out_neighbours
            .entry(from.clone())
            .or_default()
            .insert(to.clone());
        in_neighbours.entry(to).or_default().insert(from);
    }
    let mut coupling: HashMap<String, (i64, i64)> = HashMap::new();
    for (id, tos) in out_neighbours {
        coupling.entry(id).or_default().1 = i64::try_from(tos.len()).unwrap_or(i64::MAX);
    }
    for (id, froms) in in_neighbours {
        coupling.entry(id).or_default().0 = i64::try_from(froms.len()).unwrap_or(i64::MAX);
    }
    Ok(coupling)
}

fn import_adjacency_for_cycles(
    conn: &rusqlite::Connection,
    confidence: EdgeConfidence,
    scan_cap: usize,
) -> loomweave_storage::Result<(HashMap<String, Vec<String>>, bool)> {
    let in_clause = confidence_in_clause(confidence);
    let sql = format!(
        "SELECT from_id, to_id FROM edges \
         WHERE kind = 'imports' \
           AND confidence IN ({in_clause}) \
           AND {RUNTIME_IMPORT_EDGE_SQL} \
         {EDGE_SCAN_ORDER_BY} \
         LIMIT ?1"
    );
    let cap = i64::try_from(scan_cap.saturating_add(1)).unwrap_or(i64::MAX);
    let mut stmt = conn.prepare(&sql)?;
    let mut rows = stmt.query(rusqlite::params![cap])?;
    let mut adjacency: HashMap<String, Vec<String>> = HashMap::new();
    let mut edge_count = 0usize;
    let mut truncated = false;
    while let Some(row) = rows.next()? {
        if edge_count >= scan_cap {
            truncated = true;
            break;
        }
        edge_count += 1;
        let from: String = row.get(0)?;
        let to: String = row.get(1)?;
        adjacency.entry(from).or_default().push(to);
    }
    Ok((adjacency, truncated))
}

fn call_import_adjacency_with_cap(
    conn: &rusqlite::Connection,
    scan_cap: usize,
) -> loomweave_storage::Result<(HashMap<String, Vec<String>>, bool)> {
    let cap = i64::try_from(scan_cap.saturating_add(1)).unwrap_or(i64::MAX);
    let mut stmt = conn.prepare(
        "SELECT kind, from_id, to_id, confidence, properties \
         FROM edges \
         WHERE (kind = 'calls' OR (kind = 'imports' AND \
               (properties IS NULL \
                OR json_valid(properties) = 0 \
                OR (COALESCE(json_extract(properties, '$.type_only'), 0) != 1 \
                    AND COALESCE(json_extract(properties, '$.scope'), 'module') = 'module')))) \
         ORDER BY kind, from_id, to_id, confidence, \
              COALESCE(source_byte_start, -1), COALESCE(source_byte_end, -1) \
         LIMIT ?1",
    )?;
    let mut rows = stmt.query(rusqlite::params![cap])?;
    let mut adjacency: HashMap<String, Vec<String>> = HashMap::new();
    let mut edge_count = 0usize;
    let mut truncated = false;
    while let Some(row) = rows.next()? {
        if edge_count >= scan_cap {
            truncated = true;
            break;
        }
        edge_count += 1;
        let kind: String = row.get(0)?;
        let from: String = row.get(1)?;
        let to: String = row.get(2)?;
        let confidence: String = row.get(3)?;
        let properties: Option<String> = row.get(4)?;
        let targets = reachability_targets(&kind, &to, &confidence, properties.as_deref());
        adjacency.entry(from).or_default().extend(targets);
    }
    Ok((adjacency, truncated))
}

fn reachability_targets(
    kind: &str,
    to_id: &str,
    confidence: &str,
    properties_json: Option<&str>,
) -> Vec<String> {
    let mut targets = BTreeSet::from([to_id.to_owned()]);
    if kind == "calls" && confidence == "ambiguous" {
        targets.extend(candidate_ids(properties_json));
    }
    targets.into_iter().collect()
}

fn candidate_ids(properties_json: Option<&str>) -> BTreeSet<String> {
    properties_json
        .and_then(|raw| serde_json::from_str::<Value>(raw).ok())
        .and_then(|value| value.get("candidates").and_then(|c| c.as_array()).cloned())
        .into_iter()
        .flatten()
        .filter_map(|value| value.as_str().map(ToOwned::to_owned))
        .collect()
}

/// Forward-reachable closure of `seed` over `adjacency` (iterative BFS/DFS).
fn forward_reachable(
    adjacency: &HashMap<String, Vec<String>>,
    seed: HashSet<String>,
) -> HashSet<String> {
    let mut visited: HashSet<String> = HashSet::new();
    let mut stack: Vec<String> = seed.into_iter().collect();
    while let Some(node) = stack.pop() {
        if !visited.insert(node.clone()) {
            continue;
        }
        if let Some(neighbours) = adjacency.get(&node) {
            for next in neighbours {
                if !visited.contains(next) {
                    stack.push(next.clone());
                }
            }
        }
    }
    visited
}

/// Tarjan strongly-connected components over the adjacency map; returns the
/// components that form a cycle (size > 1, or a single node with a self-edge).
/// Each component's members are sorted for deterministic output, and the
/// components themselves are sorted by first member.
fn strongly_connected_cycles(adjacency: &HashMap<String, Vec<String>>) -> Vec<Vec<String>> {
    let mut index_of: HashMap<&str, usize> = HashMap::new();
    let mut low: HashMap<&str, usize> = HashMap::new();
    let mut on_stack: HashSet<&str> = HashSet::new();
    let mut stack: Vec<&str> = Vec::new();
    let mut next_index = 0usize;
    let mut components: Vec<Vec<String>> = Vec::new();

    // Iterative Tarjan to avoid deep recursion on large graphs.
    let nodes: Vec<&str> = adjacency.keys().map(String::as_str).collect();
    for &start in &nodes {
        if index_of.contains_key(start) {
            continue;
        }
        // Work stack of (node, next-neighbour-index).
        let mut work: Vec<(&str, usize)> = vec![(start, 0)];
        index_of.insert(start, next_index);
        low.insert(start, next_index);
        next_index += 1;
        stack.push(start);
        on_stack.insert(start);

        while let Some(&(node, child_idx)) = work.last() {
            let empty: &[String] = &[];
            let neighbours = adjacency.get(node).map_or(empty, Vec::as_slice);
            if child_idx < neighbours.len() {
                work.last_mut().unwrap().1 += 1;
                let next = neighbours[child_idx].as_str();
                if !index_of.contains_key(next) {
                    index_of.insert(next, next_index);
                    low.insert(next, next_index);
                    next_index += 1;
                    stack.push(next);
                    on_stack.insert(next);
                    work.push((next, 0));
                } else if on_stack.contains(next) {
                    let next_index_val = index_of[next];
                    let entry = low.get_mut(node).unwrap();
                    *entry = (*entry).min(next_index_val);
                }
            } else {
                // Done with node; propagate low-link to parent and maybe close an SCC.
                if low[node] == index_of[node] {
                    let mut component: Vec<String> = Vec::new();
                    while let Some(top) = stack.pop() {
                        on_stack.remove(top);
                        component.push(top.to_owned());
                        if top == node {
                            break;
                        }
                    }
                    let has_self_edge = adjacency
                        .get(node)
                        .is_some_and(|tos| tos.iter().any(|to| to == node));
                    if component.len() > 1 || has_self_edge {
                        component.sort();
                        components.push(component);
                    }
                }
                work.pop();
                if let Some(&(parent, _)) = work.last() {
                    let node_low = low[node];
                    let entry = low.get_mut(parent).unwrap();
                    *entry = (*entry).min(node_low);
                }
            }
        }
    }
    components.sort_by(|a, b| a.first().cmp(&b.first()));
    components
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn dead_code_summary_flags_low_confidence_when_unreachable_share_is_high() {
        // The lacuna shape: 141 candidates of ~391 analysed (36%) — implausibly
        // high, so the roots don't cover the corpus. Confidence must read LOW and
        // recruit the operator, not present 141 as a confident headline.
        let s = dead_code_summary(141, 250, 1, 3, 5, RootsMode::Explicit);
        assert_eq!(s["confidence"], "low");
        assert!(s["advisory"].as_str().unwrap().contains("LOW CONFIDENCE"));
        assert_eq!(s["dead_candidates"], 141);
        assert_eq!(s["reachable"], 250);
        assert_eq!(s["not_analysed"]["withheld_secret"], 5);
        assert_eq!(s["not_analysed"]["plugins_without_roots"], 1);
        // Explicit mode declares verbatim roots, no derived-confidence claim.
        assert_eq!(s["roots_mode"], "explicit");
        assert!(s["roots_confidence"].is_null());

        // A plausible dead share (a few orphans in a well-rooted corpus) reads
        // MODERATE with no advisory — the breakdown still leads, but no alarm.
        let s = dead_code_summary(5, 400, 0, 0, 0, RootsMode::Explicit);
        assert_eq!(s["confidence"], "moderate");
        assert!(s["advisory"].is_null());

        // Degenerate: nothing analysed → no division panic, no false alarm.
        let s = dead_code_summary(0, 0, 0, 0, 0, RootsMode::Explicit);
        assert_eq!(s["confidence"], "moderate");
        assert!(s["advisory"].is_null());

        // Auto mode declares the derived-confidence roots.
        let s = dead_code_summary(5, 400, 0, 0, 0, RootsMode::Auto);
        assert_eq!(s["roots_mode"], "auto");
        assert_eq!(s["roots_confidence"], "derived");
    }

    fn edge_scan_conn() -> rusqlite::Connection {
        let conn = rusqlite::Connection::open_in_memory().expect("open in-memory db");
        conn.execute_batch(
            "CREATE TABLE edges (
                kind TEXT NOT NULL,
                from_id TEXT NOT NULL,
                to_id TEXT NOT NULL,
                properties TEXT,
                source_file_id TEXT,
                source_byte_start INTEGER,
                source_byte_end INTEGER,
                confidence TEXT NOT NULL DEFAULT 'resolved'
            );",
        )
        .expect("create edges table");
        conn
    }

    fn insert_edge(
        conn: &rusqlite::Connection,
        kind: &str,
        from_id: &str,
        to_id: &str,
        properties: Option<&str>,
        source_byte_start: i64,
    ) {
        conn.execute(
            "INSERT INTO edges (
                kind, from_id, to_id, confidence, properties, source_byte_start, source_byte_end
            ) VALUES (?1, ?2, ?3, 'resolved', ?4, ?5, ?6)",
            rusqlite::params![
                kind,
                from_id,
                to_id,
                properties,
                source_byte_start,
                source_byte_start + 1
            ],
        )
        .expect("insert edge");
    }

    fn graph(edges: &[(&str, &str)]) -> HashMap<String, Vec<String>> {
        let mut adjacency: HashMap<String, Vec<String>> = HashMap::new();
        for (from, to) in edges {
            adjacency
                .entry((*from).to_owned())
                .or_default()
                .push((*to).to_owned());
        }
        adjacency
    }

    #[test]
    fn import_cycle_scan_truncates_in_deterministic_order() {
        let conn = edge_scan_conn();
        insert_edge(
            &conn,
            "imports",
            "python:module:z",
            "python:module:a",
            None,
            30,
        );
        insert_edge(
            &conn,
            "imports",
            "python:module:a",
            "python:module:c",
            None,
            20,
        );
        insert_edge(
            &conn,
            "imports",
            "python:module:a",
            "python:module:b",
            None,
            10,
        );

        let (adjacency, truncated) =
            import_adjacency_for_cycles(&conn, EdgeConfidence::Resolved, 2).unwrap();

        assert!(truncated);
        assert_eq!(
            adjacency.get("python:module:a").unwrap(),
            &vec!["python:module:b".to_owned(), "python:module:c".to_owned()]
        );
        assert!(!adjacency.contains_key("python:module:z"));
    }

    #[test]
    fn dead_code_edge_scan_truncates_in_deterministic_order() {
        let conn = edge_scan_conn();
        insert_edge(
            &conn,
            "calls",
            "python:function:z",
            "python:function:a",
            None,
            30,
        );
        insert_edge(
            &conn,
            "calls",
            "python:function:a",
            "python:function:c",
            None,
            20,
        );
        insert_edge(
            &conn,
            "calls",
            "python:function:a",
            "python:function:b",
            None,
            10,
        );

        let (adjacency, truncated) = call_import_adjacency_with_cap(&conn, 2).unwrap();

        assert!(truncated);
        assert_eq!(
            adjacency.get("python:function:a").unwrap(),
            &vec![
                "python:function:b".to_owned(),
                "python:function:c".to_owned()
            ]
        );
        assert!(!adjacency.contains_key("python:function:z"));
    }

    #[test]
    fn detects_a_two_node_cycle() {
        let g = graph(&[("a", "b"), ("b", "a"), ("b", "c")]);
        let cycles = strongly_connected_cycles(&g);
        assert_eq!(cycles, vec![vec!["a".to_owned(), "b".to_owned()]]);
    }

    #[test]
    fn no_cycle_in_a_dag() {
        let g = graph(&[("a", "b"), ("b", "c"), ("a", "c")]);
        assert!(strongly_connected_cycles(&g).is_empty());
    }

    #[test]
    fn detects_a_self_import() {
        let g = graph(&[("a", "a")]);
        assert_eq!(cycles_len(&g), 1);
    }

    #[test]
    fn detects_a_three_node_cycle() {
        let g = graph(&[("a", "b"), ("b", "c"), ("c", "a")]);
        let cycles = strongly_connected_cycles(&g);
        assert_eq!(cycles.len(), 1);
        assert_eq!(cycles[0].len(), 3);
    }

    fn cycles_len(g: &HashMap<String, Vec<String>>) -> usize {
        strongly_connected_cycles(g).len()
    }

    #[test]
    fn confidence_in_clause_is_a_ceiling() {
        assert_eq!(confidence_in_clause(EdgeConfidence::Resolved), "'resolved'");
        assert_eq!(
            confidence_in_clause(EdgeConfidence::Ambiguous),
            "'resolved', 'ambiguous'"
        );
        assert_eq!(
            confidence_in_clause(EdgeConfidence::Inferred),
            "'resolved', 'ambiguous', 'inferred'"
        );
    }
}
