//! MCP protocol surface for Loomweave.

mod analyze_runs;
mod catalogue;
pub mod config;
pub mod filigree;
pub mod filigree_url;
mod index_diff;
pub mod scan_results;
pub mod snapshot;
mod tools;
pub mod wardline_reconcile;

use std::collections::{BTreeSet, HashMap};
use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};

use loomweave_core::{
    EdgeConfidence, EmbeddingProvider, LlmProvider, LlmProviderError, LlmRequest, LlmResponse,
    McpErrorCode,
};
use rusqlite::{Connection, OpenFlags, OptionalExtension};
use serde::ser::SerializeStruct;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use thiserror::Error;
use time::{Date, Month, OffsetDateTime, macros::format_description};
use tokio::sync::{Mutex as AsyncMutex, Notify, broadcast, mpsc};

use loomweave_core::plugin::{ContentLengthCeiling, Frame, TransportError};
use loomweave_federation::config::{
    LlmConfigPatch, LlmProviderKind, McpConfig, ProviderSelection, SemanticConfigPatch,
    SemanticProviderKind, select_provider_with_env, update_llm_config_file,
    update_semantic_config_file,
};
use loomweave_storage::{
    CallEdgeMatch, EntityRow, InferredCallEdgeRecord, InferredEdgeCacheEntry, InferredEdgeCacheKey,
    InferredEdgeWriteStats, ReaderPool, ReferenceDirection, ReferenceEdgeMatch,
    RolledUpReferenceEdge, StorageError, SummaryCacheEntry, SummaryCacheKey, UnresolvedCallSiteRow,
    WriterCmd, call_edges_from, call_edges_targeting, containing_module_id,
    entity_briefing_block_reason, entity_by_id, import_edges_for_entity,
    inferred_edge_cache_key_id, module_reference_rollup, reference_edges_for_entity,
    relation_edges_for_entity, resolve_entity_ref, sei_for_locator, tags_for_entity,
    unresolved_call_sites_for_caller, unresolved_caller_count_for_target,
    unresolved_callers_for_target,
};

use crate::config::{LlmConfig, SemanticSearchConfig};
use crate::filigree::{
    EntityAssociation, EntityAssociationsResponse, FiligreeLookup, IssueDetail,
    ObservationCreateRequest,
};
use loomweave_storage::{
    GuidanceProposal, GuidanceSheetInput, embeddings_db_path, invalidate_summaries_for_sheet,
    upsert_guidance_sheet,
};

/// MCP protocol revision supported by the B.6 stdio server.
pub const MCP_PROTOCOL_VERSION: &str = "2025-11-25";
const EMPTY_GUIDANCE_FINGERPRINT: &str = "guidance-empty";

/// The bundled loomweave-workflow skill text, embedded for the `prompts/get`
/// surface and reused as the canonical orientation reference. The asset lives
/// in this crate's own tree; the CLI (which depends on loomweave-mcp) reaches
/// down into it to embed the same bytes for its on-disk `install --skills`
/// copy (clarion-04391392c7).
pub const LOOMWEAVE_WORKFLOW_SKILL: &str =
    include_str!("../assets/skills/loomweave-workflow/SKILL.md");

/// Orientation text returned in the MCP `initialize` result's `instructions`
/// field. The `Tools:` enumeration is derived from [`list_tools_for_policy`]
/// under the *active* policy (the single source of truth) so it can never
/// advertise a tool the server will not actually register — the
/// agent-first-feedback §2.5 bug, where the write tools were listed but absent
/// from `tools/list` unless `serve.mcp.enable_write_tools: true`. Real clients
/// truncate this blob (~2 KB observed), so the write-gate note comes FIRST and
/// the tool enumeration — recoverable from `tools/list` — comes LAST
/// (clarion-e65354898d). Kept consistent with the loomweave-workflow skill.
fn server_instructions(policy: McpToolPolicy) -> String {
    let tool_names = list_tools_for_policy(policy)
        .iter()
        .map(|tool| tool.name)
        .collect::<Vec<_>>()
        .join(", ");
    let write_tools_note = if policy.enable_write_tools {
        String::new()
    } else {
        // The bootstrap exemption is deliberate (PM ruling, weft-ac59e8e730):
        // state it loudly so a read-only session knows these two tools can
        // persistently enable writes and live LLM spend.
        "\n\nWrite-gated unless `serve.mcp.enable_write_tools: true`: \
`entity_summary_get`, `analyze_start`, `analyze_cancel`, `propose_guidance`, \
`promote_guidance`. EXEMPT by design: `llm_config_set` and \
`semantic_config_set` bypass this gate — from a read-only session they \
persistently edit loomweave.yaml and can enable write tools and live (paid) \
LLM spend; reconnect after changes."
            .to_owned()
    };
    format!(
        "Loomweave is a code-archaeology server: this project is pre-extracted into \
a queryable map of entities (functions, classes, modules, files), the call / \
reference / import edges between them, relation edges (inherits_from / decorates \
/ implements / derives), and subsystem clusters. Ask Loomweave instead of \
re-reading or grepping the tree.{write_tools_note}

Entity IDs are `{{plugin}}:{{kind}}:{{qualified_name}}`; subsystems are \
`core:subsystem:{{hash}}`. Get an id from `entity_find` or `entity_at` and copy \
it verbatim; `entity_resolve` maps pasted qualnames, Rust `::` paths, and SEI \
tokens to identity rows — never hand-construct an id.

Deep dive: the `loomweave-workflow` skill/prompt. Live counts: the \
`loomweave://context` resource.

Tools: {tool_names}."
    )
}

type InferredInflight =
    Arc<AsyncMutex<HashMap<InferredEdgeCacheKey, broadcast::Sender<InferredDispatchOutcome>>>>;

pub const RENAME_MAP: &[(&str, &str)] = &[
    ("entity_at", "entity_at"),
    ("find_entity", "entity_find"),
    ("callers_of", "entity_callers_list"),
    ("execution_paths_from", "entity_execution_path_list"),
    ("summary", "entity_summary_get"),
    ("issues_for", "entity_issue_list"),
    ("neighborhood", "entity_neighborhood_get"),
    ("subsystem_members", "subsystem_member_list"),
    ("subsystem_of", "entity_subsystem_get"),
    ("project_status", "project_status_get"),
    ("summary_preview_cost", "entity_summary_preview_cost_get"),
    ("source_for_entity", "entity_source_get"),
    ("call_sites", "entity_call_site_list"),
    ("orientation_pack", "entity_orientation_pack_get"),
    ("analyze_start", "analyze_start"),
    ("analyze_status", "analyze_status_get"),
    ("analyze_cancel", "analyze_cancel"),
    ("index_diff", "index_diff_get"),
    ("guidance_for", "entity_guidance_list"),
    ("propose_guidance", "propose_guidance"),
    ("promote_guidance", "promote_guidance"),
    ("findings_for", "entity_finding_list"),
    ("wardline_for", "entity_wardline_get"),
    ("find_by_tag", "entity_tag_list"),
    ("find_by_kind", "entity_kind_list"),
    ("find_by_wardline", "entity_wardline_list"),
    ("find_circular_imports", "module_circular_import_list"),
    ("find_coupling_hotspots", "entity_coupling_hotspot_list"),
    ("find_entry_points", "entity_entry_point_list"),
    ("find_http_routes", "entity_http_route_list"),
    ("find_data_models", "entity_data_model_list"),
    ("find_tests", "entity_test_list"),
    ("find_deprecations", "entity_deprecation_list"),
    ("find_todos", "entity_todo_list"),
    ("what_tests_this", "entity_test_caller_list"),
    ("high_churn", "entity_high_churn_list"),
    ("recently_changed", "entity_recent_change_list"),
    ("find_dead_code", "entity_dead_list"),
    ("search_semantic", "entity_semantic_search_list"),
];

pub fn rename_old_to_new(name: &str) -> &str {
    for &(old, new) in RENAME_MAP {
        if name == old {
            return new;
        }
    }
    name
}

#[allow(clippy::struct_excessive_bools)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct ToolMetadata {
    // On the wire, false flags are omitted (absent ⇒ false): the flags exist
    // for client permissioning, and serializing five booleans on every tool
    // taxed every session's tools/list (clarion-e65354898d).
    #[serde(skip_serializing_if = "is_false")]
    pub read_only: bool,
    #[serde(skip_serializing_if = "is_false")]
    pub writes_local_state: bool,
    #[serde(skip_serializing_if = "is_false")]
    pub writes_external_state: bool,
    #[serde(skip_serializing_if = "is_false")]
    pub spawns_process: bool,
    #[serde(skip_serializing_if = "is_false")]
    pub may_call_llm: bool,
}

#[allow(clippy::trivially_copy_pass_by_ref)] // signature dictated by serde
fn is_false(flag: &bool) -> bool {
    !*flag
}

impl ToolMetadata {
    const fn read_only() -> Self {
        Self {
            read_only: true,
            writes_local_state: false,
            writes_external_state: false,
            spawns_process: false,
            may_call_llm: false,
        }
    }

    #[allow(clippy::fn_params_excessive_bools)]
    const fn write_tool(
        writes_local_state: bool,
        writes_external_state: bool,
        spawns_process: bool,
        may_call_llm: bool,
    ) -> Self {
        Self {
            read_only: false,
            writes_local_state,
            writes_external_state,
            spawns_process,
            may_call_llm,
        }
    }

    const fn conditional_llm() -> Self {
        Self {
            read_only: true,
            writes_local_state: false,
            writes_external_state: false,
            spawns_process: false,
            may_call_llm: true,
        }
    }
}

pub fn tool_metadata(name: &str) -> ToolMetadata {
    match name {
        // The config-set pair shares the local-write flag shape with
        // analyze_cancel/promote_guidance, but is gate-EXEMPT — see
        // `is_bootstrap_config_tool` (weft-ac59e8e730).
        "llm_config_set" | "semantic_config_set" | "analyze_cancel" | "promote_guidance" => {
            ToolMetadata::write_tool(true, false, false, false)
        }
        "entity_summary_get" => ToolMetadata::write_tool(true, false, false, true),
        "entity_callers_list" | "entity_neighborhood_get" | "entity_execution_path_list" => {
            ToolMetadata::conditional_llm()
        }
        "analyze_start" => ToolMetadata::write_tool(true, false, true, false),
        "propose_guidance" => ToolMetadata::write_tool(false, true, false, false),
        _ => ToolMetadata::read_only(),
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct McpToolPolicy {
    pub enable_write_tools: bool,
}

impl McpToolPolicy {
    #[must_use]
    pub const fn read_only() -> Self {
        Self {
            enable_write_tools: false,
        }
    }

    #[must_use]
    pub const fn allow_write_tools() -> Self {
        Self {
            enable_write_tools: true,
        }
    }

    #[must_use]
    pub fn allows(self, name: &str) -> bool {
        self.enable_write_tools || tool_metadata(name).read_only || is_bootstrap_config_tool(name)
    }

    fn allows_arguments(self, name: &str, arguments: &serde_json::Map<String, Value>) -> bool {
        self.enable_write_tools || !tool_uses_conditional_inferred_dispatch(name, arguments)
    }
}

fn is_bootstrap_config_tool(name: &str) -> bool {
    matches!(name, "llm_config_set" | "semantic_config_set")
}

fn tool_uses_conditional_inferred_dispatch(
    name: &str,
    arguments: &serde_json::Map<String, Value>,
) -> bool {
    if !matches!(
        name,
        "entity_callers_list" | "entity_neighborhood_get" | "entity_execution_path_list"
    ) {
        return false;
    }
    matches!(
        arguments.get("confidence").and_then(Value::as_str),
        Some("inferred" | "all")
    )
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ToolDefinition {
    pub name: &'static str,
    pub description: &'static str,
    pub input_schema: Value,
}

impl Serialize for ToolDefinition {
    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        // The permissioning flags ride ONLY in the grouped `metadata` object.
        // They used to also be flattened at the top level — ~140 duplicated
        // bytes per tool in every session's tools/list (clarion-e65354898d).
        let metadata = tool_metadata(self.name);
        let mut state = serializer.serialize_struct("ToolDefinition", 4)?;
        state.serialize_field("name", self.name)?;
        state.serialize_field("description", self.description)?;
        state.serialize_field("inputSchema", &self.input_schema)?;
        state.serialize_field("metadata", &metadata)?;
        state.end()
    }
}

#[must_use]
// A flat registry of tool definitions; length tracks the tool count by design.
#[allow(clippy::too_many_lines)]
pub fn list_tools() -> Vec<ToolDefinition> {
    vec![
        ToolDefinition {
            name: "entity_at",
            description: "Innermost entity whose source range contains `file`+`line`, with an `entity_context` evidence block (match_reason, containing stack, sub-ranges, alternatives). A line only a module spans reports containing_range — never a fabricated exact match.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "file": {"type": "string"},
                    "line": {"type": "integer", "minimum": 1}
                },
                "required": ["file", "line"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_find",
            description: "Search entities by name, id, summary, and docstring before grepping. Optional `kind`; paginated; scanner-blocked docstrings never match.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "pattern": {"type": "string", "minLength": 1},
                    "limit": {"type": "integer", "minimum": 1, "maximum": 100},
                    "cursor": {"type": ["string", "null"]},
                    "kind": {"type": "string", "minLength": 1}
                },
                "required": ["pattern"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_callers_list",
            description: "List callers. `confidence` resolved (default) | ambiguous | inferred. Bounded. scope_excludes is non-empty ONLY when a candidate was skipped; an empty scope_excludes with traversal_complete:true means every candidate was searched. Skipped sites surface as unresolved_candidates.",
            input_schema: id_confidence_cursor_schema(),
        },
        ToolDefinition {
            name: "entity_execution_path_list",
            description: "Bounded calls-only paths from an entity (`max_depth` default 3). Compact: deduplicated `nodes` table + `paths` as node-id arrays, longest-first; truncation reports edge-cap/path-cap. `scope_excludes` names unsearched blind spots (attribute-receiver / unresolved static calls) — empties are not proof.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "max_depth": {"type": "integer", "minimum": 1, "maximum": 8},
                    "confidence": confidence_schema()
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_summary_get",
            description: "On-demand cached LLM summary for one entity (leaf scope: module summaries do not aggregate members). Non-JSON LLM output degrades to a cached deterministic structural fallback, so a retry is a cache hit, not a re-billed failure.",
            input_schema: id_schema(),
        },
        ToolDefinition {
            name: "entity_issue_list",
            description: "Filigree issues attached to an entity (optionally contained entities too), plus an enrich-only `wardline_findings` section. result_kind matched | no_matches | unavailable distinguishes empty-Filigree from unreachable-Filigree; `filigree_endpoint` shows which endpoint answered.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "include_contained": {"type": "boolean"}
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_neighborhood_get",
            description: "One-hop callers/callees/container/contained/references/imports/relations. Per-bucket limit. scope_excludes is non-empty only when a caller candidate was skipped; empty + traversal_complete:true confirms the callers bucket searched all candidates. Skipped sites surface as unresolved_candidates.",
            input_schema: id_confidence_schema(),
        },
        ToolDefinition {
            name: "subsystem_member_list",
            description: "List the module entities in a subsystem. Bounded: `limit` (default 50, max 100) + numeric-offset `cursor`.",
            input_schema: id_cursor_schema(),
        },
        ToolDefinition {
            name: "entity_subsystem_get",
            description: "The subsystem an entity belongs to (reverse of subsystem_member_list). Any entity id (functions/classes resolve through their containing module); reports the via-module.",
            input_schema: id_schema(),
        },
        ToolDefinition {
            name: "project_status_get",
            description: "Deterministic diagnostics: repo root, db path, latest run, entity/edge/finding counts, index staleness, per-plugin counts, LLM policy, and the resolved Filigree endpoint.",
            input_schema: json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "llm_config_get",
            description: "Read LLM/provider/live and MCP write-tool config.",
            input_schema: json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "llm_config_set",
            description: "Update LLM/provider/live/write-tool config in loomweave.yaml; reconnect.",
            input_schema: llm_config_set_schema(),
        },
        ToolDefinition {
            name: "semantic_config_get",
            description: "Read semantic-search config and sidecar diagnostics.",
            input_schema: json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "semantic_config_set",
            description: "Update semantic-search config in loomweave.yaml; rerun analyze.",
            input_schema: semantic_config_set_schema(),
        },
        ToolDefinition {
            name: "entity_summary_preview_cost_get",
            description: "Preview what entity_summary_get would cost BEFORE spending: cache_status (hit | expired | miss), real cost on a hit, an estimate on a miss, and live_spend_would_occur. A disabled LLM is reported distinctly from a cache miss. Never invokes the provider.",
            input_schema: id_schema(),
        },
        ToolDefinition {
            name: "entity_source_get",
            description: "An entity's exact indexed source span plus `context_lines` (default 10) of context, line-numbered and flagged in_entity. source_status reports ok | missing | no_range | no_source_path | binary | drifted (rerun analyze) instead of a misleading stale snippet.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "context_lines": {"type": "integer", "minimum": 0, "maximum": 200}
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_call_site_list",
            description: "Show the source sites behind calls/references edges — the evidence for WHY an edge exists. role=caller (default) lists outgoing sites; role=callee lists incoming, INCLUDING `unresolved_sites` (static calls Loomweave could not bind — the recovery surface for suspicious empty caller lists). Bounded.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "role": {"type": "string", "enum": ["caller", "callee"], "default": "caller"},
                    "kind": {"type": "string", "enum": ["calls", "references"]},
                    "confidence": confidence_schema(),
                    "path": {"type": "string", "enum": ["all", "production", "test"], "default": "all"}
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_orientation_pack_get",
            description: "One deterministic orientation packet for an `entity` id OR `file`+`line` (exactly one): primary entity, match evidence, one-hop neighbors, execution paths, Filigree issues, Wardline findings, health, suggested next reads. Optional `include` (wardline/findings/issues) folds a `dossier`. Bounded; `omitted` + degraded sections keep empties honest.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "entity": {"type": "string", "minLength": 1},
                    "file": {"type": "string", "minLength": 1},
                    "line": {"type": "integer", "minimum": 1},
                    "include": {
                        "type": "array",
                        "items": {"type": "string", "enum": ["wardline", "findings", "issues"]},
                        "uniqueItems": true
                    }
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "analyze_start",
            description: "Start a background `loomweave analyze` run; returns run_id immediately (runs can take minutes). One run per project (cross-process lock). Poll analyze_status_get; stop via analyze_cancel. No arguments.",
            input_schema: json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "analyze_status_get",
            description: "A run's live status: queued | running | completed | failed | cancelled | skipped_no_plugins, with phase, current plugin/file, processed/total files, heartbeat, and progress_observed=false when the heartbeat is stale (run may be wedged).",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "run_id": {"type": "string", "minLength": 1}
                },
                "required": ["run_id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "analyze_cancel",
            description: "Cancel a running analyze: SIGKILLs the run's process group (plugin + pyright) and marks the run cancelled — never left dangling as running. Idempotent. Partial work already written is kept.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "run_id": {"type": "string", "minLength": 1}
                },
                "required": ["run_id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "index_diff_get",
            description: "Drift since the last analyze: analyzed commit vs git HEAD, indexed files modified/missing, dirty working-tree files touching indexed paths, plugin skip/drop counters. overall: fresh | drift | unknown | never_analyzed. Entity-level diff is unavailable in v0.1.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "limit": {"type": "integer", "minimum": 1, "maximum": 2000}
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_guidance_list",
            description: "Guidance sheets applicable to an entity, scope-ranked (project → … → function). wardline_group rules are reported in `notes`, never guessed. Bounded; honest-empty when no sheet applies.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "limit": {"type": "integer", "minimum": 1, "maximum": 200},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "propose_guidance",
            description: "Propose a guidance sheet by creating a Filigree observation. Deliberately inert: nothing enters summaries until an operator promotes it (promote_guidance or the CLI).",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "entity_id": {"type": "string", "minLength": 1},
                    "content": {"type": "string", "minLength": 1},
                    "scope_level": {
                        "type": "string",
                        "enum": ["project", "subsystem", "package", "module", "class", "function"],
                        "default": "function"
                    },
                    "match_rules": {"type": "array", "items": {"type": "object"}},
                    "name": {"type": "string", "minLength": 1},
                    "pinned": {"type": "boolean", "default": false},
                    "expires": {"type": "string", "minLength": 1}
                },
                "required": ["entity_id", "content"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "promote_guidance",
            description: "Promote a reviewed Filigree observation from propose_guidance into a local guidance sheet. This operator action is the anti-poisoning boundary.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "observation_id": {"type": "string", "minLength": 1}
                },
                "required": ["observation_id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_finding_list",
            description: "Findings anchored to one entity; filter by kind / severity / status (closed value sets — see the skill). Bounded (limit/offset, page block). No findings → empty list, not an error.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "filter": {
                        "type": "object",
                        "properties": {
                            "kind": {"type": "string"},
                            "severity": {"type": "string"},
                            "status": {"type": "string"}
                        },
                        "additionalProperties": false
                    },
                    "limit": {"type": "integer", "minimum": 1, "maximum": 200},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_wardline_get",
            description: "The entity's Wardline taint metadata VERBATIM (the blob is opaque to Loomweave). result_kind present | no_facts with a missing-signal note — facts arrive via Filigree Flow-B, so a locally-empty result is honest.",
            input_schema: id_schema(),
        },
        ToolDefinition {
            name: "entity_tag_list",
            description: "Entities carrying a categorisation `tag`, optional `scope` (entity id → descendants, or path glob). Bounded; honest-empty when no entity carries the tag.",
            input_schema: scope_facet_schema(&[("tag", true)]),
        },
        ToolDefinition {
            name: "entity_kind_list",
            description: "Entities of a `kind` (function/class/module/…), optional `scope`. Bounded. An unknown kind matches no rows.",
            input_schema: scope_facet_schema(&[("kind", true)]),
        },
        ToolDefinition {
            name: "entity_wardline_list",
            description: "Entities carrying Wardline taint facts, filtered by `tier`/`group` (best-effort against the opaque blob); `has_findings: true` filters to those also carrying findings. Bounded.",
            input_schema: wardline_facet_schema(),
        },
        ToolDefinition {
            name: "module_circular_import_list",
            description: "Import cycles (SCCs over `imports` edges, plus self-imports), members sorted. Edge-derived: `confidence` is a ceiling (default resolved). Optional `scope`. Bounded.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "scope": {"type": "string", "minLength": 1},
                    "confidence": confidence_schema(),
                    "limit": {"type": "integer", "minimum": 1, "maximum": 200},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_coupling_hotspot_list",
            description: "Entities ranked by coupling (distinct fan-in + fan-out), most-coupled first. `confidence` is a ceiling (default resolved). `app_only:true` drops test-tagged/non-first-party. Optional `scope`. Bounded (limit default 20).",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "scope": {"type": "string", "minLength": 1},
                    "confidence": confidence_schema(),
                    "app_only": {"type": "boolean"},
                    "limit": {"type": "integer", "minimum": 1, "maximum": 200},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_entry_point_list",
            description: "Entities tagged `entry-point`, optional `scope`. HONEST-EMPTY when the tag is not emitted — absence of signal, not absence of entry points. Bounded.",
            input_schema: scope_page_schema(false),
        },
        ToolDefinition {
            name: "entity_http_route_list",
            description: "Entities tagged `http-route`, optional `scope`. Honest-empty when route categorisation is not emitted. Bounded.",
            input_schema: scope_page_schema(false),
        },
        ToolDefinition {
            name: "entity_data_model_list",
            description: "Entities tagged `data-model`, optional `scope`. Honest-empty when data-model categorisation is not emitted. Bounded.",
            input_schema: scope_page_schema(false),
        },
        ToolDefinition {
            name: "entity_test_list",
            description: "Entities tagged `test`, optional `scope`. Honest-empty when test categorisation is not emitted. Bounded.",
            input_schema: scope_page_schema(false),
        },
        ToolDefinition {
            name: "entity_deprecation_list",
            description: "Entities tagged `deprecated`, optional `scope`. Honest-empty when deprecation categorisation is not emitted. Bounded.",
            input_schema: scope_page_schema(false),
        },
        ToolDefinition {
            name: "entity_todo_list",
            description: "Entities carrying a TODO/FIXME marker, optional `scope`. Honest-empty when TODO extraction is not emitted. Bounded.",
            input_schema: scope_page_schema(false),
        },
        ToolDefinition {
            name: "entity_test_caller_list",
            description: "The test-tagged callers of an entity. Honest-empty when test categorisation is not emitted — NOT a guarantee the entity is untested. Bounded.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "limit": {"type": "integer", "minimum": 1, "maximum": 200},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_high_churn_list",
            description: "Entities ranked by git churn, optional `scope`. v1.0 does not populate churn, so this is honest-empty in practice. Bounded.",
            input_schema: scope_page_schema(false),
        },
        ToolDefinition {
            name: "entity_recent_change_list",
            description: "Entities changed since `since`, optional `scope`. v1.0 indexes no per-entity change time, so this is an honest no-op with a note pointing at index_diff_get. Never fabricates.",
            input_schema: scope_page_schema(true),
        },
        ToolDefinition {
            name: "entity_dead_list",
            description: "Entities unreachable from roots, optional scope. Conservative heuristic; roots depend on plugin-emitted tags. `roots:\"auto\"` derives roots from tags (derived confidence); `app_only:true` drops test-tagged/non-first-party.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "scope": {"type": "string", "minLength": 1},
                    "roots": {"type": "string", "enum": ["explicit", "auto"]},
                    "app_only": {"type": "boolean"},
                    "limit": {"type": "integer", "minimum": 1, "maximum": 200},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_semantic_search_list",
            description: "Rank entities by embedding similarity to `query`, optional `scope`; disabled returns not_enabled and stale vectors are hidden.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "query": {"type": "string", "minLength": 1},
                    "scope": {"type": "string", "minLength": 1},
                    "limit": {"type": "integer", "minimum": 1, "maximum": 100},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "required": ["query"],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "project_finding_list",
            description: "List findings across the WHOLE project — no entity id needed. Each row carries its anchoring entity {id, sei, file, line} plus tool/rule/kind/severity/status. Same filters as entity_finding_list. Bounded.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "filter": {
                        "type": "object",
                        "properties": {
                            "kind": {"type": "string"},
                            "severity": {"type": "string"},
                            "status": {"type": "string"}
                        },
                        "additionalProperties": false
                    },
                    "limit": {"type": "integer", "minimum": 1, "maximum": 200},
                    "offset": {"type": "integer", "minimum": 0}
                },
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_resolve",
            description: "Batch-resolve pasted dotted qualnames, Rust `::` paths, or SEIs to entity candidates; reports resolved/unresolved/ambiguous.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "qualnames": {
                        "type": "array",
                        "items": {"type": "string", "minLength": 1},
                        "minItems": 1,
                        "maxItems": 2000
                    },
                    "identifiers": {
                        "type": "array",
                        "items": {"type": "string", "minLength": 1},
                        "minItems": 1,
                        "maxItems": 2000
                    },
                    "kind": {"type": "string", "minLength": 1},
                    "plugin": {"type": "string", "minLength": 1}
                },
                "anyOf": [{"required": ["qualnames"]}, {"required": ["identifiers"]}],
                "additionalProperties": false
            }),
        },
        ToolDefinition {
            name: "entity_relation_list",
            description: "List declared relation edges (inherits_from/decorates/implements/derives). Use direction=in for subclasses/decorated funcs/implementors.",
            input_schema: json!({
                "type": "object",
                "properties": {
                    "id": {"type": "string", "minLength": 1},
                    "direction": {"type": "string", "enum": ["in", "out", "both"]},
                    "kind": {
                        "type": "string",
                        "enum": ["inherits_from", "decorates", "implements", "derives"]
                    },
                    "confidence": confidence_schema(),
                    "limit": {"type": "integer", "minimum": 1, "maximum": 100},
                    "cursor": {"type": ["string", "null"]}
                },
                "required": ["id"],
                "additionalProperties": false
            }),
        },
    ]
}

#[must_use]
pub fn list_tools_for_policy(policy: McpToolPolicy) -> Vec<ToolDefinition> {
    list_tools()
        .into_iter()
        .filter(|tool| policy.allows(tool.name))
        .collect()
}

/// Input schema for the scope-aware shortcut tools: optional `scope` (entity id
/// or path glob) plus `limit`/`offset` bounds, and — when `with_since` —
/// an optional ISO-8601 `since`.
fn scope_page_schema(with_since: bool) -> Value {
    let mut properties = serde_json::Map::new();
    properties.insert(
        "scope".to_owned(),
        json!({"type": "string", "minLength": 1}),
    );
    properties.insert(
        "limit".to_owned(),
        json!({"type": "integer", "minimum": 1, "maximum": 200}),
    );
    properties.insert(
        "offset".to_owned(),
        json!({"type": "integer", "minimum": 0}),
    );
    if with_since {
        properties.insert(
            "since".to_owned(),
            json!({"type": "string", "minLength": 1}),
        );
    }
    json!({
        "type": "object",
        "properties": Value::Object(properties),
        "additionalProperties": false
    })
}

/// Input schema for a faceted-search tool: the named facet fields (each
/// `required` or not) plus the shared `scope`/`limit`/`offset` bounds.
fn scope_facet_schema(facets: &[(&str, bool)]) -> Value {
    let mut properties = serde_json::Map::new();
    let mut required = Vec::new();
    for (name, is_required) in facets {
        // Wardline `tier`/`group` filter against the opaque blob and accept a
        // string or number (matching the handler's `optional_facet`); other
        // facets (e.g. `tag`) are strings.
        let schema = if *name == "tier" || *name == "group" {
            json!({"type": ["string", "integer"]})
        } else {
            json!({"type": "string", "minLength": 1})
        };
        properties.insert((*name).to_owned(), schema);
        if *is_required {
            required.push(Value::String((*name).to_owned()));
        }
    }
    properties.insert(
        "scope".to_owned(),
        json!({"type": "string", "minLength": 1}),
    );
    properties.insert(
        "limit".to_owned(),
        json!({"type": "integer", "minimum": 1, "maximum": 200}),
    );
    properties.insert(
        "offset".to_owned(),
        json!({"type": "integer", "minimum": 0}),
    );
    json!({
        "type": "object",
        "properties": Value::Object(properties),
        "required": required,
        "additionalProperties": false
    })
}

/// Input schema for `entity_wardline_list`: the faceted tier/group schema plus a
/// `has_findings` boolean. Declared explicitly because the base schema sets
/// `additionalProperties: false`, which would otherwise reject the param.
fn wardline_facet_schema() -> Value {
    let mut schema = scope_facet_schema(&[("tier", false), ("group", false)]);
    if let Some(properties) = schema.get_mut("properties").and_then(Value::as_object_mut) {
        properties.insert("has_findings".to_owned(), json!({"type": "boolean"}));
    }
    schema
}

fn confidence_schema() -> Value {
    json!({
        "type": "string",
        "enum": ["resolved", "ambiguous", "inferred"],
        "default": "resolved"
    })
}

fn id_schema() -> Value {
    json!({
        "type": "object",
        "properties": {
            "id": {"type": "string", "minLength": 1}
        },
        "required": ["id"],
        "additionalProperties": false
    })
}

/// `id` + `limit` + `cursor` — the single-relation bounded shape (id-only tools
/// with no confidence param, e.g. `subsystem_member_list`). `cursor` is a
/// numeric-offset string; null/absent starts at 0 (clarion-d76e7f7267).
fn id_cursor_schema() -> Value {
    json!({
        "type": "object",
        "properties": {
            "id": {"type": "string", "minLength": 1},
            "limit": {"type": "integer", "minimum": 1, "maximum": 100},
            "cursor": {"type": ["string", "null"]}
        },
        "required": ["id"],
        "additionalProperties": false
    })
}

fn llm_config_set_schema() -> Value {
    // Typed to match the argument parser (`llm_config_patch_from_arguments`):
    // every property is optional, but a present one must carry the right type
    // — the bare `{}` schemas shipped in c475e90 declared nothing
    // (weft-ac59e8e730). Enum values mirror `LlmProviderKind::parse`. No
    // per-property descriptions: tools/list is a per-session context tax
    // (clarion-e65354898d) and depth belongs in the loomweave-workflow skill.
    json!({
        "type": "object",
        "properties": {
            "enabled": {"type": "boolean"},
            "provider": {
                "type": "string",
                "enum": [
                    "openrouter", "open_router", "openrouter_api",
                    "codex_cli", "codex", "codex_sidecar",
                    "claude_cli", "claude_code", "claude_sidecar",
                    "recording"
                ]
            },
            "allow_live_provider": {"type": "boolean"},
            "enable_write_tools": {"type": "boolean"},
            "model_id": {"type": "string", "minLength": 1},
            "codex_model": {"type": "string", "minLength": 1},
            "claude_model": {"type": "string", "minLength": 1},
            "openrouter_api_key_env": {"type": "string", "minLength": 1},
            "openrouter_endpoint_url": {"type": "string", "minLength": 1}
        },
        "additionalProperties": false
    })
}

fn semantic_config_set_schema() -> Value {
    // Typed to match `semantic_config_patch_from_arguments`; enum values
    // mirror `SemanticProviderKind::parse` (weft-ac59e8e730).
    json!({
        "type": "object",
        "properties": {
            "enabled": {"type": "boolean"},
            "provider": {
                "type": "string",
                "enum": ["api", "openai", "openai_api", "local_openai", "local", "openai_local"]
            },
            "allow_live_provider": {"type": "boolean"},
            "model_id": {"type": "string", "minLength": 1},
            "dimensions": {"type": "integer", "minimum": 1},
            "endpoint_url": {"type": "string", "minLength": 1},
            "api_key_env": {"type": "string", "minLength": 1},
            "timeout_seconds": {"type": "integer", "minimum": 1},
            "session_token_ceiling": {"type": "integer", "minimum": 0}
        },
        "additionalProperties": false
    })
}

fn id_confidence_schema() -> Value {
    json!({
        "type": "object",
        "properties": {
            "id": {"type": "string", "minLength": 1},
            "confidence": confidence_schema(),
            // For `entity_callers_list` this is a page size; for
            // `entity_neighborhood_get` it is a PER-BUCKET cap (no cursor — see
            // `id_confidence_cursor_schema` for the cursor-paginated single-
            // relation variant). clarion-d76e7f7267.
            "limit": {"type": "integer", "minimum": 1, "maximum": 100}
        },
        "required": ["id"],
        "additionalProperties": false
    })
}

/// `id` + `confidence` + `limit` + `cursor` — the single-relation cursor-
/// paginated shape for `entity_callers_list`. Distinct from the neighborhood
/// overview, which takes a per-bucket `limit` but NO cursor (one cursor cannot
/// coherently advance nine heterogeneous buckets). clarion-d76e7f7267.
fn id_confidence_cursor_schema() -> Value {
    json!({
        "type": "object",
        "properties": {
            "id": {"type": "string", "minLength": 1},
            "confidence": confidence_schema(),
            "limit": {"type": "integer", "minimum": 1, "maximum": 100},
            "cursor": {"type": ["string", "null"]}
        },
        "required": ["id"],
        "additionalProperties": false
    })
}

/// Handle state-free MCP requests such as `initialize` and `tools/list`.
///
/// Storage-backed tool calls require [`ServerState::handle_json_rpc`]. The
/// `resources/*` and `prompts/*` RPCs are likewise served ONLY by
/// [`ServerState::handle_json_rpc`] (the production `loomweave serve` path); a
/// caller using this free function gets a deliberately narrower server, and its
/// `initialize` advertises only the `tools` capability it actually serves.
#[must_use]
pub fn handle_json_rpc(request: &Value) -> Option<Value> {
    if is_json_rpc_notification(request) {
        return None;
    }
    let id = request.get("id").cloned().unwrap_or(Value::Null);
    let Some(method) = request.get("method").and_then(Value::as_str) else {
        return Some(error_response(&id, -32600, "invalid request"));
    };

    Some(match method {
        "initialize" => result_response(&id, &initialize_result(false, McpToolPolicy::default())),
        "tools/list" => result_response(
            &id,
            &json!({"tools": list_tools_for_policy(McpToolPolicy::default())}),
        ),
        "tools/call" => error_response(
            &id,
            -32601,
            "tools/call requires ServerState::handle_json_rpc",
        ),
        _ => error_response(&id, -32601, "method not found"),
    })
}

/// Actionable chirp for a project with no index. Mirrors the `SessionStart` hook
/// wording (`hook.rs`) so the operator sees the same "install then analyze"
/// sequence whether they read it from the shell or from an MCP client. Surfaced
/// both in the degraded `initialize` instructions and from every degraded
/// `tools/call` result.
fn no_index_message(project_root: &Path) -> String {
    let root = project_root.display();
    format!(
        "Loomweave has no index for this project yet \
({root}/.weft/loomweave/loomweave.db is missing), so the structural graph has not been \
built and every Loomweave tool is unavailable. Run `loomweave install --path {root}` \
then `loomweave analyze {root}` in a terminal to extract the entity / edge graph, \
then reconnect this MCP server."
    )
}

/// Degraded-mode orientation for the `initialize` `instructions` field. Distinct
/// from [`server_instructions`] (the healthy-index orientation) so the normal
/// path — and its `server_instructions_enumerate_every_tool` guard — is
/// untouched.
fn server_instructions_no_index(project_root: &Path) -> String {
    format!(
        "⚠ NO INDEX. {}\n\nNormally Loomweave answers \"what calls X\", \"where is X \
defined\", \"what subsystem is X in\" from a pre-extracted graph instead of grepping \
the tree — but it needs an index first. `tools/list` still shows the surface; any tool \
call returns this same instruction until the index exists.",
        no_index_message(project_root)
    )
}

/// The `initialize` result for the degraded no-index server. Advertises `tools`
/// and `prompts` (the static `loomweave-workflow` prompt works without a DB) but
/// not `resources` (the `loomweave://context` resource needs the index).
fn initialize_result_no_index(project_root: &Path) -> Value {
    json!({
        "protocolVersion": MCP_PROTOCOL_VERSION,
        "capabilities": { "tools": {}, "prompts": {} },
        "serverInfo": {
            "name": "loomweave",
            "version": env!("CARGO_PKG_VERSION")
        },
        "instructions": server_instructions_no_index(project_root)
    })
}

/// JSON-RPC dispatch for the degraded "no index" stdio server: the project has
/// no `.weft/loomweave/loomweave.db`, so there is no graph to query. `initialize`
/// succeeds (the client connects cleanly rather than seeing the server die) and
/// `tools/call` returns the actionable chirp as a tool result with
/// `isError: true` — the load-bearing channel, since not every client surfaces
/// the `initialize` `instructions`. `tools/list` and the static
/// `loomweave-workflow` prompt answer normally so the surface looks healthy.
/// clarion-ac36f51c2b.
#[must_use]
pub fn handle_json_rpc_no_index(request: &Value, project_root: &Path) -> Option<Value> {
    if is_json_rpc_notification(request) {
        return None;
    }
    let id = request.get("id").cloned().unwrap_or(Value::Null);
    let Some(method) = request.get("method").and_then(Value::as_str) else {
        return Some(error_response(&id, -32600, "invalid request"));
    };

    Some(match method {
        "initialize" => result_response(&id, &initialize_result_no_index(project_root)),
        "tools/list" => result_response(
            &id,
            &json!({"tools": list_tools_for_policy(McpToolPolicy::default())}),
        ),
        "tools/call" => result_response(
            &id,
            &json!({
                "content": [
                    { "type": "text", "text": no_index_message(project_root) }
                ],
                "isError": true
            }),
        ),
        "prompts/list" => result_response(&id, &prompts_list()),
        "prompts/get" => prompts_get(&id, request.get("params")),
        _ => error_response(&id, -32601, "method not found"),
    })
}

/// Serve a degraded MCP stdio session for a project with no index. Mirrors
/// [`serve_stdio`] (synchronous — there are no storage-backed async tools to
/// drive) but routes every request through [`handle_json_rpc_no_index`]. Used by
/// `loomweave serve` when `.weft/loomweave/loomweave.db` is absent, so the client
/// connects and is told to run analyze rather than watching the server exit.
pub fn serve_stdio_no_index(
    project_root: &Path,
    reader: &mut impl std::io::BufRead,
    writer: &mut impl std::io::Write,
) -> Result<(), McpError> {
    loop {
        let Some(frame) = read_stdio_frame(reader)? else {
            return Ok(());
        };
        let framing = frame.framing;
        let request: Value = serde_json::from_slice(&frame.body)?;
        if let Some(response) = handle_json_rpc_no_index(&request, project_root) {
            write_stdio_response(writer, &encode_response_frame(&response)?, framing)?;
        }
    }
}

/// Deterministic, non-storage diagnostics threaded in at server construction so
/// `project_status` can report the LLM policy and the resolved Filigree
/// endpoint without re-reading config or re-running URL resolution. Optional:
/// servers built via [`ServerState::new`] (e.g. storage-only tests) omit it and
/// `project_status` reports those blocks as unconfigured.
#[derive(Debug, Clone)]
pub struct DiagnosticsContext {
    pub llm: LlmDiagnostics,
    pub filigree: crate::filigree_url::FiligreeUrlResolution,
}

/// The LLM policy posture, captured at construction. `live` reflects whether a
/// provider is actually wired (vs. merely permitted by config).
#[derive(Debug, Clone)]
pub struct LlmDiagnostics {
    /// Provider label, e.g. `"openrouter"`, `"codex_cli"`, `"recording"`, or
    /// `"disabled"` when no provider is wired.
    pub provider: String,
    /// Whether LLM summaries are enabled at all (`llm.enabled`, configured).
    pub enabled: bool,
    /// A live provider is wired and summaries will dispatch to it (effective).
    pub live: bool,
    /// Whether config permits a live provider at all (`llm.allow_live_provider`,
    /// configured).
    pub allow_live_provider: bool,
    /// Summary-cache freshness horizon in days (`llm.cache_max_age_days`).
    pub cache_max_age_days: u32,
}

#[derive(Clone)]
pub struct ServerState {
    project_root: PathBuf,
    readers: ReaderPool,
    execution_edge_cap: usize,
    execution_path_cap: usize,
    summary_llm: Option<SummaryLlmState>,
    semantic_search: Option<SemanticSearchState>,
    clock: Arc<dyn Fn() -> String + Send + Sync>,
    budget: Arc<Mutex<BudgetLedger>>,
    inferred_inflight: InferredInflight,
    filigree_client: Option<Arc<dyn FiligreeLookup>>,
    diagnostics: Option<DiagnosticsContext>,
    tool_policy: McpToolPolicy,
    /// Supervised `loomweave analyze` runs launched via `analyze_start`.
    analyze_runs: crate::analyze_runs::RunRegistry,
    active_requests: Arc<AsyncMutex<BTreeSet<String>>>,
    cancelled_requests: Arc<AsyncMutex<BTreeSet<String>>>,
    cancellation_notify: Arc<Notify>,
    /// Launcher for `analyze_start` to spawn. `None` → `current_exe()`; tests
    /// inject a stub via [`ServerState::with_analyze_command`].
    analyze_program: Option<PathBuf>,
    /// Config file the active `serve` was launched with, forwarded as
    /// `--config` to an `analyze_start`-spawned analyze so the child parses the
    /// same configuration (review #12). `None` → the child uses its default
    /// config discovery (serve was started without an explicit `--config`).
    analyze_config_path: Option<PathBuf>,
}

impl ServerState {
    #[must_use]
    pub fn new(project_root: PathBuf, readers: ReaderPool) -> Self {
        Self {
            project_root,
            readers,
            execution_edge_cap: 500,
            execution_path_cap: 200,
            summary_llm: None,
            semantic_search: None,
            clock: Arc::new(default_now_string),
            budget: Arc::new(Mutex::new(BudgetLedger::default())),
            inferred_inflight: Arc::new(AsyncMutex::new(HashMap::new())),
            filigree_client: None,
            diagnostics: None,
            tool_policy: McpToolPolicy::default(),
            analyze_runs: Arc::new(Mutex::new(HashMap::new())),
            active_requests: Arc::new(AsyncMutex::new(BTreeSet::new())),
            cancelled_requests: Arc::new(AsyncMutex::new(BTreeSet::new())),
            cancellation_notify: Arc::new(Notify::new()),
            analyze_program: None,
            analyze_config_path: None,
        }
    }

    /// Override the program `analyze_start` launches (default: `current_exe()`).
    /// Tests inject a stub binary so the lifecycle can be exercised without a
    /// full analyze run.
    #[must_use]
    pub fn with_analyze_command(mut self, program: PathBuf) -> Self {
        self.analyze_program = Some(program);
        self
    }

    /// Forward `serve`'s `--config` path to `analyze_start`-spawned analyze runs
    /// so the child parses the same configuration (review #12). Call only when
    /// serve was launched with an explicit, on-disk config file.
    #[must_use]
    pub fn with_analyze_config(mut self, config_path: PathBuf) -> Self {
        self.analyze_config_path = Some(config_path);
        self
    }

    #[must_use]
    pub fn with_tool_policy(mut self, policy: McpToolPolicy) -> Self {
        self.tool_policy = policy;
        self
    }

    /// Test-only: number of analyze run handles currently held in the registry.
    /// Lets a test assert that finished runs are evicted (clarion-7e0c21558a)
    /// rather than accumulating across a long-lived `serve`.
    #[doc(hidden)]
    #[must_use]
    pub fn tracked_analyze_runs(&self) -> usize {
        self.analyze_runs
            .lock()
            .expect("analyze run registry mutex")
            .len()
    }

    #[must_use]
    pub fn with_edge_cap(mut self, execution_edge_cap: usize) -> Self {
        self.execution_edge_cap = execution_edge_cap;
        self
    }

    #[must_use]
    pub fn with_path_cap(mut self, execution_path_cap: usize) -> Self {
        self.execution_path_cap = execution_path_cap;
        self
    }

    #[must_use]
    pub fn with_summary_llm(
        mut self,
        writer: mpsc::Sender<WriterCmd>,
        config: LlmConfig,
        provider: Arc<dyn LlmProvider>,
    ) -> Self {
        self.summary_llm = Some(SummaryLlmState {
            writer,
            config,
            provider,
        });
        self
    }

    /// Attach the embeddings provider + policy for `search_semantic` (`WS5b`).
    /// Absent (or `config.enabled == false`) → the tool degrades honestly to
    /// "not enabled". The provider is constructed by the caller (`serve`) only
    /// when opted in with a key present, so the trait — not the choice — is
    /// load-bearing.
    #[must_use]
    pub fn with_semantic_search(
        mut self,
        config: SemanticSearchConfig,
        provider: Arc<dyn EmbeddingProvider>,
    ) -> Self {
        self.semantic_search = Some(SemanticSearchState { config, provider });
        self
    }

    #[must_use]
    pub fn with_clock(mut self, clock: impl Fn() -> String + Send + Sync + 'static) -> Self {
        self.clock = Arc::new(clock);
        self
    }

    #[must_use]
    pub fn with_filigree_client(mut self, client: Arc<dyn FiligreeLookup>) -> Self {
        self.filigree_client = Some(client);
        self
    }

    #[must_use]
    pub fn with_diagnostics(mut self, diagnostics: DiagnosticsContext) -> Self {
        self.diagnostics = Some(diagnostics);
        self
    }

    async fn tool_llm_config_get(
        &self,
        _arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let path = self.config_file_path();
        let active_write_tools = self.tool_policy.enable_write_tools;
        let read = tokio::task::spawn_blocking(move || read_llm_config_status(&path)).await;
        match read {
            Ok(Ok(status)) => Ok(success_envelope(with_active_session_policy(
                status,
                active_write_tools,
                false,
            ))),
            Ok(Err(err)) => Ok(tool_error_envelope(
                McpErrorCode::StorageError,
                &err.to_string(),
                false,
            )),
            Err(err) => Ok(tool_error_envelope(
                McpErrorCode::Internal,
                &format!("read LLM config task failed: {err}"),
                true,
            )),
        }
    }

    #[allow(clippy::similar_names)] // path/patch are both the precise domain terms
    async fn tool_llm_config_set(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let patch = llm_config_patch_from_arguments(arguments)?;
        if llm_config_patch_is_empty(&patch) {
            return Err(ParamError::new("no LLM config changes requested"));
        }
        let path = self.config_file_path();
        let active_write_tools = self.tool_policy.enable_write_tools;
        let write = tokio::task::spawn_blocking(move || {
            let result = update_llm_config_file(&path, &patch)?;
            Ok::<_, loomweave_federation::config::ConfigError>(llm_config_status_json(
                &path,
                result.created,
                &result.config,
            ))
        })
        .await;
        match write {
            Ok(Ok(status)) => Ok(success_envelope(with_active_session_policy(
                status,
                active_write_tools,
                true,
            ))),
            Ok(Err(err)) => Ok(tool_error_envelope(
                McpErrorCode::StorageError,
                &err.to_string(),
                false,
            )),
            Err(err) => Ok(tool_error_envelope(
                McpErrorCode::Internal,
                &format!("write LLM config task failed: {err}"),
                true,
            )),
        }
    }

    async fn tool_semantic_config_get(
        &self,
        _arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let path = self.config_file_path();
        let project_root = self.project_root.clone();
        let active_write_tools = self.tool_policy.enable_write_tools;
        let read =
            tokio::task::spawn_blocking(move || read_semantic_config_status(&path, &project_root))
                .await;
        match read {
            Ok(Ok(status)) => Ok(success_envelope(with_active_session_policy(
                status,
                active_write_tools,
                false,
            ))),
            Ok(Err(err)) => Ok(tool_error_envelope(
                McpErrorCode::StorageError,
                &err.to_string(),
                false,
            )),
            Err(err) => Ok(tool_error_envelope(
                McpErrorCode::Internal,
                &format!("read semantic config task failed: {err}"),
                true,
            )),
        }
    }

    #[allow(clippy::similar_names)] // path/patch are both the precise domain terms
    async fn tool_semantic_config_set(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let patch = semantic_config_patch_from_arguments(arguments)?;
        if semantic_config_patch_is_empty(&patch) {
            return Err(ParamError::new(
                "no semantic search config changes requested",
            ));
        }
        let path = self.config_file_path();
        let project_root = self.project_root.clone();
        let active_write_tools = self.tool_policy.enable_write_tools;
        let write = tokio::task::spawn_blocking(move || {
            let result = update_semantic_config_file(&path, &patch)?;
            Ok::<_, loomweave_federation::config::ConfigError>(semantic_config_status_json(
                &path,
                &project_root,
                result.created,
                &result.config,
            ))
        })
        .await;
        match write {
            Ok(Ok(status)) => Ok(success_envelope(with_active_session_policy(
                status,
                active_write_tools,
                true,
            ))),
            Ok(Err(err)) => Ok(tool_error_envelope(
                McpErrorCode::StorageError,
                &err.to_string(),
                false,
            )),
            Err(err) => Ok(tool_error_envelope(
                McpErrorCode::Internal,
                &format!("write semantic config task failed: {err}"),
                true,
            )),
        }
    }

    fn config_file_path(&self) -> PathBuf {
        self.analyze_config_path
            .clone()
            .unwrap_or_else(|| self.project_root.join("loomweave.yaml"))
    }

    pub async fn handle_json_rpc(&self, request: &Value) -> Option<Value> {
        if is_json_rpc_notification(request) {
            self.handle_json_rpc_notification(request).await;
            return None;
        }
        let id = request.get("id").cloned().unwrap_or(Value::Null);
        let Some(method) = request.get("method").and_then(Value::as_str) else {
            return Some(error_response(&id, -32600, "invalid request"));
        };
        let request_key = request_id_key(&id);
        if let Some(key) = &request_key {
            self.begin_request(key).await;
        }

        let dispatch = async {
            match method {
                "initialize" => result_response(&id, &initialize_result(true, self.tool_policy)),
                "tools/list" => result_response(
                    &id,
                    &json!({"tools": list_tools_for_policy(self.tool_policy)}),
                ),
                "tools/call" => self.handle_tool_call(&id, request.get("params")).await,
                "resources/list" => result_response(&id, &resources_list()),
                "resources/read" => self.handle_resources_read(&id, request.get("params")).await,
                "prompts/list" => result_response(&id, &prompts_list()),
                "prompts/get" => prompts_get(&id, request.get("params")),
                _ => error_response(&id, -32601, "method not found"),
            }
        };

        let response = if let Some(key) = request_key.clone() {
            tokio::select! {
                response = dispatch => response,
                () = self.wait_for_cancellation(key) => {
                    error_response(&id, -32800, "request cancelled")
                }
            }
        } else {
            dispatch.await
        };

        if let Some(key) = &request_key {
            self.finish_request(key).await;
        }
        Some(response)
    }

    async fn handle_json_rpc_notification(&self, request: &Value) {
        let Some(method) = request.get("method").and_then(Value::as_str) else {
            return;
        };
        if method != "notifications/cancelled" {
            return;
        }
        let Some(request_id) = request
            .get("params")
            .and_then(Value::as_object)
            .and_then(|params| params.get("requestId"))
            .and_then(request_id_key)
        else {
            return;
        };
        self.cancel_request(&request_id).await;
    }

    async fn begin_request(&self, request_id: &str) {
        self.active_requests
            .lock()
            .await
            .insert(request_id.to_owned());
    }

    async fn finish_request(&self, request_id: &str) {
        self.active_requests.lock().await.remove(request_id);
        self.cancelled_requests.lock().await.remove(request_id);
    }

    async fn cancel_request(&self, request_id: &str) {
        if self.active_requests.lock().await.contains(request_id) {
            self.cancelled_requests
                .lock()
                .await
                .insert(request_id.to_owned());
            self.cancellation_notify.notify_waiters();
        }
    }

    async fn wait_for_cancellation(&self, request_id: String) {
        loop {
            if self.cancelled_requests.lock().await.contains(&request_id) {
                return;
            }
            self.cancellation_notify.notified().await;
        }
    }

    // A flat dispatch table over every tool; length tracks the tool count by
    // design (mirrors the `#[allow]` on `list_tools`).
    #[allow(clippy::too_many_lines)]
    async fn handle_tool_call(&self, id: &Value, params: Option<&Value>) -> Value {
        let Some(params) = params.and_then(Value::as_object) else {
            return error_response(id, -32602, "invalid tools/call params");
        };
        let Some(name) = params.get("name").and_then(Value::as_str) else {
            return error_response(id, -32602, "invalid tools/call params: missing name");
        };
        let canonical_name = rename_old_to_new(name);
        let Some(tool) = list_tools()
            .into_iter()
            .find(|tool| tool.name == canonical_name)
        else {
            return error_response(id, -32601, &format!("unknown tool: {name}"));
        };
        if !self.tool_policy.allows(canonical_name) {
            return error_response(
                id,
                -32601,
                &format!("tool disabled by MCP tool policy: {canonical_name}"),
            );
        }
        let arguments = params.get("arguments").unwrap_or(&Value::Null);
        let Some(arguments) = arguments.as_object() else {
            return error_response(
                id,
                -32602,
                "invalid tools/call params: arguments must be object",
            );
        };
        if let Err(err) = validate_tool_arguments_against_schema(&tool, arguments) {
            return err.to_json_rpc(id);
        }
        if !self.tool_policy.allows_arguments(canonical_name, arguments) {
            return error_response(
                id,
                -32602,
                "confidence=inferred/all is disabled by MCP tool policy because it may call an LLM and write inferred-edge cache rows",
            );
        }

        let envelope = match canonical_name {
            "entity_at" => match self.tool_entity_at(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_find" => match self.tool_find_entity(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_callers_list" => match self.tool_callers_of(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_execution_path_list" => match self.tool_execution_paths_from(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_neighborhood_get" => match self.tool_neighborhood(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_summary_get" => match self.tool_summary(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_issue_list" => match self.tool_issues_for(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "subsystem_member_list" => match self.tool_subsystem_members(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_subsystem_get" => match self.tool_subsystem_of(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "project_status_get" => match self.tool_project_status(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "llm_config_get" => match self.tool_llm_config_get(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "llm_config_set" => match self.tool_llm_config_set(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "semantic_config_get" => match self.tool_semantic_config_get(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "semantic_config_set" => match self.tool_semantic_config_set(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_summary_preview_cost_get" => {
                match self.tool_summary_preview_cost(arguments).await {
                    Ok(value) => value,
                    Err(response) => return response.to_json_rpc(id),
                }
            }
            "entity_source_get" => match self.tool_source_for_entity(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_call_site_list" => match self.tool_call_sites(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_orientation_pack_get" => match self.tool_orientation_pack(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "analyze_start" => match self.tool_analyze_start(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "analyze_status_get" => match self.tool_analyze_status(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "analyze_cancel" => match self.tool_analyze_cancel(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "index_diff_get" => match self.tool_index_diff(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_guidance_list" => match self.tool_guidance_for(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "propose_guidance" => match self.tool_propose_guidance(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "promote_guidance" => match self.tool_promote_guidance(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_finding_list" => match self.tool_findings_for(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_wardline_get" => match self.tool_wardline_for(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_tag_list" => match self.tool_find_by_tag(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_kind_list" => match self.tool_find_by_kind(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_wardline_list" => match self.tool_find_by_wardline(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "module_circular_import_list" => match self.tool_find_circular_imports(arguments).await
            {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_coupling_hotspot_list" => {
                match self.tool_find_coupling_hotspots(arguments).await {
                    Ok(value) => value,
                    Err(response) => return response.to_json_rpc(id),
                }
            }
            "entity_entry_point_list" => match self.tool_find_entry_points(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_http_route_list" => match self.tool_find_http_routes(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_data_model_list" => match self.tool_find_data_models(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_test_list" => match self.tool_find_tests(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_deprecation_list" => match self.tool_find_deprecations(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_todo_list" => match self.tool_find_todos(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_test_caller_list" => match self.tool_what_tests_this(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_high_churn_list" => match self.tool_high_churn(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_recent_change_list" => match self.tool_recently_changed(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_dead_list" => match self.tool_find_dead_code(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_semantic_search_list" => match self.tool_search_semantic(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "project_finding_list" => match self.tool_project_findings(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_resolve" => match self.tool_entity_resolve(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            "entity_relation_list" => match self.tool_relation_list(arguments).await {
                Ok(value) => value,
                Err(response) => return response.to_json_rpc(id),
            },
            _ => unreachable!("known tools checked above"),
        };

        tool_json_rpc_response(id, &envelope)
    }

    async fn handle_resources_read(&self, id: &Value, params: Option<&Value>) -> Value {
        let Some(uri) = params
            .and_then(Value::as_object)
            .and_then(|p| p.get("uri"))
            .and_then(Value::as_str)
        else {
            return error_response(id, -32602, "invalid resources/read params: missing uri");
        };
        if uri != "loomweave://context" {
            return error_response(id, -32602, &format!("unknown resource: {uri}"));
        }
        let snapshot_json = self.context_snapshot_json().await;
        result_response(
            id,
            &json!({
                "contents": [
                    {
                        "uri": "loomweave://context",
                        "mimeType": "application/json",
                        "text": snapshot_json
                    }
                ]
            }),
        )
    }

    async fn context_snapshot_json(&self) -> String {
        // Single fallback used by both the reader-error and serialize-error
        // branches: serialize a real `ProjectSnapshot` so the shape stays in
        // lock-step with the type as it gains fields. `degraded: true` — this
        // path is only reached when the reader pool errored or a healthy
        // snapshot failed to serialize, so the consumer must not read the zero
        // counts as a genuinely empty index.
        let fallback = || {
            let snap = crate::snapshot::unreadable_db_snapshot();
            serde_json::to_string(&snap).unwrap_or_default()
        };

        let project_root = self.project_root.clone();
        let snapshot = self
            .readers
            .with_reader(move |conn| Ok(crate::snapshot::project_snapshot(conn, &project_root)))
            .await;
        match snapshot {
            Ok(snap) => serde_json::to_string(&snap).unwrap_or_else(|_| fallback()),
            Err(err) => {
                tracing::warn!(error = %err, "loomweave://context snapshot failed");
                fallback()
            }
        }
    }

    async fn tool_propose_guidance(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let entity_id = required_str(arguments, "entity_id")?.to_owned();
        let content = required_str(arguments, "content")?.to_owned();
        let scope_level = arguments
            .get("scope_level")
            .and_then(Value::as_str)
            .unwrap_or("function")
            .to_owned();
        let pinned = optional_bool(arguments, "pinned")?.unwrap_or(false);
        let name = arguments
            .get("name")
            .and_then(Value::as_str)
            .map(ToOwned::to_owned);
        let expires = arguments
            .get("expires")
            .and_then(Value::as_str)
            .map(ToOwned::to_owned);
        let project_root = self.project_root.clone();
        // Accept an id-or-SEI and resolve to the canonical locator FIRST
        // (clarion-d76e7f7267). The default match_rule and the proposal's
        // `entity_id` MUST carry the resolved `entity.id` (a locator), not the
        // raw arg — otherwise a pasted SEI silently makes the stored guidance
        // SEI-keyed instead of locator-keyed (a persisted-data-shape drift).
        let entity_lookup_id = entity_id.clone();
        let entity = match self
            .readers
            .with_reader(move |conn| resolve_entity_ref(conn, &entity_lookup_id))
            .await
        {
            Ok(Some(entity)) => entity,
            Ok(None) => {
                return Ok(tool_error_envelope(
                    McpErrorCode::EntityNotFound,
                    &format!("entity {entity_id} was not found"),
                    false,
                ));
            }
            Err(err) => {
                return Ok(tool_error_envelope(
                    McpErrorCode::StorageError,
                    &format!("read entity for guidance proposal: {err}"),
                    storage_retryable(&err),
                ));
            }
        };

        let resolved_id = entity.id.clone();
        let match_rules = arguments
            .get("match_rules")
            .and_then(Value::as_array)
            .cloned()
            .unwrap_or_else(|| vec![json!({"type": "entity", "id": resolved_id})]);

        let proposal = GuidanceProposal {
            entity_id: entity.id.clone(),
            content,
            scope_level,
            match_rules,
            name,
            pinned,
            expires,
        };
        let detail = match proposal.to_observation_detail() {
            Ok(detail) => detail,
            Err(err) => {
                return Ok(tool_error_envelope(
                    McpErrorCode::StorageError,
                    &format!("build guidance proposal: {err}"),
                    false,
                ));
            }
        };

        let Some(client) = self.filigree_client.clone() else {
            return Ok(tool_error_envelope(
                McpErrorCode::IoError,
                "Filigree integration is not configured; cannot create guidance proposal observation",
                true,
            ));
        };
        let file_path = entity.source_file_path.as_deref().map(|path| {
            std::path::Path::new(path)
                .strip_prefix(&project_root)
                .ok()
                .and_then(|rel| rel.to_str())
                .unwrap_or(path)
                .to_owned()
        });
        let request = ObservationCreateRequest {
            summary: format!("Loomweave guidance proposal for {entity_id}"),
            detail,
            file_path,
            line: entity.source_line_start,
            priority: 2,
            actor: "loomweave".to_owned(),
        };

        let response =
            match tokio::task::spawn_blocking(move || client.create_observation(request)).await {
                Ok(Ok(response)) => response,
                Ok(Err(err)) => {
                    return Ok(tool_error_envelope(
                        McpErrorCode::IoError,
                        &format!("create Filigree guidance proposal observation: {err}"),
                        true,
                    ));
                }
                Err(err) => {
                    return Ok(tool_error_envelope(
                        McpErrorCode::Internal,
                        &format!("create Filigree guidance proposal task failed: {err}"),
                        true,
                    ));
                }
            };

        Ok(success_envelope(json!({
            "observation_id": response.observation_id,
            "promoted": false,
        })))
    }

    async fn tool_promote_guidance(
        &self,
        arguments: &serde_json::Map<String, Value>,
    ) -> std::result::Result<Value, ParamError> {
        let observation_id = required_str(arguments, "observation_id")?.to_owned();
        let Some(client) = self.filigree_client.clone() else {
            return Ok(tool_error_envelope(
                McpErrorCode::IoError,
                "Filigree integration is not configured; cannot read guidance proposal observation",
                true,
            ));
        };
        let lookup_client = client.clone();
        let lookup_id = observation_id.clone();
        let observation =
            match tokio::task::spawn_blocking(move || lookup_client.observation_by_id(&lookup_id))
                .await
            {
                Ok(Ok(Some(observation))) => observation,
                Ok(Ok(None)) => {
                    return Ok(tool_error_envelope(
                        McpErrorCode::NotFound,
                        &format!("observation {observation_id} was not found"),
                        false,
                    ));
                }
                Ok(Err(err)) => {
                    return Ok(tool_error_envelope(
                        McpErrorCode::IoError,
                        &format!("read Filigree observation {observation_id}: {err}"),
                        true,
                    ));
                }
                Err(err) => {
                    return Ok(tool_error_envelope(
                        McpErrorCode::Internal,
                        &format!("read Filigree observation task failed: {err}"),
                        true,
                    ));
                }
            };

        let proposal = match GuidanceProposal::from_observation_detail(&observation.detail) {
            Ok(proposal) => proposal,
            Err(err) => {
                return Ok(tool_error_envelope(
                    McpErrorCode::StorageError,
                    &format!("observation {observation_id} is not a guidance proposal: {err}"),
                    false,
                ));
            }
        };
        let authored_at = guidance_authored_at_from_clock(&(self.clock)());
        let promoted = match proposal.to_promoted_sheet(&authored_at) {
            Ok(promoted) => promoted,
            Err(err) => {
                return Ok(tool_error_envelope(
                    McpErrorCode::StorageError,
                    &format!("build promoted guidance sheet: {err}"),
                    false,
                ));
            }
        };

        let db_path = loomweave_core::store::db_path(&self.project_root);
        let project_root = self.project_root.clone();
        let sheet_id = promoted.id.clone();
        let write_result =
            tokio::task::spawn_blocking(move || -> std::result::Result<usize, String> {
                let conn =
                    open_guidance_write_connection(&db_path).map_err(|err| err.to_string())?;
                upsert_guidance_sheet(
                    &conn,
                    &GuidanceSheetInput {
                        id: &promoted.id,
                        name: &promoted.name,
                        short_name: &promoted.short_name,
                        properties: &promoted.properties,
                    },
                )
                .map_err(|err| err.to_string())?;
                let Some(sheet) = loomweave_storage::get_guidance_sheet(&conn, &promoted.id)
                    .map_err(|err| err.to_string())?
                else {
                    return Ok(0);
                };
                invalidate_summaries_for_sheet(&conn, &sheet, &project_root)
                    .map_err(|err| err.to_string())
            })
            .await;
        let invalidated = match write_result {
            Ok(Ok(invalidated)) => invalidated,
            Ok(Err(err)) => {
                return Ok(tool_error_envelope(
                    McpErrorCode::StorageError,
                    &format!("write promoted guidance sheet {sheet_id}: {err}"),
                    true,
                ));
            }
            Err(err) => {
                return Ok(tool_error_envelope(
                    McpErrorCode::Internal,
                    &format!("write promoted guidance sheet task failed: {err}"),
                    true,
                ));
            }
        };

        let dismiss_id = observation_id.clone();
        let dismissed = tokio::task::spawn_blocking(move || {
            client.dismiss_observation(&dismiss_id, "promoted to Loomweave guidance sheet")
        })
        .await
        .is_ok_and(|result| result.is_ok());

        Ok(success_envelope(json!({
            "observation_id": observation_id,
            "sheet_id": sheet_id,
            "invalidated_summaries": invalidated,
            "observation_dismissed": dismissed,
        })))
    }
}

async fn invoke_llm_provider(
    provider: Arc<dyn LlmProvider>,
    request: LlmRequest,
) -> Result<LlmResponse, LlmProviderError> {
    provider.invoke(request).await
}

fn open_guidance_write_connection(path: &std::path::Path) -> rusqlite::Result<Connection> {
    let conn = Connection::open_with_flags(
        path,
        OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_URI,
    )?;
    conn.busy_timeout(std::time::Duration::from_secs(5))?;
    conn.pragma_update(None, "foreign_keys", "ON")?;
    Ok(conn)
}

#[derive(Clone)]
struct SummaryLlmState {
    writer: mpsc::Sender<WriterCmd>,
    config: LlmConfig,
    provider: Arc<dyn LlmProvider>,
}

#[derive(Clone)]
struct SemanticSearchState {
    config: SemanticSearchConfig,
    provider: Arc<dyn EmbeddingProvider>,
}

#[derive(Default)]
struct BudgetLedger {
    spent_tokens: u64,
    reserved_tokens: u64,
    blocked: bool,
}

struct BudgetReservation {
    budget: Arc<Mutex<BudgetLedger>>,
    amount_tokens: u64,
    active: bool,
}

impl BudgetReservation {
    fn commit(mut self, actual_tokens: u64, ceiling_tokens: u64) -> bool {
        let mut budget = self
            .budget
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if self.active {
            budget.reserved_tokens = budget.reserved_tokens.saturating_sub(self.amount_tokens);
            self.active = false;
        }
        // `budget.blocked` gates *new* reservations, not in-flight commits.
        // A reservation that already cleared reserve_budget paid for its
        // dispatch slot; commit it iff the actual usage fits the ceiling.
        if budget.spent_tokens.saturating_add(actual_tokens) > ceiling_tokens {
            budget.blocked = true;
            return false;
        }
        budget.spent_tokens = budget.spent_tokens.saturating_add(actual_tokens);
        true
    }
}

impl Drop for BudgetReservation {
    fn drop(&mut self) {
        if !self.active {
            return;
        }
        let mut budget = self
            .budget
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        budget.reserved_tokens = budget.reserved_tokens.saturating_sub(self.amount_tokens);
        self.active = false;
    }
}

enum SummaryRead {
    Ready(Box<SummaryReady>),
    EntityNotFound(String),
    MissingContentHash(String),
    // The pre-serialized entity payload (carrying its `sei`) is built in the
    // reader closure; the summary envelope is otherwise assembled after async
    // writer/LLM calls where no reader connection is in scope (REQ-C-04 /
    // ADR-038).
    ScopeDeferred(Value),
    BriefingBlocked(Value, String),
}

struct SummaryReady {
    entity: EntityRow,
    /// Pre-serialized entity payload (including the SEI read-time join), built
    /// in the reader closure for the same reason as the variants above.
    entity_json: Value,
    key: SummaryCacheKey,
    cached: Option<SummaryCacheEntry>,
    guidance_text: String,
    caller_count: i64,
    fan_out: i64,
}

struct IssuesForRead {
    entities: Vec<EntityRow>,
    /// Pre-serialized entity payloads keyed by locator, built inside the reader
    /// closure so the SEI read-time join (REQ-C-04 / ADR-038) happens while a
    /// connection is in scope. `tool_issues_for` runs outside any reader closure
    /// (it interleaves Filigree HTTP calls), so the `sei` field cannot be
    /// resolved during accumulation — it is resolved here, ahead of the loop.
    entity_json_by_id: HashMap<String, Value>,
    entity_cap_truncated: bool,
}

struct IssuesForAccumulator {
    entities_by_id: HashMap<String, EntityRow>,
    entity_json_by_id: HashMap<String, Value>,
    association_aliases: HashMap<String, String>,
    seen_issue_ids: BTreeSet<String>,
    matched: Vec<Value>,
    drifted: Vec<Value>,
    not_found: Vec<Value>,
    diagnostics: Vec<Value>,
    emitted: usize,
    issue_cap_truncated: bool,
}

impl IssuesForAccumulator {
    fn new(entities: &[EntityRow], entity_json_by_id: HashMap<String, Value>) -> Self {
        // Map every key Filigree might echo back in `loomweave_entity_id` to the
        // current locator (`entity.id`). A SEI-bearing entity is queried by SEI
        // only (see `tool_issues_for`), so the SEI→locator alias is the live
        // path for those rows; the locator self-mapping covers no-SEI entities
        // and any straggler locator-keyed rows during the SEI migration window.
        let mut association_aliases = HashMap::new();
        for entity in entities {
            association_aliases.insert(entity.id.clone(), entity.id.clone());
            if let Some(sei) = entity_json_by_id
                .get(&entity.id)
                .and_then(|json| json.get("sei"))
                .and_then(Value::as_str)
                .filter(|sei| !sei.trim().is_empty())
            {
                association_aliases.insert(sei.to_owned(), entity.id.clone());
            }
        }
        Self {
            entities_by_id: entities
                .iter()
                .map(|entity| (entity.id.clone(), entity.clone()))
                .collect(),
            entity_json_by_id,
            association_aliases,
            seen_issue_ids: BTreeSet::new(),
            matched: Vec::new(),
            drifted: Vec::new(),
            not_found: Vec::new(),
            diagnostics: Vec::new(),
            emitted: 0,
            issue_cap_truncated: false,
        }
    }

    fn add_response(&mut self, response: EntityAssociationsResponse) {
        for association in response.associations {
            if self.emitted >= 100 {
                self.issue_cap_truncated = true;
                break;
            }
            if !self.seen_issue_ids.insert(association.issue_id.clone()) {
                continue;
            }
            self.emitted += 1;
            self.add_association(&association);
        }
    }

    fn add_association(&mut self, association: &EntityAssociation) {
        // The nested entity payload (including its `sei`) was pre-built in the
        // reader closure; `entities_by_id` is retained only for content-hash
        // drift classification.
        let canonical_entity_id = self
            .association_aliases
            .get(&association.loomweave_entity_id)
            .map_or(association.loomweave_entity_id.as_str(), String::as_str);
        let entity_json = self.entity_json_by_id.get(canonical_entity_id).cloned();
        match self.entities_by_id.get(canonical_entity_id) {
            None => {
                self.not_found
                    .push(association_json(association, None, None, "not_found", None));
            }
            Some(entity) => match entity.content_hash.as_deref() {
                Some(current_hash) if current_hash == association.content_hash_at_attach => {
                    self.matched.push(association_json(
                        association,
                        entity_json.as_ref(),
                        Some(current_hash),
                        "matched",
                        Some(&entity.id),
                    ));
                }
                Some(current_hash) => {
                    self.drifted.push(association_json(
                        association,
                        entity_json.as_ref(),
                        Some(current_hash),
                        "drifted",
                        Some(&entity.id),
                    ));
                }
                None => {
                    self.diagnostics.push(json!({
                        "code": "LMWV-ENTITY-CONTENT-HASH-MISSING",
                        "entity_id": entity.id
                    }));
                    self.matched.push(association_json(
                        association,
                        entity_json.as_ref(),
                        None,
                        "unknown",
                        Some(&entity.id),
                    ));
                }
            },
        }
    }

    /// Unique `issue_id`s across matched + drifted entries, in first-seen
    /// order. `add_response` already dedupes `issue_id`s globally, so this is
    /// the set of distinct issues to fetch detail for — one request each (no
    /// N+1).
    fn enrichable_issue_ids(&self) -> Vec<String> {
        let mut seen = BTreeSet::new();
        let mut ids = Vec::new();
        for entry in self.matched.iter().chain(self.drifted.iter()) {
            if let Some(id) = entry.get("issue_id").and_then(Value::as_str)
                && seen.insert(id.to_owned())
            {
                ids.push(id.to_owned());
            }
        }
        ids
    }

    /// Attach an `issue` field (id/title/status/priority — the hydrated stub,
    /// weft-4a46553503) to every matched and drifted entry. The value is the
    /// fetched [`IssueDetail`] when available, else `null` — a stable shape
    /// that signals "enrichment attempted, no detail" without forcing the
    /// consumer to probe for a missing key. The stub's `id` is backfilled from
    /// the entry's own `issue_id` when the detail route omitted it (an older
    /// server), so a present stub ALWAYS carries the complete tuple.
    fn apply_issue_details(&mut self, details: &HashMap<String, Option<IssueDetail>>) {
        for entry in self.matched.iter_mut().chain(self.drifted.iter_mut()) {
            let issue_id = entry
                .get("issue_id")
                .and_then(Value::as_str)
                .map(ToOwned::to_owned);
            let issue_value = issue_id
                .as_deref()
                .and_then(|id| details.get(id))
                .and_then(Option::as_ref)
                .map(|detail| {
                    let mut detail = detail.clone();
                    if detail.id.is_empty()
                        && let Some(id) = issue_id.as_deref()
                    {
                        id.clone_into(&mut detail.id);
                    }
                    detail
                })
                .and_then(|detail| serde_json::to_value(detail).ok())
                .unwrap_or(Value::Null);
            if let Some(object) = entry.as_object_mut() {
                object.insert("issue".to_owned(), issue_value);
            }
        }
    }

    fn into_envelope(
        self,
        entity_cap_truncated: bool,
        requests_total: usize,
        detail_requests_total: usize,
        filigree_endpoint: &Value,
    ) -> Value {
        let truncation_reason = if self.issue_cap_truncated {
            Some("issue-cap")
        } else {
            entity_cap_truncated.then_some("entity-cap")
        };
        // result_kind lets a consumer act on the outcome without re-deriving it
        // from array lengths, and — paired with `available: false` from
        // `issues_unavailable` — distinguishes "Filigree reachable but no issues
        // are attached" (no_matches) from "Filigree unreachable/disabled"
        // (unavailable).
        let result_kind = if self.matched.is_empty() && self.drifted.is_empty() {
            "no_matches"
        } else {
            "matched"
        };
        let mut envelope = success_envelope_with_truncation_and_stats(
            json!({
                "available": true,
                "result_kind": result_kind,
                "filigree_endpoint": filigree_endpoint,
                "matched": self.matched,
                "drifted": self.drifted,
                "not_found": self.not_found
            }),
            truncation_reason,
            json!({
                "filigree_requests_total": requests_total,
                "filigree_issues_returned_total": self.emitted,
                "filigree_detail_requests_total": detail_requests_total
            }),
        );
        if let Some(object) = envelope.as_object_mut()
            && !self.diagnostics.is_empty()
        {
            object.insert("diagnostics".to_owned(), Value::Array(self.diagnostics));
        }
        envelope
    }
}

#[derive(Clone)]
struct InferenceLlmState {
    writer: mpsc::Sender<WriterCmd>,
    config: LlmConfig,
    provider: Arc<dyn LlmProvider>,
}

struct InferredInflightGuard {
    in_flight: InferredInflight,
    key: InferredEdgeCacheKey,
    sender: broadcast::Sender<InferredDispatchOutcome>,
    active: bool,
}

impl InferredInflightGuard {
    fn new(
        in_flight: InferredInflight,
        key: InferredEdgeCacheKey,
        sender: broadcast::Sender<InferredDispatchOutcome>,
    ) -> Self {
        Self {
            in_flight,
            key,
            sender,
            active: true,
        }
    }

    async fn remove(mut self) -> Option<broadcast::Sender<InferredDispatchOutcome>> {
        let removed = remove_matching_inferred_inflight(
            Arc::clone(&self.in_flight),
            self.key.clone(),
            self.sender.clone(),
        )
        .await;
        self.active = false;
        removed
    }
}

impl Drop for InferredInflightGuard {
    fn drop(&mut self) {
        if !self.active {
            return;
        }
        let in_flight = Arc::clone(&self.in_flight);
        let key = self.key.clone();
        let sender = self.sender.clone();
        if let Ok(handle) = tokio::runtime::Handle::try_current() {
            handle.spawn(async move {
                let _ = remove_matching_inferred_inflight(in_flight, key, sender).await;
            });
        }
    }
}

async fn remove_matching_inferred_inflight(
    in_flight: InferredInflight,
    key: InferredEdgeCacheKey,
    sender: broadcast::Sender<InferredDispatchOutcome>,
) -> Option<broadcast::Sender<InferredDispatchOutcome>> {
    let mut map = in_flight.lock().await;
    if map
        .get(&key)
        .is_some_and(|current| current.same_channel(&sender))
    {
        map.remove(&key)
    } else {
        None
    }
}

#[derive(Clone)]
struct InferredRead {
    caller: EntityRow,
    sites: Vec<UnresolvedCallSiteRow>,
    candidates: Vec<EntityRow>,
    key: InferredEdgeCacheKey,
    cached: Option<InferredEdgeCacheEntry>,
}

#[derive(Debug, Clone, Default)]
struct InferredDispatchStats {
    cache_hits_total: u64,
    cache_misses_total: u64,
    edges_materialized_total: u64,
    edges_skipped_static_duplicates_total: u64,
    /// LLM-proposed `to_id` values that did not resolve in the `entities`
    /// table at write time (clarion-df58379de4). Counted here, dropped from
    /// the persisted edge set to avoid the FK violation that previously
    /// poisoned the cache row and re-burned LLM tokens on retry.
    unresolved_targets_dropped_total: u64,
    briefing_blocked_total: u64,
    candidate_callers_considered: u64,
    coalesced_waits_total: u64,
    tokens_input: i64,
    tokens_cached_input: i64,
    tokens_output: i64,
    tokens_total: i64,
    cost_usd: f64,
}

impl InferredDispatchStats {
    fn cache_hit(write: InferredEdgeWriteStats) -> Self {
        Self {
            cache_hits_total: 1,
            edges_materialized_total: write.inserted_edges,
            edges_skipped_static_duplicates_total: write.skipped_static_duplicates,
            ..Self::default()
        }
    }

    fn cache_miss(write: InferredEdgeWriteStats, response: &LlmResponse) -> Self {
        Self {
            cache_misses_total: 1,
            edges_materialized_total: write.inserted_edges,
            edges_skipped_static_duplicates_total: write.skipped_static_duplicates,
            tokens_input: i64::from(response.input_tokens),
            tokens_cached_input: i64::from(response.cached_input_tokens),
            tokens_output: i64::from(response.output_tokens),
            tokens_total: i64::from(response.total_tokens),
            cost_usd: response.cost_usd,
            ..Self::default()
        }
    }

    fn briefing_blocked() -> Self {
        Self {
            briefing_blocked_total: 1,
            ..Self::default()
        }
    }

    fn merge(&mut self, other: &Self) {
        self.cache_hits_total += other.cache_hits_total;
        self.cache_misses_total += other.cache_misses_total;
        self.edges_materialized_total += other.edges_materialized_total;
        self.edges_skipped_static_duplicates_total += other.edges_skipped_static_duplicates_total;
        self.unresolved_targets_dropped_total += other.unresolved_targets_dropped_total;
        self.briefing_blocked_total += other.briefing_blocked_total;
        self.candidate_callers_considered += other.candidate_callers_considered;
        self.coalesced_waits_total += other.coalesced_waits_total;
        self.tokens_input += other.tokens_input;
        self.tokens_cached_input += other.tokens_cached_input;
        self.tokens_output += other.tokens_output;
        self.tokens_total += other.tokens_total;
        self.cost_usd += other.cost_usd;
    }

    fn to_json(&self) -> Value {
        json!({
            "inferred_dispatch_cache_hits_total": self.cache_hits_total,
            "inferred_dispatch_misses_total": self.cache_misses_total,
            "inferred_edges_materialized_total": self.edges_materialized_total,
            "inferred_edges_skipped_static_duplicates_total": self.edges_skipped_static_duplicates_total,
            "inferred_unresolved_targets_dropped_total": self.unresolved_targets_dropped_total,
            "inferred_dispatch_briefing_blocked_total": self.briefing_blocked_total,
            "inferred_candidate_callers_considered": self.candidate_callers_considered,
            "inferred_dispatch_coalesced_total": self.coalesced_waits_total,
            "inferred_tokens_input": self.tokens_input,
            "inferred_tokens_cached_input": self.tokens_cached_input,
            "inferred_tokens_output": self.tokens_output,
            "inferred_tokens_total": self.tokens_total,
            "inferred_cost_usd": self.cost_usd
        })
    }
}

#[derive(Debug, Clone)]
struct InferredDispatchFailure {
    code: McpErrorCode,
    message: String,
    retryable: bool,
    stats_delta: Value,
    diagnostics: Vec<Value>,
}

impl InferredDispatchFailure {
    fn new(code: McpErrorCode, message: &str, retryable: bool) -> Self {
        Self {
            code,
            message: message.to_owned(),
            retryable,
            stats_delta: json!({}),
            diagnostics: Vec::new(),
        }
    }

    fn from_storage(err: &StorageError) -> Self {
        // FK violations are deterministic against the same row set; treating
        // them as `retryable=true` causes the client to re-issue the LLM call
        // and re-pay the token cost (clarion-df58379de4). Mark them
        // non-retryable so a client honouring the hint gives up immediately.
        Self {
            code: McpErrorCode::StorageError,
            message: err.to_string(),
            retryable: !err.is_foreign_key_violation(),
            stats_delta: json!({}),
            diagnostics: Vec::new(),
        }
    }

    fn with_stats(mut self, stats_delta: Value, diagnostics: Vec<Value>) -> Self {
        self.stats_delta = stats_delta;
        self.diagnostics = diagnostics;
        self
    }

    fn to_envelope(&self) -> Value {
        if self.code == McpErrorCode::TokenCeilingExceeded {
            return token_ceiling_envelope(&self.message);
        }
        tool_error_envelope_with_diagnostics(
            self.code,
            &self.message,
            self.retryable,
            self.stats_delta.clone(),
            self.diagnostics.clone(),
        )
    }
}

#[derive(Debug, Clone)]
enum InferredDispatchOutcome {
    Ok(InferredDispatchStats),
    Err(InferredDispatchFailure),
}

impl InferredDispatchOutcome {
    fn from_result(result: Result<InferredDispatchStats, InferredDispatchFailure>) -> Self {
        match result {
            Ok(stats) => Self::Ok(stats),
            Err(err) => Self::Err(err),
        }
    }

    fn into_result(self) -> Result<InferredDispatchStats, InferredDispatchFailure> {
        match self {
            Self::Ok(stats) => Ok(stats),
            Self::Err(err) => Err(err),
        }
    }
}

#[derive(Debug, Deserialize)]
struct InferredCallsResponse {
    #[serde(default)]
    edges: Vec<InferredCallsResponseEdge>,
}

#[derive(Debug, Deserialize)]
struct InferredCallsResponseEdge {
    site_key: Option<String>,
    target_id: String,
    confidence: Option<f64>,
    rationale: Option<String>,
}

#[derive(Debug, Error)]
pub enum McpError {
    #[error("invalid JSON-RPC frame body: {0}")]
    Json(#[from] serde_json::Error),

    #[error("MCP transport error: {0}")]
    Transport(#[from] TransportError),

    #[error("MCP runtime error: {0}")]
    Runtime(#[from] std::io::Error),
}

/// Decode and handle a state-free MCP frame.
///
/// Storage-backed tool calls require [`handle_frame_with_state`].
pub fn handle_frame(frame: &Frame) -> Result<Option<Frame>, McpError> {
    let request = serde_json::from_slice(&frame.body)?;
    let Some(response) = handle_json_rpc(&request) else {
        return Ok(None);
    };
    Ok(Some(encode_response_frame(&response)?))
}

pub async fn handle_frame_with_state(
    state: &ServerState,
    frame: &Frame,
) -> Result<Option<Frame>, McpError> {
    let request = serde_json::from_slice(&frame.body)?;
    handle_request_value_with_state(state, request).await
}

async fn handle_request_value_with_state(
    state: &ServerState,
    request: Value,
) -> Result<Option<Frame>, McpError> {
    let Some(response) = state.handle_json_rpc(&request).await else {
        return Ok(None);
    };
    Ok(Some(encode_response_frame(&response)?))
}

fn handle_stdio_frame(frame: &Frame) -> Result<Option<Frame>, McpError> {
    handle_frame(frame)
}

fn encode_response_frame(response: &Value) -> Result<Frame, McpError> {
    Ok(Frame {
        body: serde_json::to_vec(response)?,
    })
}

fn is_json_rpc_notification(request: &Value) -> bool {
    request
        .as_object()
        .is_some_and(|object| object.get("method").is_some() && object.get("id").is_none())
}

fn request_id_key(id: &Value) -> Option<String> {
    match id {
        Value::String(value) => Some(format!("s:{value}")),
        Value::Number(value) => Some(format!("n:{value}")),
        _ => None,
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StdioFraming {
    ContentLength,
    JsonLine,
}

struct StdioFrame {
    body: Vec<u8>,
    framing: StdioFraming,
}

fn read_stdio_frame(reader: &mut impl std::io::BufRead) -> Result<Option<StdioFrame>, McpError> {
    let Some(first_byte) = peek_stdio_frame_start(reader)? else {
        return Ok(None);
    };
    if first_byte == b'{' || first_byte == b'[' || first_byte.is_ascii_whitespace() {
        return Ok(Some(read_json_line_frame(reader)?));
    }
    match loomweave_core::plugin::read_frame(reader, ContentLengthCeiling::DEFAULT) {
        Ok(frame) => Ok(Some(StdioFrame {
            body: frame.body,
            framing: StdioFraming::ContentLength,
        })),
        Err(TransportError::Io(err)) if err.kind() == std::io::ErrorKind::UnexpectedEof => Ok(None),
        Err(err) => Err(err.into()),
    }
}

fn peek_stdio_frame_start(reader: &mut impl std::io::BufRead) -> Result<Option<u8>, McpError> {
    loop {
        let buffer = reader.fill_buf()?;
        if buffer.is_empty() {
            return Ok(None);
        }
        let blank_prefix = buffer
            .iter()
            .take_while(|byte| matches!(byte, b'\r' | b'\n'))
            .count();
        if blank_prefix == 0 {
            return Ok(Some(buffer[0]));
        }
        reader.consume(blank_prefix);
    }
}

fn read_json_line_frame(reader: &mut impl std::io::BufRead) -> Result<StdioFrame, McpError> {
    let mut body = Vec::new();
    let read = std::io::BufRead::read_until(reader, b'\n', &mut body)?;
    if read == 0 {
        return Err(TransportError::Io(std::io::Error::new(
            std::io::ErrorKind::UnexpectedEof,
            "EOF while reading MCP JSON line",
        ))
        .into());
    }
    while matches!(body.last(), Some(b'\n' | b'\r')) {
        body.pop();
    }
    Ok(StdioFrame {
        body,
        framing: StdioFraming::JsonLine,
    })
}

fn write_stdio_response(
    writer: &mut impl std::io::Write,
    response: &Frame,
    framing: StdioFraming,
) -> Result<(), McpError> {
    match framing {
        StdioFraming::ContentLength => {
            loomweave_core::plugin::write_frame(writer, response)?;
        }
        StdioFraming::JsonLine => {
            writer.write_all(&response.body)?;
            writer.write_all(b"\n")?;
            writer.flush()?;
        }
    }
    Ok(())
}

/// Serve state-free MCP protocol metadata over stdio.
///
/// Storage-backed tool calls require [`serve_stdio_with_state`].
pub fn serve_stdio(
    reader: &mut impl std::io::BufRead,
    writer: &mut impl std::io::Write,
) -> Result<(), McpError> {
    loop {
        let Some(frame) = read_stdio_frame(reader)? else {
            return Ok(());
        };
        let framing = frame.framing;
        if let Some(response) = handle_stdio_frame(&Frame { body: frame.body })? {
            write_stdio_response(writer, &response, framing)?;
        }
    }
}

pub fn serve_stdio_with_state(
    state: &ServerState,
    reader: &mut (impl std::io::BufRead + Send),
    writer: &mut impl std::io::Write,
) -> Result<(), McpError> {
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()?;
    serve_stdio_with_state_on_runtime(&runtime, state, reader, writer)
}

pub fn serve_stdio_with_state_on_runtime(
    runtime: &tokio::runtime::Runtime,
    state: &ServerState,
    reader: &mut (impl std::io::BufRead + Send),
    writer: &mut impl std::io::Write,
) -> Result<(), McpError> {
    let _guard = runtime.enter();
    let (frame_tx, frame_rx) = mpsc::unbounded_channel();
    let state = state.clone();
    std::thread::scope(|scope| {
        scope.spawn(move || {
            loop {
                let message = read_stdio_frame(reader);
                let done = !matches!(message, Ok(Some(_)));
                if frame_tx.send(message).is_err() || done {
                    break;
                }
            }
        });
        runtime.block_on(serve_stdio_with_state_event_loop(state, frame_rx, writer))
    })
}

async fn serve_stdio_with_state_event_loop(
    state: ServerState,
    mut frame_rx: mpsc::UnboundedReceiver<Result<Option<StdioFrame>, McpError>>,
    writer: &mut impl std::io::Write,
) -> Result<(), McpError> {
    let (response_tx, mut response_rx) =
        mpsc::unbounded_channel::<(StdioFraming, Result<Option<Frame>, McpError>)>();
    let mut input_closed = false;
    let mut pending_responses = 0usize;

    loop {
        tokio::select! {
            maybe_frame = frame_rx.recv(), if !input_closed => {
                match maybe_frame {
                    Some(Ok(Some(frame))) => {
                        let request: Value = serde_json::from_slice(&frame.body)?;
                        if should_spawn_stateful_stdio_request(&request) {
                            if let Some(key) = request.get("id").and_then(request_id_key) {
                                state.begin_request(&key).await;
                            }
                            pending_responses += 1;
                            let task_state = state.clone();
                            let task_tx = response_tx.clone();
                            let framing = frame.framing;
                            tokio::spawn(async move {
                                let result = handle_request_value_with_state(&task_state, request).await;
                                let _ = task_tx.send((framing, result));
                            });
                        } else if let Some(response) = handle_request_value_with_state(&state, request).await? {
                            write_stdio_response(writer, &response, frame.framing)?;
                        }
                    }
                    Some(Ok(None)) | None => {
                        input_closed = true;
                    }
                    Some(Err(err)) => return Err(err),
                }
            }
            maybe_response = response_rx.recv(), if pending_responses > 0 => {
                let Some((framing, result)) = maybe_response else {
                    return Err(McpError::Runtime(std::io::Error::new(
                        std::io::ErrorKind::BrokenPipe,
                        "MCP stdio response channel closed with pending requests",
                    )));
                };
                pending_responses = pending_responses.saturating_sub(1);
                if let Some(response) = result? {
                    write_stdio_response(writer, &response, framing)?;
                }
            }
            else => {
                if input_closed && pending_responses == 0 {
                    return Ok(());
                }
            }
        }
    }
}

fn should_spawn_stateful_stdio_request(request: &Value) -> bool {
    request
        .as_object()
        .is_some_and(|object| object.get("id").is_some())
        && request
            .get("method")
            .and_then(Value::as_str)
            .is_some_and(|method| method == "tools/call")
}

/// Build the `initialize` result, advertising only the capabilities the
/// handling path actually serves. The stateless free [`handle_json_rpc`] serves
/// `tools` only (it returns method-not-found for `resources/*` and `prompts/*`),
/// so it passes `stateful = false`; [`ServerState::handle_json_rpc`] serves the
/// full surface and passes `stateful = true`. The `instructions` field is static
/// orientation guidance (not a capability) and is included in both.
fn initialize_result(stateful: bool, policy: McpToolPolicy) -> Value {
    let capabilities = if stateful {
        json!({ "tools": {}, "prompts": {}, "resources": {} })
    } else {
        json!({ "tools": {} })
    };
    json!({
        "protocolVersion": MCP_PROTOCOL_VERSION,
        "capabilities": capabilities,
        "serverInfo": {
            "name": "loomweave",
            "version": env!("CARGO_PKG_VERSION")
        },
        "instructions": server_instructions(policy)
    })
}

fn resources_list() -> Value {
    json!({
        "resources": [
            {
                "uri": "loomweave://context",
                "name": "Loomweave project context",
                "description": "Live entity / subsystem / finding counts and index freshness for this project.",
                "mimeType": "application/json"
            }
        ]
    })
}

fn prompts_list() -> Value {
    json!({
        "prompts": [
            {
                "name": "loomweave-workflow",
                "description": "How to use Loomweave's MCP tools to navigate this codebase."
            }
        ]
    })
}

fn prompts_get(id: &Value, params: Option<&Value>) -> Value {
    let name = params
        .and_then(Value::as_object)
        .and_then(|p| p.get("name"))
        .and_then(Value::as_str);
    if name != Some("loomweave-workflow") {
        return error_response(id, -32602, "unknown prompt");
    }
    result_response(
        id,
        &json!({
            "description": "How to use Loomweave's MCP tools to navigate this codebase.",
            "messages": [
                {
                    "role": "user",
                    "content": { "type": "text", "text": LOOMWEAVE_WORKFLOW_SKILL }
                }
            ]
        }),
    )
}

fn validate_tool_arguments_against_schema(
    tool: &ToolDefinition,
    arguments: &serde_json::Map<String, Value>,
) -> std::result::Result<(), ParamError> {
    if tool
        .input_schema
        .get("additionalProperties")
        .and_then(Value::as_bool)
        != Some(false)
    {
        return Ok(());
    }
    let Some(properties) = tool
        .input_schema
        .get("properties")
        .and_then(Value::as_object)
    else {
        return Ok(());
    };
    for key in arguments.keys() {
        if !properties.contains_key(key) {
            return Err(ParamError::new(&format!(
                "unknown argument for {}: {key}",
                tool.name
            )));
        }
    }
    Ok(())
}

#[derive(Debug)]
struct ParamError {
    message: String,
}

impl ParamError {
    fn new(message: &str) -> Self {
        Self {
            message: message.to_owned(),
        }
    }

    fn to_json_rpc(&self, id: &Value) -> Value {
        error_response(id, -32602, &self.message)
    }
}

struct PathTraversal {
    edge_cap: usize,
    edge_count_visited: usize,
    truncated: bool,
    paths: Vec<Vec<String>>,
}

impl PathTraversal {
    fn new(edge_cap: usize) -> Self {
        Self {
            edge_cap,
            edge_count_visited: 0,
            truncated: false,
            paths: Vec::new(),
        }
    }

    fn walk(
        &mut self,
        conn: &rusqlite::Connection,
        current_id: &str,
        path: &mut Vec<String>,
        remaining_depth: usize,
        confidence: EdgeConfidence,
    ) -> Result<(), StorageError> {
        if remaining_depth == 0 || self.truncated {
            return Ok(());
        }
        for edge in call_edges_from(conn, current_id, confidence)? {
            self.edge_count_visited += 1;
            if self.edge_count_visited > self.edge_cap {
                self.truncated = true;
                return Ok(());
            }
            if path.iter().any(|seen| seen == &edge.to_id) {
                continue;
            }
            path.push(edge.to_id.clone());
            self.paths.push(path.clone());
            self.walk(conn, &edge.to_id, path, remaining_depth - 1, confidence)?;
            path.pop();
            if self.truncated {
                return Ok(());
            }
        }
        Ok(())
    }
}

fn required_str<'a>(
    arguments: &'a serde_json::Map<String, Value>,
    field: &str,
) -> std::result::Result<&'a str, ParamError> {
    arguments
        .get(field)
        .and_then(Value::as_str)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| ParamError::new(&format!("{field} must be a non-empty string")))
}

fn required_i64(
    arguments: &serde_json::Map<String, Value>,
    field: &str,
) -> std::result::Result<i64, ParamError> {
    arguments
        .get(field)
        .and_then(Value::as_i64)
        .ok_or_else(|| ParamError::new(&format!("{field} must be an integer")))
}

fn optional_usize(
    arguments: &serde_json::Map<String, Value>,
    field: &str,
) -> std::result::Result<Option<usize>, ParamError> {
    let Some(value) = arguments.get(field) else {
        return Ok(None);
    };
    let Some(raw) = value.as_u64() else {
        return Err(ParamError::new(&format!(
            "{field} must be a non-negative integer"
        )));
    };
    usize::try_from(raw)
        .map(Some)
        .map_err(|_| ParamError::new(&format!("{field} is too large")))
}

/// Parse the numeric-offset `cursor` argument shared by the bounded single-
/// relation tools (mirrors `tool_find_entity`'s cursor handling). Null/absent
/// starts at offset 0; a non-numeric string is a param error (clarion-d76e7f7267).
fn parse_cursor_offset(
    arguments: &serde_json::Map<String, Value>,
) -> std::result::Result<usize, ParamError> {
    match arguments.get("cursor") {
        None | Some(Value::Null) => Ok(0),
        Some(Value::String(cursor)) => cursor
            .parse::<usize>()
            .map_err(|_| ParamError::new("cursor must be a numeric offset")),
        _ => Err(ParamError::new("cursor must be a string or null")),
    }
}

fn optional_bool(
    arguments: &serde_json::Map<String, Value>,
    field: &str,
) -> std::result::Result<Option<bool>, ParamError> {
    let Some(value) = arguments.get(field) else {
        return Ok(None);
    };
    value
        .as_bool()
        .map(Some)
        .ok_or_else(|| ParamError::new(&format!("{field} must be a boolean")))
}

fn optional_non_empty_string(
    arguments: &serde_json::Map<String, Value>,
    field: &str,
) -> std::result::Result<Option<String>, ParamError> {
    let Some(value) = arguments.get(field) else {
        return Ok(None);
    };
    let Some(value) = value.as_str().filter(|value| !value.trim().is_empty()) else {
        return Err(ParamError::new(&format!(
            "{field} must be a non-empty string"
        )));
    };
    Ok(Some(value.to_owned()))
}

fn optional_u64(
    arguments: &serde_json::Map<String, Value>,
    field: &str,
) -> std::result::Result<Option<u64>, ParamError> {
    let Some(value) = arguments.get(field) else {
        return Ok(None);
    };
    value
        .as_u64()
        .map(Some)
        .ok_or_else(|| ParamError::new(&format!("{field} must be a non-negative integer")))
}

fn llm_config_patch_from_arguments(
    arguments: &serde_json::Map<String, Value>,
) -> std::result::Result<LlmConfigPatch, ParamError> {
    let provider = optional_non_empty_string(arguments, "provider")?
        .as_deref()
        .map(LlmProviderKind::parse)
        .transpose()
        .map_err(|err| ParamError::new(&err.to_string()))?;
    Ok(LlmConfigPatch {
        enabled: optional_bool(arguments, "enabled")?,
        provider,
        allow_live_provider: optional_bool(arguments, "allow_live_provider")?,
        enable_write_tools: optional_bool(arguments, "enable_write_tools")?,
        model_id: optional_non_empty_string(arguments, "model_id")?,
        codex_model: optional_non_empty_string(arguments, "codex_model")?,
        claude_model: optional_non_empty_string(arguments, "claude_model")?,
        openrouter_api_key_env: optional_non_empty_string(arguments, "openrouter_api_key_env")?,
        openrouter_endpoint_url: optional_non_empty_string(arguments, "openrouter_endpoint_url")?,
    })
}

fn semantic_config_patch_from_arguments(
    arguments: &serde_json::Map<String, Value>,
) -> std::result::Result<SemanticConfigPatch, ParamError> {
    let provider = optional_non_empty_string(arguments, "provider")?
        .as_deref()
        .map(SemanticProviderKind::parse)
        .transpose()
        .map_err(|err| ParamError::new(&err.to_string()))?;
    Ok(SemanticConfigPatch {
        enabled: optional_bool(arguments, "enabled")?,
        provider,
        allow_live_provider: optional_bool(arguments, "allow_live_provider")?,
        model_id: optional_non_empty_string(arguments, "model_id")?,
        dimensions: optional_usize(arguments, "dimensions")?,
        endpoint_url: optional_non_empty_string(arguments, "endpoint_url")?,
        api_key_env: optional_non_empty_string(arguments, "api_key_env")?,
        timeout_seconds: optional_u64(arguments, "timeout_seconds")?,
        session_token_ceiling: optional_u64(arguments, "session_token_ceiling")?,
    })
}

fn llm_config_patch_is_empty(patch: &LlmConfigPatch) -> bool {
    patch.enabled.is_none()
        && patch.provider.is_none()
        && patch.allow_live_provider.is_none()
        && patch.enable_write_tools.is_none()
        && patch.model_id.is_none()
        && patch.codex_model.is_none()
        && patch.claude_model.is_none()
        && patch.openrouter_api_key_env.is_none()
        && patch.openrouter_endpoint_url.is_none()
}

fn semantic_config_patch_is_empty(patch: &SemanticConfigPatch) -> bool {
    patch.enabled.is_none()
        && patch.provider.is_none()
        && patch.allow_live_provider.is_none()
        && patch.model_id.is_none()
        && patch.dimensions.is_none()
        && patch.endpoint_url.is_none()
        && patch.api_key_env.is_none()
        && patch.timeout_seconds.is_none()
        && patch.session_token_ceiling.is_none()
}

fn read_llm_config_status(
    path: &Path,
) -> std::result::Result<Value, loomweave_federation::config::ConfigError> {
    if path.exists() {
        let config = McpConfig::from_path(path)?;
        Ok(llm_config_status_json(path, false, &config))
    } else {
        Ok(llm_config_status_json(path, true, &McpConfig::default()))
    }
}

fn llm_config_status_json(path: &Path, created_or_absent: bool, config: &McpConfig) -> Value {
    let selection = select_provider_with_env(config, |name| std::env::var(name).ok());
    let (live, selection_error) = match &selection {
        Ok(
            ProviderSelection::OpenRouter { .. }
            | ProviderSelection::CodexCli
            | ProviderSelection::ClaudeCli,
        ) => (true, Value::Null),
        Ok(ProviderSelection::Disabled | ProviderSelection::Recording) => (false, Value::Null),
        Err(err) => (false, Value::String(err.to_string())),
    };
    json!({
        "config_path": path.display().to_string(),
        "config_absent": created_or_absent && !path.exists(),
        "created": created_or_absent && path.exists(),
        "llm": {
            "enabled": config.llm.enabled,
            "provider": config.llm.provider.as_str(),
            "allow_live_provider": config.llm.allow_live_provider,
            "effective_model": config.llm.effective_model_label(),
            "live": live,
            "selection_error": selection_error,
            "warnings": config.llm_warnings(),
        },
        "serve": {
            "mcp": {
                "enable_write_tools": config.serve.mcp.enable_write_tools,
            }
        }
    })
}

fn read_semantic_config_status(
    path: &Path,
    project_root: &Path,
) -> std::result::Result<Value, loomweave_federation::config::ConfigError> {
    if path.exists() {
        let config = McpConfig::from_path(path)?;
        Ok(semantic_config_status_json(
            path,
            project_root,
            false,
            &config,
        ))
    } else {
        Ok(semantic_config_status_json(
            path,
            project_root,
            true,
            &McpConfig::default(),
        ))
    }
}

fn semantic_config_status_json(
    path: &Path,
    project_root: &Path,
    created_or_absent: bool,
    config: &McpConfig,
) -> Value {
    let semantic = &config.semantic_search;
    // Probe the override-aware store path (the SAME helper `EmbeddingStore::
    // open_in_store_dir` writes to) so a `[loomweave].store_dir` relocation does
    // not make a populated sidecar report as absent (read-vs-status parity).
    let sidecar_path = embeddings_db_path(project_root);
    let sidecar_count = semantic_sidecar_count(&sidecar_path);
    let has_key = std::env::var(&semantic.api_key_env)
        .ok()
        .as_deref()
        .is_some_and(|value| !value.trim().is_empty());
    let provider_error = semantic_provider_error(semantic, has_key).map(Value::String);
    let provider_available = semantic.enabled && provider_error.is_none();
    let vector_count = sidecar_count.as_ref().ok().and_then(|value| *value);
    json!({
        "config_path": path.display().to_string(),
        "config_absent": created_or_absent && !path.exists(),
        "created": created_or_absent && path.exists(),
        "semantic_search": {
            "enabled": semantic.enabled,
            "provider": semantic.provider.as_str(),
            "allow_live_provider": semantic.allow_live_provider,
            "endpoint_url": semantic.endpoint_url,
            "model_id": semantic.model_id,
            "dimensions": semantic.dimensions,
            "api_key_env": semantic.api_key_env,
            "api_key_present": has_key,
            "timeout_seconds": semantic.timeout_seconds,
            "session_token_ceiling": semantic.session_token_ceiling,
            "provider_available": provider_available,
            "provider_error": provider_error.unwrap_or(Value::Null),
            "next_action": semantic_next_action(semantic, has_key, vector_count),
        },
        "embeddings_sidecar": {
            "path": sidecar_path.display().to_string(),
            "present": sidecar_path.exists(),
            "vector_count": vector_count,
            "error": sidecar_count.err(),
        }
    })
}

fn semantic_provider_error(semantic: &SemanticSearchConfig, has_key: bool) -> Option<String> {
    if !semantic.enabled {
        return Some("semantic_search.enabled is false".to_owned());
    }
    match semantic.provider {
        SemanticProviderKind::Api if !semantic.allow_live_provider => {
            Some("hosted API provider requires allow_live_provider=true".to_owned())
        }
        SemanticProviderKind::Api if !has_key => Some(format!(
            "hosted API provider requires non-empty ${}",
            semantic.api_key_env
        )),
        SemanticProviderKind::LocalOpenAi => semantic
            .validate_endpoint_trust()
            .err()
            .map(|err| err.to_string()),
        SemanticProviderKind::Api => None,
    }
}

fn semantic_next_action(
    semantic: &SemanticSearchConfig,
    has_key: bool,
    vector_count: Option<i64>,
) -> String {
    if !semantic.enabled {
        return "enable semantic search, then run `loomweave analyze`".to_owned();
    }
    match semantic.provider {
        SemanticProviderKind::Api if !semantic.allow_live_provider => {
            return "set semantic_search.allow_live_provider=true for hosted API calls".to_owned();
        }
        SemanticProviderKind::Api if !has_key => {
            return format!("export ${} before analyzing", semantic.api_key_env);
        }
        SemanticProviderKind::LocalOpenAi if vector_count.unwrap_or(0) == 0 => {
            return format!(
                "start the local embeddings server at {}, then run `loomweave analyze`",
                semantic.endpoint_url
            );
        }
        _ => {}
    }
    if vector_count.unwrap_or(0) == 0 {
        "run `loomweave analyze` to populate semantic embeddings".to_owned()
    } else {
        "reconnect/restart `loomweave serve` after config changes".to_owned()
    }
}

fn semantic_sidecar_count(path: &Path) -> std::result::Result<Option<i64>, String> {
    if !path.exists() {
        return Ok(None);
    }
    let conn = Connection::open(path).map_err(|err| err.to_string())?;
    conn.query_row("SELECT COUNT(*) FROM entity_embeddings", [], |row| {
        row.get(0)
    })
    .optional()
    .map_err(|err| err.to_string())
}

fn with_active_session_policy(
    mut status: Value,
    active_write_tools: bool,
    restart_required: bool,
) -> Value {
    if let Some(object) = status.as_object_mut() {
        object.insert(
            "active_session".to_owned(),
            json!({
                "enable_write_tools": active_write_tools,
                "restart_required": restart_required,
                "note": if restart_required {
                    "reconnect/restart loomweave serve for provider and write-tool policy changes to load"
                } else {
                    "active session state"
                }
            }),
        );
    }
    status
}

fn optional_confidence(
    arguments: &serde_json::Map<String, Value>,
) -> std::result::Result<EdgeConfidence, ParamError> {
    let Some(value) = arguments.get("confidence") else {
        return Ok(EdgeConfidence::Resolved);
    };
    match value.as_str() {
        Some("resolved") => Ok(EdgeConfidence::Resolved),
        Some("ambiguous") => Ok(EdgeConfidence::Ambiguous),
        Some("inferred") => Ok(EdgeConfidence::Inferred),
        Some(_) => Err(ParamError::new(
            "confidence must be one of resolved, ambiguous, inferred",
        )),
        None => Err(ParamError::new("confidence must be a string")),
    }
}

/// Call-graph blind spots a query did **not** search, so a consumer never reads
/// an empty or partial caller/path result as a true negative (clarion-0d204a3f16).
/// The static resolver cannot bind a call made through an attribute receiver
/// (e.g. `ctx.orchestrator.resume()`); only `inferred` (LLM) dispatch attempts
/// those, so `resolved`/`ambiguous` queries exclude them and `inferred` does not.
///
/// This is the base vocabulary, used directly by `entity_call_site_list`, which
/// SEARCHES the unresolved-call-site table (returning `unresolved_sites`
/// evidence rows) and so must not declare it as an unsearched blind spot.
/// Navigation tools use [`navigation_scope_excludes`].
fn call_graph_scope_excludes(confidence: EdgeConfidence) -> Vec<&'static str> {
    match confidence {
        EdgeConfidence::Resolved | EdgeConfidence::Ambiguous => vec!["attribute-receiver-calls"],
        EdgeConfidence::Inferred => Vec::new(),
    }
}

/// Scope excludes for the caller-navigation surface (`entity_callers_list`,
/// `entity_neighborhood_get`, `entity_execution_path_list`, the orientation
/// pack). Unlike `entity_call_site_list`, these read only resolved/ambiguous
/// `calls` edges — when the project holds live unresolved call sites, that
/// whole category is an unsearched blind spot and must be named, or an empty
/// callers list reads as a confident true negative (clarion-df87b4f381).
/// `inferred` stays empty: its dispatch pass attempts the unresolved category.
pub(crate) fn navigation_scope_excludes(
    confidence: EdgeConfidence,
    live_unresolved_sites: bool,
) -> Vec<&'static str> {
    let mut excludes = call_graph_scope_excludes(confidence);
    if live_unresolved_sites && confidence != EdgeConfidence::Inferred {
        excludes.push("unresolved-static-calls");
    }
    excludes
}

/// Per-query scope excludes for `entity_callers_list` / `entity_neighborhood_get`
/// (clarion-76c31b730a). Unlike [`navigation_scope_excludes`] (which flags the
/// whole attribute-receiver category on every resolved/ambiguous read), this is
/// populated ONLY when THIS traversal actually skipped a candidate — i.e. a
/// live unresolved call site whose textual callee name-matches the target
/// (`unresolved_name_matches > 0`). When nothing was skipped it returns empty,
/// and the caller pairs that with `traversal_complete: true` so an empty
/// `callers` list reads as a true negative rather than an untrustable blind
/// spot. `inferred` stays empty: its dispatch pass attempts the unresolved
/// category, so it skips nothing.
fn caller_navigation_scope_excludes(
    confidence: EdgeConfidence,
    skipped_a_candidate: bool,
) -> Vec<&'static str> {
    if !skipped_a_candidate || confidence == EdgeConfidence::Inferred {
        return Vec::new();
    }
    // A skipped candidate is, by construction, an attribute-receiver or dynamic
    // call site the static resolver could not bind — name both blind-spot
    // categories that the traversal therefore did not search.
    vec!["attribute-receiver-calls", "unresolved-static-calls"]
}

/// The `unresolved_name_matches` count + `next_action` recovery pointer for a
/// caller-navigation result: how many live unresolved call sites name-match
/// `target`, and where to see them. The pointer names `entity_call_site_list`
/// because it works in the default read-only posture, where `confidence=
/// inferred` is rejected by the MCP tool policy (clarion-df87b4f381).
///
/// NOTE (clarion-2b87cd7a59 A1): unlike `unresolved_candidates`, the count and
/// pointer are *not* gated behind `confidence != Inferred`. This is intentional,
/// not the contradiction A1 closed. `unresolved_candidates` is a completeness
/// claim — listing it alongside `traversal_complete: true` would assert "these
/// were skipped" about a traversal that skipped nothing, so it is gated. The
/// count/pointer are purely *informational*: "N name-matching sites exist; here
/// is how to inspect them" stays true whether or not the inferred dispatch bound
/// them, and gives the caller a navigation handle without claiming incompleteness.
pub(crate) fn unresolved_match_fields(
    conn: &rusqlite::Connection,
    target: &EntityRow,
) -> Result<(i64, Value), StorageError> {
    let count = unresolved_caller_count_for_target(conn, target)?;
    let next_action = if count > 0 {
        json!(format!(
            "{count} unresolved call site(s) name-match this entity and are NOT in `callers`; \
             list them with entity_call_site_list id={id} role=callee",
            id = target.id
        ))
    } else {
        Value::Null
    };
    Ok((count, next_action))
}

/// Per-query honesty for the caller-navigation surface (clarion-76c31b730a):
/// the actual call sites that textually name-match `target` but the static
/// resolver could not bind into `callers` — the in-tool grep-fallback. Each
/// candidate carries `{path, line, callee_text, why}` where `why` splits an
/// attribute/method receiver (`obj.target(...)`, i.e. the dotted `callee_text`)
/// from a bare dynamic dispatch (`target(...)` resolved through a variable).
///
/// This is the evidence behind a non-empty `scope_excludes` / `traversal_complete:
/// false`: when this returns rows, the traversal skipped real candidates, so an
/// empty `callers` list is NOT a true negative. Returns `[]` when nothing was
/// skipped — paired with `traversal_complete: true`, that confirms every
/// candidate was searched. Mirrors `build_call_sites`'s disclosure guards: a
/// briefing-blocked owner's `line_text` is redacted, never the file content read.
pub(crate) fn build_unresolved_candidates(
    conn: &rusqlite::Connection,
    target: &EntityRow,
) -> Result<Vec<Value>, StorageError> {
    let sites = unresolved_callers_for_target(conn, target, CALL_SITES_MAX)?;
    let mut owner_meta: HashMap<String, OwnerMeta> = HashMap::new();
    let mut file_content: HashMap<String, Option<Vec<u8>>> = HashMap::new();
    let mut candidates = Vec::with_capacity(sites.len());
    for site in sites {
        let owner = resolve_owner(conn, &mut owner_meta, &site.caller_entity_id)?;
        let anchor = anchor_line(&mut file_content, owner, Some(site.source_byte_start));
        // A dotted callee text is an attribute/method receiver the static
        // resolver cannot bind (e.g. `ctx.svc.target()`); anything else is a
        // bare name resolved dynamically (through a variable / dispatch table).
        let why = if site.callee_expr.contains('.') {
            "attribute-receiver"
        } else {
            "dynamic"
        };
        candidates.push(json!({
            "path": owner.path,
            "line": anchor.line,
            "callee_text": site.callee_expr,
            "why": why,
        }));
    }
    Ok(candidates)
}

fn envelope_from_storage_result(result: Result<Value, StorageError>) -> Value {
    match result {
        Ok(result) => success_envelope(result),
        Err(err) => tool_error_envelope(
            McpErrorCode::StorageError,
            &err.to_string(),
            storage_retryable(&err),
        ),
    }
}

/// Fail-soft scalar count for `project_status`: a query hiccup degrades to 0
/// (logged), never failing the diagnostics tool. Matches `snapshot.rs`'s policy.
fn scalar_count_fail_soft(conn: &rusqlite::Connection, sql: &str) -> i64 {
    conn.query_row(sql, [], |row| row.get::<_, i64>(0))
        .unwrap_or_else(|err| {
            tracing::warn!(error = %err, sql, "project_status count query failed; reporting 0");
            0
        })
}

/// Per-plugin entity counts from the current index (`entities.plugin_id`), the
/// queryable proxy for "which plugins produced this graph." Fail-soft: any
/// query error degrades to an empty array (logged).
fn plugin_entity_counts(conn: &rusqlite::Connection) -> Value {
    let mut stmt = match conn
        .prepare("SELECT plugin_id, COUNT(*) FROM entities GROUP BY plugin_id ORDER BY plugin_id")
    {
        Ok(stmt) => stmt,
        Err(err) => {
            tracing::warn!(error = %err, "project_status plugin-count prepare failed");
            return Value::Array(Vec::new());
        }
    };
    let rows = stmt.query_map([], |row| {
        Ok(json!({
            "plugin_id": row.get::<_, String>(0)?,
            "entity_count": row.get::<_, i64>(1)?,
        }))
    });
    match rows {
        Ok(mapped) => Value::Array(mapped.filter_map(Result::ok).collect()),
        Err(err) => {
            tracing::warn!(error = %err, "project_status plugin-count query failed");
            Value::Array(Vec::new())
        }
    }
}

/// The most-recent run by `started_at`, regardless of terminal status, so a
/// `skipped_no_plugins` or `failed` run is visible (not just the last
/// `completed` one). Fail-soft: no rows or a query error → `null`.
fn latest_run_row(conn: &rusqlite::Connection) -> Value {
    match conn.query_row(
        "SELECT id, status, started_at, completed_at, owner_pid, heartbeat_at, \
            analyzed_at_commit FROM runs \
         ORDER BY started_at DESC LIMIT 1",
        [],
        |row| {
            Ok(json!({
                "id": row.get::<_, String>(0)?,
                "status": row.get::<_, String>(1)?,
                "started_at": row.get::<_, String>(2)?,
                "completed_at": row.get::<_, Option<String>>(3)?,
                "owner_pid": row.get::<_, Option<i64>>(4)?,
                "heartbeat_at": row.get::<_, Option<String>>(5)?,
                "analyzed_at_commit": row.get::<_, Option<String>>(6)?,
            }))
        },
    ) {
        Ok(value) => value,
        Err(rusqlite::Error::QueryReturnedNoRows) => Value::Null,
        Err(err) => {
            tracing::warn!(error = %err, "project_status latest-run query failed");
            Value::Null
        }
    }
}

fn flatten_storage_envelope_result(result: Result<Value, StorageError>) -> Value {
    match result {
        Ok(envelope) => envelope,
        Err(err) => tool_error_envelope(
            McpErrorCode::StorageError,
            &err.to_string(),
            storage_retryable(&err),
        ),
    }
}

/// `storage-error` retryable hint. FK violations are deterministic against
/// the same row set; everything else (`SQLITE_BUSY`, disk-full, pool errors)
/// stays retryable (clarion-df58379de4).
fn storage_retryable(err: &StorageError) -> bool {
    !err.is_foreign_key_violation()
}

fn success_envelope(result: Value) -> Value {
    success_envelope_with_truncation(result, None)
}

fn success_envelope_with_truncation(result: Value, truncation_reason: Option<&str>) -> Value {
    let mut envelope = serde_json::Map::new();
    envelope.insert("ok".to_owned(), Value::Bool(true));
    envelope.insert("result".to_owned(), result);
    envelope.insert("error".to_owned(), Value::Null);
    envelope.insert("diagnostics".to_owned(), Value::Array(Vec::new()));
    envelope.insert(
        "truncated".to_owned(),
        Value::Bool(truncation_reason.is_some()),
    );
    envelope.insert(
        "truncation_reason".to_owned(),
        truncation_reason.map_or(Value::Null, |reason| Value::String(reason.to_owned())),
    );
    envelope.insert("stats_delta".to_owned(), json!({}));
    Value::Object(envelope)
}

fn success_envelope_with_truncation_and_stats(
    result: Value,
    truncation_reason: Option<&str>,
    stats_delta: Value,
) -> Value {
    let mut envelope = success_envelope_with_truncation(result, truncation_reason);
    if let Some(object) = envelope.as_object_mut() {
        object.insert("stats_delta".to_owned(), stats_delta);
    }
    envelope
}

fn success_envelope_with_stats(result: Value, stats_delta: Value) -> Value {
    let mut envelope = success_envelope(result);
    if let Some(object) = envelope.as_object_mut() {
        object.insert("stats_delta".to_owned(), stats_delta);
    }
    envelope
}

fn tool_error_envelope(code: McpErrorCode, message: &str, retryable: bool) -> Value {
    tool_error_envelope_with_diagnostics(code, message, retryable, json!({}), Vec::new())
}

/// `EntityNotFound` envelope echoing the caller's original input string. Used by
/// the id-or-SEI canonicalize-at-top path so an unresolvable SEI reports the
/// exact token the caller pasted (clarion-d76e7f7267).
fn entity_not_found_envelope(requested_id: &str) -> Value {
    tool_error_envelope(
        McpErrorCode::EntityNotFound,
        &format!("entity {requested_id} was not found"),
        false,
    )
}

fn tool_error_envelope_with_diagnostics(
    code: McpErrorCode,
    message: &str,
    retryable: bool,
    stats_delta: Value,
    diagnostics: Vec<Value>,
) -> Value {
    let mut envelope = serde_json::Map::new();
    envelope.insert("ok".to_owned(), Value::Bool(false));
    envelope.insert("result".to_owned(), Value::Null);
    envelope.insert(
        "error".to_owned(),
        json!({
            "code": code.as_str(),
            "message": message,
            "retryable": retryable,
        }),
    );
    envelope.insert("diagnostics".to_owned(), Value::Array(diagnostics));
    envelope.insert("truncated".to_owned(), Value::Bool(false));
    envelope.insert("truncation_reason".to_owned(), Value::Null);
    envelope.insert("stats_delta".to_owned(), stats_delta);
    Value::Object(envelope)
}

fn llm_usage_json(response: &LlmResponse) -> Value {
    json!({
        "tokens_input": response.input_tokens,
        "tokens_cached_input": response.cached_input_tokens,
        "tokens_output": response.output_tokens,
        "tokens_total": response.total_tokens,
        "cost_usd": response.cost_usd
    })
}

fn summary_usage_stats(response: &LlmResponse, invalid_json: bool) -> Value {
    let mut stats = serde_json::Map::new();
    stats.insert("summary_cache_misses_total".to_owned(), json!(1));
    stats.insert(
        "summary_tokens_input".to_owned(),
        json!(response.input_tokens),
    );
    stats.insert(
        "summary_tokens_cached_input".to_owned(),
        json!(response.cached_input_tokens),
    );
    stats.insert(
        "summary_tokens_output".to_owned(),
        json!(response.output_tokens),
    );
    stats.insert(
        "summary_tokens_total".to_owned(),
        json!(response.total_tokens),
    );
    stats.insert("summary_cost_usd".to_owned(), json!(response.cost_usd));
    if invalid_json {
        stats.insert("llm_invalid_json_total".to_owned(), json!(1));
    }
    Value::Object(stats)
}

fn inferred_usage_stats(response: &LlmResponse, invalid_json: bool) -> Value {
    let mut stats = serde_json::Map::new();
    stats.insert("inferred_dispatch_misses_total".to_owned(), json!(1));
    stats.insert(
        "inferred_tokens_input".to_owned(),
        json!(response.input_tokens),
    );
    stats.insert(
        "inferred_tokens_cached_input".to_owned(),
        json!(response.cached_input_tokens),
    );
    stats.insert(
        "inferred_tokens_output".to_owned(),
        json!(response.output_tokens),
    );
    stats.insert(
        "inferred_tokens_total".to_owned(),
        json!(response.total_tokens),
    );
    stats.insert("inferred_cost_usd".to_owned(), json!(response.cost_usd));
    if invalid_json {
        stats.insert("llm_invalid_json_total".to_owned(), json!(1));
    }
    Value::Object(stats)
}

fn token_ceiling_envelope(message: &str) -> Value {
    json!({
        "ok": false,
        "result": null,
        "error": {
            "code": McpErrorCode::TokenCeilingExceeded.as_str(),
            "message": message,
            "retryable": false
        },
        "diagnostics": [
            {
                "code": "LMWV-LLM-TOKEN-CEILING-EXCEEDED",
                "message": message
            }
        ],
        "truncated": false,
        "truncation_reason": null,
        "stats_delta": {
            "token_ceiling_exceeded_total": 1
        }
    })
}

/// `reason` is a degraded-result discriminant for the `issues_for` tool
/// (`filigree-disabled`, `entity-not-found`, `filigree-unreachable`,
/// `filigree-client-error`), NOT a member of the `McpErrorCode` error-code
/// vocabulary. It lives on a `success_envelope` (`available: false`), not an
/// error envelope. `entity-not-found` here coincidentally matches
/// `McpErrorCode::EntityNotFound`'s wire spelling but is intentionally a bare
/// string: most of this closed set are not error codes, and this surface has
/// its own consumers, so the two axes are kept independent (see ADR-037).
fn issues_unavailable(filigree_endpoint: &Value, reason: &str, message: &str) -> Value {
    success_envelope(json!({
        "available": false,
        "result_kind": "unavailable",
        "filigree_endpoint": filigree_endpoint,
        "reason": reason,
        "message": message,
        "matched": [],
        "drifted": [],
        "not_found": []
    }))
}

/// The degraded `wardline_findings` section returned when the findings cannot
/// be fetched (transport/HTTP error) or the blocking task panicked. Single
/// source of truth for the four-key `unavailable` shape.
fn wardline_unavailable(reason: &str) -> Value {
    serde_json::json!({
        "result_kind": "unavailable",
        "items": [],
        "omitted_no_qualname": 0,
        "reason": reason,
    })
}

/// Build the `wardline_findings` enrich section for one entity. Enrich-only:
/// a fetch error degrades to `result_kind: "unavailable"` rather than failing
/// the tool.
fn wardline_section_for_entity(
    client: &std::sync::Arc<dyn crate::filigree::FiligreeLookup>,
    project_root: &Path,
    entity_id: &str,
    source_file_path: Option<&str>,
) -> Value {
    let Some(path) = source_file_path else {
        return serde_json::json!({ "result_kind": "no_matches", "items": [], "omitted_no_qualname": 0 });
    };
    let path = match project_relative_lookup_path(project_root, path) {
        Ok(path) => path,
        Err(err) => {
            return wardline_unavailable(&format!(
                "cannot normalize source_file_path for Filigree lookup: {err}"
            ));
        }
    };
    match client.wardline_findings_for_path(&path) {
        Ok(findings) => {
            let result = crate::wardline_reconcile::reconcile_for_entity(entity_id, findings);
            let items: Vec<Value> = result
                .matched
                .iter()
                .map(|m| {
                    serde_json::json!({
                        "rule_id": m.finding.rule_id,
                        "message": m.finding.message,
                        "severity": m.finding.severity,
                        "status": m.finding.status,
                        "line_start": m.finding.line_start,
                        "line_end": m.finding.line_end,
                        "fingerprint": m.finding.fingerprint,
                        "resolution_confidence": m.resolution_confidence,
                        "wardline": m.finding.metadata.get("wardline").cloned().unwrap_or(Value::Null),
                    })
                })
                .collect();
            let result_kind = if items.is_empty() {
                "no_matches"
            } else {
                "matched"
            };
            serde_json::json!({
                "result_kind": result_kind,
                "items": items,
                "omitted_no_qualname": result.omitted_no_qualname,
            })
        }
        Err(err) => wardline_unavailable(&err.to_string()),
    }
}

pub(crate) fn project_relative_lookup_path(
    project_root: &Path,
    source_file_path: &str,
) -> Result<String, String> {
    let root = project_root
        .canonicalize()
        .map_err(|err| format!("canonicalize project root: {err}"))?;
    let input = Path::new(source_file_path);
    let absolute = if input.is_absolute() {
        normalize_path_lexically(input)
    } else {
        normalize_path_lexically(&root.join(input))
    };
    if !absolute.starts_with(&root) {
        return Err(format!(
            "{source_file_path:?} escapes project root {}",
            root.display()
        ));
    }
    let relative = absolute
        .strip_prefix(&root)
        .map_err(|err| format!("strip project root: {err}"))?;
    let mut parts = Vec::new();
    for component in relative.components() {
        match component {
            Component::Normal(part) => {
                let Some(part) = part.to_str() else {
                    return Err(format!("{source_file_path:?} is not valid UTF-8"));
                };
                parts.push(part.to_owned());
            }
            Component::CurDir => {}
            _ => {
                return Err(format!(
                    "{source_file_path:?} is not a clean project-relative path"
                ));
            }
        }
    }
    if parts.is_empty() {
        return Err(format!("{source_file_path:?} does not name a file path"));
    }
    Ok(parts.join("/"))
}

fn normalize_path_lexically(path: &Path) -> PathBuf {
    let mut normalized = PathBuf::new();
    for component in path.components() {
        match component {
            Component::CurDir => {}
            Component::ParentDir => {
                normalized.pop();
            }
            other => normalized.push(other.as_os_str()),
        }
    }
    normalized
}

fn association_json(
    association: &EntityAssociation,
    entity: Option<&Value>,
    current_content_hash: Option<&str>,
    drift_status: &str,
    canonical_entity_id: Option<&str>,
) -> Value {
    let entity_id = canonical_entity_id.unwrap_or(&association.loomweave_entity_id);
    let mut value = json!({
        "issue_id": association.issue_id,
        "entity_id": entity_id,
        "entity": entity,
        "content_hash_at_attach": association.content_hash_at_attach,
        "current_content_hash": current_content_hash,
        "attached_at": association.attached_at,
        "attached_by": association.attached_by,
        "drift_status": drift_status
    });
    if entity_id != association.loomweave_entity_id
        && let Some(object) = value.as_object_mut()
    {
        object.insert(
            "association_entity_id".to_owned(),
            json!(association.loomweave_entity_id),
        );
    }
    value
}

fn summary_read_error(read: SummaryRead) -> Value {
    match read {
        SummaryRead::EntityNotFound(id) => tool_error_envelope(
            McpErrorCode::EntityNotFound,
            &format!("entity {id} was not found"),
            false,
        ),
        SummaryRead::MissingContentHash(id) => tool_error_envelope(
            McpErrorCode::ContentHashMissing,
            &format!("entity {id} has no content hash for summary cache keying"),
            false,
        ),
        SummaryRead::ScopeDeferred(entity_json) => summary_scope_deferred(&entity_json),
        SummaryRead::BriefingBlocked(entity_json, reason) => {
            summary_briefing_blocked(&entity_json, &reason)
        }
        SummaryRead::Ready(_) => unreachable!("ready summary read is not an error"),
    }
}

#[derive(Debug)]
struct SourceExcerptError {
    entity_id: String,
    stored_content_hash: String,
    current_content_hash: String,
}

impl SourceExcerptError {
    fn message(&self) -> String {
        format!(
            "entity {} source content drifted: stored content_hash {} but current file hashes to {}; rerun `loomweave analyze` before requesting LLM output",
            self.entity_id, self.stored_content_hash, self.current_content_hash
        )
    }

    fn to_envelope(&self) -> Value {
        tool_error_envelope(McpErrorCode::ContentDrift, &self.message(), false)
    }

    fn to_inferred_failure(&self) -> InferredDispatchFailure {
        InferredDispatchFailure::new(McpErrorCode::ContentDrift, &self.message(), false)
    }
}

/// Deterministic structural summary used when the LLM returns non-JSON
/// (clarion-ed246ca3aa). Carries the entity's identity plus a bounded head of
/// its own source (where the signature and docstring live), so the caller gets
/// usable orientation instead of a billed error. `kind: structural-fallback`
/// lets consumers distinguish it from a real LLM summary.
fn structural_summary_json(entity: &EntityRow, source_excerpt: &str) -> String {
    const STRUCTURAL_SUMMARY_MAX_CHARS: usize = 1200;
    let mut source_head: String = source_excerpt
        .chars()
        .take(STRUCTURAL_SUMMARY_MAX_CHARS)
        .collect();
    let source_truncated = source_excerpt.chars().count() > STRUCTURAL_SUMMARY_MAX_CHARS;
    if source_truncated {
        source_head.push('…');
    }
    serde_json::to_string(&json!({
        "kind": "structural-fallback",
        "note": "LLM summary unavailable (provider returned non-JSON); deterministic structural summary derived from the entity source.",
        "entity_kind": entity.kind,
        "short_name": entity.short_name,
        "qualified_name": entity.name,
        "source_head": source_head,
        "source_truncated": source_truncated
    }))
    .expect("structural summary serializes")
}

fn summary_success_envelope(
    entity_json: &Value,
    entry: &SummaryCacheEntry,
    cache_hit: bool,
    stale_semantic: bool,
    cached_input_tokens: Option<i64>,
    stats_delta: Value,
) -> Value {
    let summary = serde_json::from_str::<Value>(&entry.summary_json).unwrap_or_else(|_| {
        json!({
            "raw": entry.summary_json
        })
    });
    let mut usage = serde_json::Map::new();
    usage.insert("tokens_input".to_owned(), json!(entry.tokens_input));
    if let Some(tokens) = cached_input_tokens {
        usage.insert("tokens_cached_input".to_owned(), json!(tokens));
    }
    usage.insert("tokens_output".to_owned(), json!(entry.tokens_output));
    usage.insert(
        "tokens_total".to_owned(),
        json!(entry.tokens_input + entry.tokens_output),
    );
    success_envelope_with_stats(
        json!({
            "available": true,
            "entity": entity_json,
            "summary": summary,
            "cache": {
                "hit": cache_hit,
                "prompt_template_id": entry.key.prompt_template_id,
                "model_id": entry.key.model_tier,
                "guidance_fingerprint": entry.key.guidance_fingerprint,
                "stale_semantic": stale_semantic,
                "created_at": entry.created_at,
                "last_accessed_at": entry.last_accessed_at
            },
            "usage": Value::Object(usage)
        }),
        stats_delta,
    )
}

fn summary_scope_deferred(entity_json: &Value) -> Value {
    success_envelope(json!({
        "available": false,
        "reason": "summary-scope-deferred",
        "message": "subsystem summaries are deferred to v0.2",
        "entity": entity_json
    }))
}

fn summary_briefing_blocked(entity_json: &Value, reason: &str) -> Value {
    let remediation = briefing_block_remediation(reason);
    let entity_id = entity_json
        .get("id")
        .and_then(Value::as_str)
        .map(str::to_owned);
    success_envelope(json!({
        "available": false,
        "entity_id": entity_id,
        "entity": entity_json,
        "summary": null,
        "briefing_blocked": reason,
        "remediation": remediation
    }))
}

fn briefing_block_reason(entity: &EntityRow) -> Option<String> {
    entity_briefing_block_reason(&entity.properties_json)
}

fn tool_json_rpc_response(id: &Value, envelope: &Value) -> Value {
    let is_error = !envelope
        .get("ok")
        .and_then(Value::as_bool)
        .unwrap_or_default();
    result_response(
        id,
        &json!({
            "content": [
                {
                    "type": "text",
                    "text": serde_json::to_string(&envelope).expect("tool envelope serializes")
                }
            ],
            "isError": is_error
        }),
    )
}

/// Serialize an entity's stable identity fields, without the SEI.
///
/// This is the conn-free core used both by [`entity_json`] (which adds the SEI
/// for client-facing tool responses) and by internal payloads — notably the LLM
/// inference prompt — that must *not* gain a `sei` field (it is neither a tool
/// return surface nor allowed to change shape; see REQ-C-04 scope).
fn entity_identity_json(entity: &EntityRow) -> Value {
    json!({
        "id": entity.id,
        "kind": entity.kind,
        "name": entity.name,
        "short_name": entity.short_name,
        "source_file_path": entity.source_file_path,
        "source_line_start": entity.source_line_start,
        "source_line_end": entity.source_line_end,
        "content_hash": entity.content_hash
    })
}

/// Serialize an entity for an MCP tool response.
///
/// Per REQ-C-04 / ADR-038, every tool surface that returns an entity `id` must
/// also carry its SEI: the `id` (locator) is the *mutable* address that changes
/// on rename/move, while the `sei` is the durable cross-tool binding key. The
/// SEI is read via a read-time join (`sei_for_locator`) and graceful-degrades
/// to JSON `null` on a pre-SEI database or an orphaned/unbound locator — the
/// lookup must never fail the tool call.
fn entity_json(conn: &rusqlite::Connection, entity: &EntityRow) -> Value {
    // A secret-scan-blocked entity (ADR-013) must not have its identity disclosed
    // by a discovery/structure MCP read — matching the federation read API, whose
    // BRIEFING_BLOCKED response omits id/name/path/hash (ADR-034). This is the
    // single choke point: every list/structure surface projects entities (and
    // their caller/callee/reference/import neighbors) through here
    // (clarion-307668e2be). The deliberate exception — `summary`, which echoes a
    // caller-named entity's identity + remediation — builds identity via
    // `entity_identity_json` instead, bypassing this gate.
    if let Some(reason) = briefing_block_reason(entity) {
        return blocked_entity_stub(entity, &reason);
    }
    let mut value = entity_identity_json(entity);
    if let Some(object) = value.as_object_mut() {
        object.insert(
            "sei".to_owned(),
            json!(sei_for_locator(conn, &entity.id).ok().flatten()),
        );
        // Inline the entity's own categorisation tags (clarion-057ff2b330) so a
        // tool response shows them without a reverse-index `entity_tag_list`
        // round-trip. Tags already live in `entity_tags`; this is read-only.
        // Graceful-degrade to `[]` on any lookup error — a tag read must never
        // fail the structural tool call (same posture as the SEI join above).
        object.insert(
            "tags".to_owned(),
            json!(tags_for_entity(conn, &entity.id).unwrap_or_default()),
        );
    }
    value
}

/// Shannon entropy (bits/byte) of a string, used by the briefing-block guard to
/// detect the rare case where an entity *name* is itself a high-entropy token
/// (e.g. a generated symbol embedding a secret). Mirrors the pre-ingest
/// scanner's measure (`loomweave-scanner::entropy`) but kept local: the scan ran
/// at ingest, this is only a read-path safety net for an identity field.
fn name_entropy(text: &str) -> f64 {
    let bytes = text.as_bytes();
    if bytes.is_empty() {
        return 0.0;
    }
    let mut counts = [0u32; 256];
    for &b in bytes {
        counts[b as usize] += 1;
    }
    let len = f64::from(u32::try_from(bytes.len()).unwrap_or(u32::MAX));
    counts
        .iter()
        .filter(|&&c| c > 0)
        .map(|&c| {
            let p = f64::from(c) / len;
            -p * p.log2()
        })
        .sum()
}

/// Entropy threshold above which a path *segment* is itself treated as secret
/// and its containing field re-withheld. Real base64 secret blobs measure
/// ~4.6–4.8 bits/byte (an AWS-key-shaped blob ≈ 4.63); ordinary identifier
/// segments — even verbose `snake_case` like `function_with_a_long_name` — sit
/// below ~4.2, so this separates the two populations with margin. The length
/// floor avoids flagging short-but-varied names.
const NAME_ENTROPY_REDACT_THRESHOLD: f64 = 4.5;
const NAME_ENTROPY_MIN_LEN: usize = 20;

/// Whether a single identity *value* (`name`, `short_name`, or the locator `id`)
/// is itself high-entropy enough to be a secret, so the A3 identity-preserving
/// projection must keep redacting that one field (clarion-719e7320f5 guard).
///
/// The value is judged PER PATH SEGMENT, splitting on `:`/`.` (so `::` yields
/// empty segments the length floor discards). A `plugin:kind:qualname` locator
/// and a dotted/`::` qualname are separator-delimited, so a secret embedded as
/// ONE segment (`python:function:pkg.<blob>`) trips the guard even though the
/// surrounding segments are ordinary words — while a long *descriptive*
/// identifier stays navigable because none of its word-segments is high-entropy.
/// Checking per-segment (not the whole string) is what closes the qualified-name
/// hole: the blob's entropy is not diluted by the `pkg.`/`module::` prefix.
fn name_value_is_secretlike(text: &str) -> bool {
    text.split([':', '.']).any(|segment| {
        segment.len() >= NAME_ENTROPY_MIN_LEN
            && name_entropy(segment) >= NAME_ENTROPY_REDACT_THRESHOLD
    })
}

/// Project one identity *value* for a briefing-blocked row: the value verbatim,
/// unless it is itself high-entropy (a generated symbol embedding a secret), in
/// which case that single field is re-withheld to JSON `null` (A3 guard).
fn redact_secretlike(value: &str) -> Value {
    if name_value_is_secretlike(value) {
        Value::Null
    } else {
        json!(value)
    }
}

/// The projection of a briefing-blocked entity (ADR-013 secret scan).
///
/// Per clarion-719e7320f5 (A3) the briefing-block flag rides ALONGSIDE a real,
/// navigable identity — `id`/`kind`/`name`/`short_name`/`source_file_path`/line
/// span/`content_hash` are all KEPT. The secret is the file *content*, not the
/// entity's structural identity (`project_finding_list` already prints those same
/// paths), so only the secret-bearing content (summary/source/docstring) is
/// withheld — and those are never part of this projection to begin with. The
/// cross-tool `sei` binding key stays null (ADR-034): a blocked row is navigable
/// by locator but not bound across siblings.
///
/// Guard: in the rare case where a `name`/`short_name`/`id` is *itself*
/// high-entropy (a generated symbol embedding a secret), that single field is
/// re-withheld — the rest of the identity still rides along.
fn blocked_entity_stub(entity: &EntityRow, reason: &str) -> Value {
    json!({
        "id": redact_secretlike(&entity.id),
        "sei": Value::Null,
        "kind": entity.kind,
        "name": redact_secretlike(&entity.name),
        "short_name": redact_secretlike(&entity.short_name),
        "source_file_path": entity.source_file_path,
        "source_line_start": entity.source_line_start,
        "source_line_end": entity.source_line_end,
        "content_hash": entity.content_hash,
        "briefing_blocked": reason,
    })
}

/// Placeholder substituted for a briefing-blocked entity's id in execution-path
/// arrays. Distinct blocked nodes collapse to this one token (uncorrelatable by
/// design — the accepted price of withholding identity, clarion-307668e2be).
const BRIEFING_BLOCKED_PATH_SENTINEL: &str = "[briefing-blocked]";

/// Operator-facing remediation for a briefing block, by reason. Shared by every
/// refusal envelope (summary / neighborhood / orientation) so the "fix the
/// secret" guidance stays consistent.
fn briefing_block_remediation(reason: &str) -> &'static str {
    if reason == "unscanned_source" {
        "Entity source file was not covered by the pre-ingest secret scan. Re-run with scanner coverage for that path or fix the plugin source path before requesting a summary."
    } else {
        "File flagged by pre-ingest secret scan. Fix the secret or whitelist via .weft/loomweave/secrets-baseline.yaml. See ADR-013."
    }
}

/// Refusal envelope for a structure-fan-out read (`neighborhood`,
/// `orientation`) whose queried entity is itself briefing-blocked. Withholds the
/// structure *around* the withheld entity (ADR-034) and discloses no identity.
fn blocked_entity_refusal(reason: &str) -> Value {
    success_envelope(json!({
        "available": false,
        "briefing_blocked": reason,
        "remediation": briefing_block_remediation(reason),
    }))
}

fn entity_properties_json(entity: &EntityRow) -> Value {
    serde_json::from_str::<Value>(&entity.properties_json)
        .expect("entity properties_json should be valid JSON")
}

/// Maximum number of ambiguity alternatives `entity_at` reports. Genuine
/// same-granularity overlaps are rare; this only bounds a pathological file.
const ENTITY_CONTEXT_MAX_ALTERNATIVES: usize = 8;

/// The decorator/declaration/body sub-ranges a plugin records for a
/// function/class entity in `properties_json.definition` (clarion-460def6a51).
/// All optional: older indexes, modules, and plugins that don't emit the block
/// leave every field `None`.
struct DefinitionSpan {
    decl_line: Option<i64>,
    body_line_start: Option<i64>,
    decorator_line_start: Option<i64>,
    decorator_line_end: Option<i64>,
}

impl DefinitionSpan {
    fn from_entity(entity: &EntityRow) -> Self {
        let def = serde_json::from_str::<Value>(&entity.properties_json)
            .ok()
            .and_then(|props| props.get("definition").cloned());
        let get = |key: &str| -> Option<i64> {
            def.as_ref()
                .and_then(|d| d.get(key))
                .and_then(Value::as_i64)
        };
        Self {
            decl_line: get("decl_line"),
            body_line_start: get("body_line_start"),
            decorator_line_start: get("decorator_line_start"),
            decorator_line_end: get("decorator_line_end"),
        }
    }
}

/// Classify *why* `line` resolved to `entity`: a decorator line, the
/// declaration/signature, the body, or merely a containing scope (file-scope
/// plugin entity, or an entity without recorded sub-ranges). Honest by
/// construction — a blank or comment line that only a file-scope entity spans
/// reports `containing_range`, never a fabricated exact match
/// (clarion-460def6a51 acceptance #3).
fn match_reason_for(line: i64, entity: &EntityRow) -> &'static str {
    if is_plugin_file_scope_entity(entity) {
        return "containing_range";
    }
    let def = DefinitionSpan::from_entity(entity);
    let Some(decl_line) = def.decl_line else {
        return "containing_range";
    };
    if let Some(decorator_start) = def.decorator_line_start
        && line >= decorator_start
        && line < decl_line
    {
        return "decorator_range";
    }
    if let Some(body_line_start) = def.body_line_start {
        if line >= body_line_start {
            return "body_range";
        }
        return "declaration";
    }
    if line == decl_line {
        return "declaration";
    }
    "containing_range"
}

/// Span length in lines used to detect same-granularity ambiguity. `None` when
/// either bound is missing.
fn span_len(entity: &EntityRow) -> Option<i64> {
    Some(entity.source_line_end? - entity.source_line_start?)
}

/// Compact entity descriptor for the containing stack — enough to orient
/// without the full `entity_json` payload.
fn stack_entity_json(conn: &rusqlite::Connection, entity: &EntityRow) -> Value {
    // A blocked entity in the containing stack (the matched node, or a blocked
    // ancestor module) keeps its navigable identity but rides the briefing-block
    // flag (clarion-719e7320f5, A3): the secret is the file content, not the
    // structural identity. The `sei` cross-tool binding key stays null. The
    // high-entropy-name guard re-withholds id/name/short_name only when the
    // value is itself secret-like.
    if let Some(reason) = briefing_block_reason(entity) {
        return json!({
            "id": redact_secretlike(&entity.id),
            "sei": Value::Null,
            "kind": entity.kind,
            "short_name": redact_secretlike(&entity.short_name),
            "name": redact_secretlike(&entity.name),
            "source_line_start": entity.source_line_start,
            "source_line_end": entity.source_line_end,
            "briefing_blocked": reason,
        });
    }
    // REQ-C-04: a containing-stack ancestor is sometimes the ONLY place an
    // entity appears in the response, so it carries its `sei` (the durable
    // binding key) — not just the mutable `id`/locator. Graceful-degrade null.
    json!({
        "id": entity.id,
        "sei": sei_for_locator(conn, &entity.id).ok().flatten(),
        "kind": entity.kind,
        "short_name": entity.short_name,
        "name": entity.name,
        "source_line_start": entity.source_line_start,
        "source_line_end": entity.source_line_end,
    })
}

/// Build the additive `entity_context` evidence block for `entity_at`
/// (clarion-460def6a51): the match reason, the module→entity containing stack,
/// the matched entity's sub-ranges, same-granularity ambiguity alternatives,
/// and index freshness. Returns a `no_match` shell when no entity spans the
/// line (e.g. an unindexed file).
fn entity_context_json(
    conn: &rusqlite::Connection,
    line: Option<i64>,
    matched: Option<&EntityRow>,
    candidates: &[EntityRow],
    ancestors: &[EntityRow],
    snapshot: &crate::snapshot::ProjectSnapshot,
) -> Value {
    let freshness = json!({
        "staleness": snapshot.staleness(),
        "last_analyzed_at": snapshot.last_analyzed_at(),
        "degraded": snapshot.degraded(),
        "scan_truncated": snapshot.scan_truncated(),
    });

    let Some(matched) = matched else {
        return json!({
            "query_line": line,
            "match_reason": "no_match",
            "containing_stack": [],
            "ranges": Value::Null,
            "alternatives": [],
            "freshness": freshness,
        });
    };

    // Containing stack outermost (module) → matched entity, inclusive.
    let mut containing_stack: Vec<Value> = ancestors
        .iter()
        .rev()
        .map(|e| stack_entity_json(conn, e))
        .collect();
    containing_stack.push(stack_entity_json(conn, matched));

    // A blocked matched entity keeps its line span (clarion-719e7320f5, A3): the
    // line numbers are structural identity, not the secret content the block
    // hides — and `project_finding_list` already prints the same file+line. The
    // `briefing_blocked` flag rides alongside.
    let def = DefinitionSpan::from_entity(matched);
    let ranges = if let Some(reason) = briefing_block_reason(matched) {
        json!({
            "source_line_start": matched.source_line_start,
            "source_line_end": matched.source_line_end,
            "decl_line": def.decl_line,
            "body_line_start": def.body_line_start,
            "decorator_line_start": def.decorator_line_start,
            "decorator_line_end": def.decorator_line_end,
            "briefing_blocked": reason,
        })
    } else {
        json!({
            "source_line_start": matched.source_line_start,
            "source_line_end": matched.source_line_end,
            "decl_line": def.decl_line,
            "body_line_start": def.body_line_start,
            "decorator_line_start": def.decorator_line_start,
            "decorator_line_end": def.decorator_line_end,
        })
    };

    // Ambiguity: other candidates sharing the winner's span length are genuine
    // same-granularity overlaps. Strictly larger spans are the nesting stack
    // already captured above, so they are not alternatives. Only meaningful for
    // a line query; an `entity`-id lookup has no line to disambiguate.
    let matched_len = span_len(matched);
    let alternatives: Vec<Value> = match line {
        Some(line) => candidates
            .iter()
            .skip(1)
            .filter(|cand| matched_len.is_some() && span_len(cand) == matched_len)
            .take(ENTITY_CONTEXT_MAX_ALTERNATIVES)
            .map(|cand| {
                json!({
                    "entity": entity_json(conn, cand),
                    "match_reason": match_reason_for(line, cand),
                })
            })
            .collect(),
        None => Vec::new(),
    };

    // For a line query, explain why that line matched; for a direct `entity`-id
    // lookup there is no line, so the reason is simply "entity".
    let match_reason = match line {
        Some(line) => match_reason_for(line, matched),
        None => "entity",
    };

    json!({
        "query_line": line,
        "match_reason": match_reason,
        "containing_stack": containing_stack,
        "ranges": ranges,
        "alternatives": alternatives,
        "freshness": freshness,
    })
}

/// Per-section neighbor cap for `orientation_pack` — keeps the packet bounded
/// while still surfacing the most relevant edges; overflow is reported in
/// `omitted`.
const ORIENTATION_PACK_MAX_NEIGHBORS: usize = 10;

/// Call-graph traversal depth for `orientation_pack`'s compact execution paths.
/// Matches the `execution_paths_from` default so the packet's paths line up
/// with a follow-up call to that tool.
const ORIENTATION_PACK_PATH_DEPTH: usize = 3;

/// Sync portion of an `orientation_pack`: everything one reader snapshot can
/// produce, plus the flags the async assembly stage needs for warnings and the
/// `omitted` block. Issues + health are layered on afterward.
struct OrientationCore {
    primary_id: Option<String>,
    primary_kind: Option<String>,
    /// True when the request used the `entity`-id form (so a `None`
    /// `primary_id` is a hard not-found, not a graceful `no_match`).
    lookup_was_id: bool,
    packet: Value,
    freshness: Value,
    staleness_stale: bool,
    /// Whether the served index has any alive SEI bindings (REQ-C-04 /
    /// ADR-038). Resolved in the reader closure and surfaced under `health.sei`.
    sei_populated: bool,
    neighbors_omitted: serde_json::Map<String, Value>,
    paths_truncation_reason: Option<String>,
    /// Set when the resolved primary entity is briefing-blocked: the pack is
    /// refused (no identity, no surrounding structure) rather than built
    /// (clarion-307668e2be).
    briefing_blocked: Option<String>,
}

/// Sort a neighbor list by entity id (stable, deterministic) and cap it,
/// returning the kept list and how many were dropped.
fn cap_neighbor_list(mut list: Vec<Value>, cap: usize) -> (Vec<Value>, usize) {
    list.sort_by(|a, b| neighbor_sort_key(a).cmp(neighbor_sort_key(b)));
    let omitted = list.len().saturating_sub(cap);
    list.truncate(cap);
    (list, omitted)
}

/// Entity id of a neighbor JSON record (`{"entity": {"id": ...}, ...}` or a
/// bare entity object), used as a deterministic sort key.
fn neighbor_sort_key(value: &Value) -> &str {
    value
        .get("entity")
        .and_then(|entity| entity.get("id"))
        .or_else(|| value.get("id"))
        .and_then(Value::as_str)
        .unwrap_or("")
}

/// Deterministic follow-up reads for an `orientation_pack`. Suggests the full
/// source, a pre-spend cost preview, the owning subsystem, and a drill-down
/// into the first callee (by sorted id) when one exists. Empty when there is no
/// primary entity.
fn orientation_suggested_reads(
    packet: &Value,
    primary_id: Option<&str>,
    primary_kind: Option<&str>,
) -> Vec<Value> {
    let Some(primary_id) = primary_id else {
        return Vec::new();
    };
    let mut reads = vec![
        json!({
            "tool": "entity_source_get",
            "args": {"id": primary_id},
            "why": "read the entity's source with line numbers",
        }),
        json!({
            "tool": "entity_summary_preview_cost_get",
            "args": {"id": primary_id},
            "why": "estimate the cost of an LLM briefing before spending",
        }),
    ];
    // A subsystem's useful drill-down is its members; for any other kind it is
    // the owning subsystem.
    if primary_kind == Some("subsystem") {
        reads.push(json!({
            "tool": "subsystem_member_list",
            "args": {"id": primary_id},
            "why": "list the entities clustered into this subsystem",
        }));
    } else {
        reads.push(json!({
            "tool": "entity_subsystem_get",
            "args": {"id": primary_id},
            "why": "see which subsystem this entity belongs to",
        }));
    }
    // Drill into the first callee (lists are already id-sorted), if any.
    if let Some(callee_id) = packet
        .get("neighbors")
        .and_then(|neighbors| neighbors.get("callees"))
        .and_then(Value::as_array)
        .and_then(|callees| callees.first())
        .and_then(|callee| callee.get("entity"))
        .and_then(|entity| entity.get("id"))
        .and_then(Value::as_str)
    {
        reads.push(json!({
            "tool": "entity_orientation_pack_get",
            "args": {"entity": callee_id},
            "why": "orient on the primary callee",
        }));
    }
    reads
}

/// How recently the analyze progress file must have been stamped for
/// `analyze_status` to call progress "observed" rather than possibly stalled.
const ANALYZE_HEARTBEAT_STALE_SECS: i64 = 30;

/// Live state of a registry-tracked analyze run, captured under the lock.
enum LiveRun {
    Alive {
        started_at: String,
        progress_path: PathBuf,
    },
    Exited {
        started_at: String,
        cancelled: bool,
    },
    Absent,
}

/// Result of an `analyze_cancel` attempt against the registry.
enum CancelOutcome {
    Cancelled,
    AlreadyExited { cancelled: bool },
    Absent,
}

/// Read and parse the analyze progress snapshot, if present and valid JSON.
fn read_progress_snapshot(path: &std::path::Path) -> Option<Value> {
    let body = std::fs::read_to_string(path).ok()?;
    serde_json::from_str(&body).ok()
}

/// Parse a timestamp to Unix seconds, accepting both the MCP clock's
/// `unix:<seconds>` form and the RFC3339 form analyze writes into the progress
/// file's `heartbeat_at`. `None` if neither parses.
pub(crate) fn parse_to_unix_seconds(value: &str) -> Option<i64> {
    use time::OffsetDateTime;
    use time::format_description::well_known::Rfc3339;
    if let Some(rest) = value.strip_prefix("unix:") {
        return rest.trim().parse().ok();
    }
    OffsetDateTime::parse(value, &Rfc3339)
        .ok()
        .map(OffsetDateTime::unix_timestamp)
}

/// Whole seconds between two timestamps (`now - start`), or `None` if either
/// fails to parse. Accepts mixed `unix:`/RFC3339 forms.
fn elapsed_seconds(start: &str, now: &str) -> Option<i64> {
    Some(parse_to_unix_seconds(now)? - parse_to_unix_seconds(start)?)
}

/// Whether the heartbeat is recent enough to treat the run as actively
/// progressing (heartbeat age within `max_age_secs`).
fn progress_observed(heartbeat: &str, now: &str, max_age_secs: i64) -> bool {
    elapsed_seconds(heartbeat, now).is_some_and(|age| (0..=max_age_secs).contains(&age))
}

/// True when a run's `stats` JSON marks it as cancelled (vs an ordinary
/// failure) — the MCP writes `terminal_reason="cancelled"` on cancel.
fn run_stats_is_cancelled(stats_json: &str) -> bool {
    serde_json::from_str::<Value>(stats_json)
        .ok()
        .and_then(|stats| {
            stats
                .get("terminal_reason")
                .and_then(Value::as_str)
                .map(|reason| reason == "cancelled")
        })
        .unwrap_or(false)
}

/// Map a `runs.status` value to the `analyze_status` vocabulary. A row still
/// reading `running` after the process exited is abnormal (the process died
/// without finalizing) and reported as `failed`.
fn map_run_status(db_status: &str, stats_json: &str) -> &'static str {
    match db_status {
        "completed" => "completed",
        "skipped_no_plugins" => "skipped_no_plugins",
        "failed" if run_stats_is_cancelled(stats_json) => "cancelled",
        _ => "failed",
    }
}

/// Maximum number of (span + context) lines `source_for_entity` will emit
/// before truncating, so a pathologically large entity never floods an agent's
/// context. The span itself is bounded by the entity; this caps the total.
const SOURCE_FOR_ENTITY_MAX_LINES: usize = 2_000;

/// Direction for `call_sites`: outgoing (this entity is the caller) or incoming
/// (this entity is the callee).
#[derive(Clone, Copy)]
enum CallSiteRole {
    Caller,
    Callee,
}

/// Edge-kind filter for `call_sites`.
#[derive(Clone, Copy)]
enum CallSiteKind {
    Both,
    Calls,
    References,
}

impl CallSiteKind {
    fn includes_calls(self) -> bool {
        matches!(self, Self::Both | Self::Calls)
    }
    fn includes_references(self) -> bool {
        matches!(self, Self::Both | Self::References)
    }
}

/// Production/test path scope for `call_sites`. Best-effort: source-file
/// production/test partitioning is not indexed, so this is a heuristic over the
/// file path (documented as such in the tool description).
#[derive(Clone, Copy)]
enum PathScope {
    All,
    Production,
    Test,
}

impl PathScope {
    fn admits(self, path: Option<&str>) -> bool {
        match (self, path) {
            (Self::All, _) => true,
            // A site whose owning file can't be resolved is excluded from a
            // narrowed scope rather than guessed into it.
            (_, None) => false,
            (Self::Test, Some(p)) => is_test_path(p),
            (Self::Production, Some(p)) => !is_test_path(p),
        }
    }
}

/// Conventional Python test-path heuristic (pytest/unittest layouts). Not a
/// substitute for indexed metadata — see `PathScope`.
fn is_test_path(path: &str) -> bool {
    let lower = path.replace('\\', "/").to_ascii_lowercase();
    let file = lower.rsplit('/').next().unwrap_or(lower.as_str());
    lower.contains("/tests/")
        || lower.contains("/test/")
        || file.starts_with("test_")
        || file.ends_with("_test.py")
        || file == "conftest.py"
}

/// Per-call cap on each of the resolved and unresolved site lists.
const CALL_SITES_MAX: usize = 200;

/// 1-based line and 0-based byte column for a byte offset into `content`.
// The slices are a single source file; a dependency on `bytecount` for the
// newline count is not warranted here.
#[allow(clippy::naive_bytecount)]
fn byte_line_col(content: &str, byte_offset: i64) -> Option<(i64, i64)> {
    let off = usize::try_from(byte_offset).ok()?;
    let bytes = content.as_bytes();
    if off > bytes.len() {
        return None;
    }
    let line = bytes[..off].iter().filter(|&&b| b == b'\n').count() + 1;
    let line_start = bytes[..off]
        .iter()
        .rposition(|&b| b == b'\n')
        .map_or(0, |p| p + 1);
    Some((
        i64::try_from(line).ok()?,
        i64::try_from(off - line_start).ok()?,
    ))
}

/// The text of `line` (1-based) in `content`, or empty if out of range.
fn line_text_at(content: &str, line: i64) -> String {
    let idx = usize::try_from(line - 1).unwrap_or(usize::MAX);
    content.lines().nth(idx).unwrap_or("").to_owned()
}

/// One resolved call/reference site before file resolution.
struct ResolvedSite {
    owner_id: String,
    edge_kind: &'static str,
    other_id: String,
    confidence: EdgeConfidence,
    byte_start: Option<i64>,
    byte_end: Option<i64>,
}

/// One static call Loomweave could not bind (kept separate from resolved sites).
struct UnboundSite {
    owner_id: String,
    callee_expr: String,
    byte_start: i64,
    byte_end: i64,
}

/// Gather the resolved and unbound (statically-unbindable) call/reference sites
/// for `entity` in the requested direction, applying the edge-kind and
/// confidence filters. File/line resolution happens in [`build_call_sites`].
fn collect_call_sites(
    conn: &rusqlite::Connection,
    entity: &EntityRow,
    role: CallSiteRole,
    kind: CallSiteKind,
    confidence: EdgeConfidence,
) -> Result<(Vec<ResolvedSite>, Vec<UnboundSite>), StorageError> {
    let mut resolved: Vec<ResolvedSite> = Vec::new();
    let mut unbound: Vec<UnboundSite> = Vec::new();

    match role {
        CallSiteRole::Caller => {
            if kind.includes_calls() {
                for edge in call_edges_from(conn, &entity.id, confidence)? {
                    resolved.push(ResolvedSite {
                        owner_id: entity.id.clone(),
                        edge_kind: "calls",
                        other_id: edge.to_id,
                        confidence: edge.confidence,
                        byte_start: edge.source_byte_start,
                        byte_end: edge.source_byte_end,
                    });
                }
                for site in unresolved_call_sites_for_caller(conn, &entity.id, CALL_SITES_MAX)? {
                    unbound.push(UnboundSite {
                        owner_id: entity.id.clone(),
                        callee_expr: site.callee_expr,
                        byte_start: site.source_byte_start,
                        byte_end: site.source_byte_end,
                    });
                }
            }
            if kind.includes_references() {
                for r in reference_edges_for_entity(conn, &entity.id, ReferenceDirection::Out)? {
                    if r.confidence <= confidence {
                        resolved.push(ResolvedSite {
                            owner_id: entity.id.clone(),
                            edge_kind: "references",
                            other_id: r.neighbor_id,
                            confidence: r.confidence,
                            byte_start: r.source_byte_start,
                            byte_end: r.source_byte_end,
                        });
                    }
                }
            }
        }
        CallSiteRole::Callee => {
            if kind.includes_calls() {
                for edge in call_edges_targeting(conn, &entity.id, confidence)? {
                    resolved.push(ResolvedSite {
                        owner_id: edge.from_id.clone(),
                        edge_kind: "calls",
                        other_id: edge.from_id,
                        confidence: edge.confidence,
                        byte_start: edge.source_byte_start,
                        byte_end: edge.source_byte_end,
                    });
                }
                for site in unresolved_callers_for_target(conn, entity, CALL_SITES_MAX)? {
                    unbound.push(UnboundSite {
                        owner_id: site.caller_entity_id,
                        callee_expr: site.callee_expr,
                        byte_start: site.source_byte_start,
                        byte_end: site.source_byte_end,
                    });
                }
            }
            if kind.includes_references() {
                for r in reference_edges_for_entity(conn, &entity.id, ReferenceDirection::In)? {
                    if r.confidence <= confidence {
                        resolved.push(ResolvedSite {
                            owner_id: r.neighbor_id.clone(),
                            edge_kind: "references",
                            other_id: r.neighbor_id,
                            confidence: r.confidence,
                            byte_start: r.source_byte_start,
                            byte_end: r.source_byte_end,
                        });
                    }
                }
            }
        }
    }

    Ok((resolved, unbound))
}

/// Build the `call_sites` payload. Returns `Ok(None)` when the entity does not
/// exist (so the caller can emit a not-found envelope).
fn build_call_sites(
    conn: &rusqlite::Connection,
    entity_id: &str,
    role: CallSiteRole,
    kind: CallSiteKind,
    confidence: EdgeConfidence,
    path: PathScope,
) -> Result<Option<Value>, StorageError> {
    let Some(entity) = entity_by_id(conn, entity_id)? else {
        return Ok(None);
    };

    let (resolved, unbound) = collect_call_sites(conn, &entity, role, kind, confidence)?;

    // Resolve each site's owning file + disclosure state once, mapping the byte
    // anchor to a line only after scanner and drift guards pass.
    let mut owner_meta: HashMap<String, OwnerMeta> = HashMap::new();
    let mut file_content: HashMap<String, Option<Vec<u8>>> = HashMap::new();
    // The queried entity is known without a lookup.
    owner_meta.insert(entity.id.clone(), OwnerMeta::from_entity(entity.clone()));

    let mut site_values = Vec::new();
    let mut truncated = false;
    for site in resolved {
        if site_values.len() >= CALL_SITES_MAX {
            truncated = true;
            break;
        }
        let owner = resolve_owner(conn, &mut owner_meta, &site.owner_id)?;
        if !path.admits(owner.path.as_deref()) {
            continue;
        }
        let anchor = anchor_line(&mut file_content, owner, site.byte_start);
        site_values.push(json!({
            "edge_kind": site.edge_kind,
            "other_id": site.other_id,
            "confidence": site.confidence.as_str(),
            "file": owner.path,
            "line": anchor.line,
            "column": anchor.column,
            "line_text": anchor.line_text,
            "source_status": anchor.source_status,
            "briefing_blocked": anchor.briefing_blocked,
            "drift": anchor.drift,
            "byte_start": site.byte_start,
            "byte_end": site.byte_end
        }));
    }

    let mut unresolved_values = Vec::new();
    for site in unbound {
        if unresolved_values.len() >= CALL_SITES_MAX {
            truncated = true;
            break;
        }
        let owner = resolve_owner(conn, &mut owner_meta, &site.owner_id)?;
        if !path.admits(owner.path.as_deref()) {
            continue;
        }
        let anchor = anchor_line(&mut file_content, owner, Some(site.byte_start));
        unresolved_values.push(json!({
            "callee_expr": site.callee_expr,
            "file": owner.path,
            "line": anchor.line,
            "column": anchor.column,
            "line_text": anchor.line_text,
            "source_status": anchor.source_status,
            "briefing_blocked": anchor.briefing_blocked,
            "drift": anchor.drift,
            "byte_start": site.byte_start,
            "byte_end": site.byte_end
        }));
    }

    Ok(Some(json!({
        "entity": entity_json(conn, &entity),
        "role": match role { CallSiteRole::Caller => "caller", CallSiteRole::Callee => "callee" },
        "filters": {
            "kind": match kind {
                CallSiteKind::Both => "both",
                CallSiteKind::Calls => "calls",
                CallSiteKind::References => "references",
            },
            "confidence": confidence.as_str(),
            "path": match path {
                PathScope::All => "all",
                PathScope::Production => "production",
                PathScope::Test => "test",
            }
        },
        "sites": site_values,
        "unresolved_sites": unresolved_values,
        "truncated": truncated,
        "scope_excludes": call_graph_scope_excludes(confidence)
    })))
}

#[derive(Clone)]
struct OwnerMeta {
    entity: Option<EntityRow>,
    path: Option<String>,
    briefing_blocked: bool,
}

impl OwnerMeta {
    fn from_entity(entity: EntityRow) -> Self {
        let briefing_blocked = briefing_block_reason(&entity).is_some();
        let path = entity.source_file_path.clone();
        Self {
            entity: Some(entity),
            path,
            briefing_blocked,
        }
    }

    fn missing() -> Self {
        Self {
            entity: None,
            path: None,
            briefing_blocked: false,
        }
    }
}

struct AnchorLine {
    line: Value,
    column: Value,
    line_text: String,
    source_status: &'static str,
    briefing_blocked: bool,
    drift: Value,
}

impl AnchorLine {
    fn redacted(source_status: &'static str, briefing_blocked: bool, drift: Value) -> Self {
        Self {
            line: Value::Null,
            column: Value::Null,
            line_text: String::new(),
            source_status,
            briefing_blocked,
            drift,
        }
    }

    fn ok(line: i64, column: i64, line_text: String) -> Self {
        Self {
            line: json!(line),
            column: json!(column),
            line_text,
            source_status: "ok",
            briefing_blocked: false,
            drift: Value::Null,
        }
    }
}

/// Memoized lookup of an owner entity's disclosure metadata. A
/// briefing-blocked owner's source bytes must never be read — the pre-ingest
/// scanner withholds them — so `call_sites` redacts `line_text` for such owners
/// rather than disclosing the file content behind an edge.
fn resolve_owner<'a>(
    conn: &rusqlite::Connection,
    cache: &'a mut HashMap<String, OwnerMeta>,
    owner_id: &str,
) -> Result<&'a OwnerMeta, StorageError> {
    if !cache.contains_key(owner_id) {
        let meta = match entity_by_id(conn, owner_id)? {
            Some(entity) => OwnerMeta::from_entity(entity),
            None => OwnerMeta::missing(),
        };
        cache.insert(owner_id.to_owned(), meta);
    }
    Ok(cache
        .get(owner_id)
        .expect("owner metadata inserted before lookup"))
}

/// Map a byte anchor to line evidence after enforcing source disclosure guards.
/// Any piece that can't be resolved degrades to JSON null / empty rather than
/// failing the whole query.
fn anchor_line(
    file_content: &mut HashMap<String, Option<Vec<u8>>>,
    owner: &OwnerMeta,
    byte_start: Option<i64>,
) -> AnchorLine {
    if owner.briefing_blocked {
        return AnchorLine::redacted("briefing_blocked", true, Value::Null);
    }
    let (Some(entity), Some(path), Some(byte_start)) =
        (owner.entity.as_ref(), owner.path.as_deref(), byte_start)
    else {
        return AnchorLine::redacted("unavailable", false, Value::Null);
    };

    let bytes = file_content
        .entry(path.to_owned())
        .or_insert_with(|| std::fs::read(path).ok());
    let Some(bytes) = bytes.as_deref() else {
        return AnchorLine::redacted("missing", false, Value::Null);
    };
    let Ok(content) = String::from_utf8(bytes.to_vec()) else {
        return AnchorLine::redacted("binary", false, Value::Null);
    };

    if let (Some(stored), Some(current)) = (
        entity.content_hash.as_deref(),
        current_source_content_hash(entity, bytes, Some(&content)),
    ) && stored != current
    {
        return AnchorLine::redacted(
            "drifted",
            false,
            json!({
                "stored_content_hash": stored,
                "current_content_hash": current
            }),
        );
    }

    match byte_line_col(&content, byte_start) {
        Some((line, column)) => AnchorLine::ok(line, column, line_text_at(&content, line)),
        None => AnchorLine::redacted("unavailable", false, Value::Null),
    }
}

/// Build the `source_for_entity` payload: the entity's exact indexed line span
/// plus `context_lines` of surrounding context, line-numbered and drift-checked.
///
/// Returns an explicit `source_status` rather than a stale or misleading
/// snippet when the source cannot be trusted: `missing` (file gone),
/// `no_source_path` / `no_range` (no anchor to read), `binary` (non-UTF-8), or
/// `drifted` (the file no longer hashes to the indexed `content_hash`).
fn source_for_entity_json(
    conn: &rusqlite::Connection,
    entity: &EntityRow,
    context_lines: usize,
) -> Value {
    let identity = entity_json(conn, entity);

    // Refuse to read or return bytes for an entity whose file the pre-ingest
    // scanner marked `briefing_blocked`. Without this guard, an agent holding
    // the id of a function/class in a secret-bearing file could use
    // source_for_entity to disclose exactly the bytes the scanner policy
    // withholds (the summary / HTTP read surfaces already refuse these).
    if let Some(reason) = briefing_block_reason(entity) {
        return json!({
            "entity": identity,
            "source_status": "briefing_blocked",
            "briefing_blocked": reason
        });
    }

    let Some(path) = entity.source_file_path.as_deref() else {
        return json!({"entity": identity, "source_status": "no_source_path"});
    };

    let source_anchor = source_anchor_for_entity(conn, entity, path);
    if let Some(reason) = source_anchor
        .as_ref()
        .and_then(|anchor| anchor.briefing_blocked.as_deref())
    {
        return json!({
            "entity": identity,
            "source_file_path": path,
            "source_status": "briefing_blocked",
            "briefing_blocked": reason
        });
    }

    let (Some(start_line), Some(end_line)) = (entity.source_line_start, entity.source_line_end)
    else {
        return json!({
            "entity": identity,
            "source_file_path": path,
            "source_status": "no_range"
        });
    };

    let Ok(bytes) = std::fs::read(path) else {
        return json!({
            "entity": identity,
            "source_file_path": path,
            "source_status": "missing"
        });
    };
    let Ok(source) = String::from_utf8(bytes.clone()) else {
        return json!({
            "entity": identity,
            "source_file_path": path,
            "source_status": "binary"
        });
    };

    // Refuse to hand back a snippet that no longer matches what was indexed.
    if let (Some(stored), Some(current)) = (
        entity.content_hash.as_deref(),
        current_source_content_hash(entity, &bytes, Some(&source)),
    ) && stored != current
    {
        return json!({
            "entity": identity,
            "source_file_path": path,
            "source_status": "drifted",
            "drift": {
                "stored_content_hash": stored,
                "current_content_hash": current
            }
        });
    }

    let (emitted_context_lines, context_omitted_reason) =
        match guard_source_context(entity, context_lines, source_anchor.as_ref(), &bytes) {
            SourceContextGuard::Emit(lines) => (lines, None),
            SourceContextGuard::Omit(reason) => (0, Some(reason)),
            SourceContextGuard::Drift {
                stored_content_hash,
                current_content_hash,
            } => {
                return json!({
                    "entity": identity,
                    "source_file_path": path,
                    "source_status": "drifted",
                    "drift": {
                        "stored_content_hash": stored_content_hash,
                        "current_content_hash": current_content_hash
                    }
                });
            }
        };

    let lines: Vec<&str> = source.lines().collect();
    let total = i64::try_from(lines.len()).unwrap_or(i64::MAX);
    // Clamp the span to the file, then widen by the context window. 1-based,
    // inclusive on both ends.
    let span_start = start_line.max(1);
    let span_end = end_line.min(total).max(span_start);
    let ctx = i64::try_from(emitted_context_lines).unwrap_or(i64::MAX);
    let window_start = (span_start - ctx).max(1);
    let window_end = (span_end + ctx).min(total);

    let mut emitted = Vec::new();
    let mut truncated = false;
    let mut number = window_start;
    while number <= window_end {
        if emitted.len() >= SOURCE_FOR_ENTITY_MAX_LINES {
            truncated = true;
            break;
        }
        let idx = usize::try_from(number - 1).unwrap_or(usize::MAX);
        let text = lines.get(idx).copied().unwrap_or("");
        emitted.push(json!({
            "number": number,
            "text": text,
            "in_entity": number >= span_start && number <= span_end
        }));
        number += 1;
    }

    json!({
        "entity": identity,
        "source_file_path": path,
        "source_status": "ok",
        "line_start": span_start,
        "line_end": span_end,
        "context_lines": emitted_context_lines,
        "requested_context_lines": context_lines,
        "context_omitted_reason": context_omitted_reason,
        "window_start": window_start,
        "window_end": window_end,
        "lines": emitted,
        "truncated": truncated
    })
}

enum SourceContextGuard {
    Emit(usize),
    Omit(&'static str),
    Drift {
        stored_content_hash: String,
        current_content_hash: String,
    },
}

fn guard_source_context(
    entity: &EntityRow,
    context_lines: usize,
    source_anchor: Option<&SourceAnchorGuard>,
    bytes: &[u8],
) -> SourceContextGuard {
    // Non-file-scope entities store a span hash, but context lines are outside
    // that span. Only return caller-requested context when we can verify the
    // enclosing source-file hash; otherwise emit the exact entity span only.
    if context_lines == 0 || is_plugin_file_scope_entity(entity) {
        return SourceContextGuard::Emit(context_lines);
    }
    let Some(stored_file_hash) = source_anchor.and_then(|anchor| anchor.content_hash.as_deref())
    else {
        return SourceContextGuard::Omit("unverified_context");
    };
    let current_file_hash = blake3::hash(bytes).to_hex().to_string();
    if stored_file_hash == current_file_hash {
        SourceContextGuard::Emit(context_lines)
    } else {
        SourceContextGuard::Drift {
            stored_content_hash: stored_file_hash.to_owned(),
            current_content_hash: current_file_hash,
        }
    }
}

#[derive(Debug, Clone)]
struct SourceAnchorGuard {
    content_hash: Option<String>,
    briefing_blocked: Option<String>,
}

fn source_anchor_for_entity(
    conn: &rusqlite::Connection,
    entity: &EntityRow,
    source_file_path: &str,
) -> Option<SourceAnchorGuard> {
    if let Some(source_file_id) = entity.source_file_id.as_deref()
        && source_file_id != entity.id
        && let Ok(Some(anchor)) = entity_by_id(conn, source_file_id)
    {
        let briefing_blocked = briefing_block_reason(&anchor);
        return Some(SourceAnchorGuard {
            content_hash: anchor.content_hash,
            briefing_blocked,
        });
    }

    conn.query_row(
        "SELECT properties, content_hash \
         FROM entities \
         WHERE source_file_path = ?1 AND kind = 'file' \
         ORDER BY CASE plugin_id WHEN 'core' THEN 0 ELSE 1 END, id ASC \
         LIMIT 1",
        [source_file_path],
        |row| {
            let properties_json: String = row.get(0)?;
            let content_hash: Option<String> = row.get(1)?;
            Ok(SourceAnchorGuard {
                content_hash,
                briefing_blocked: entity_briefing_block_reason(&properties_json),
            })
        },
    )
    .optional()
    .ok()
    .flatten()
}

fn verified_source_excerpt(entity: &EntityRow) -> Result<String, SourceExcerptError> {
    let Some(path) = entity.source_file_path.as_deref() else {
        return Ok(String::new());
    };
    let Ok(bytes) = std::fs::read(path) else {
        return Ok(String::new());
    };
    let source = String::from_utf8(bytes.clone()).ok();
    if let (Some(stored_content_hash), Some(current_content_hash)) = (
        entity.content_hash.as_deref(),
        current_source_content_hash(entity, &bytes, source.as_deref()),
    ) && stored_content_hash != current_content_hash
    {
        return Err(SourceExcerptError {
            entity_id: entity.id.clone(),
            stored_content_hash: stored_content_hash.to_owned(),
            current_content_hash,
        });
    }
    let Some(source) = source else {
        return Ok(String::new());
    };
    let excerpt = line_range_excerpt(&source, entity.source_line_start, entity.source_line_end)
        .unwrap_or(source);
    Ok(truncate_excerpt(excerpt))
}

fn current_source_content_hash(
    entity: &EntityRow,
    file_bytes: &[u8],
    source: Option<&str>,
) -> Option<String> {
    // A core `file` catalog row hashes its whole contents (analyze's
    // `whole_file_hash`), same as a plugin file-scope entity. Relation-edge
    // anchors resolve their owner from the edge's `source_file_id`, so drift
    // must be detectable on file rows too. Gated on core ownership, not the
    // kind name alone — a plugin-owned "file" kind makes no whole-file-hash
    // promise.
    if (entity.plugin_id == "core" && entity.kind == "file") || is_plugin_file_scope_entity(entity)
    {
        return Some(blake3::hash(file_bytes).to_hex().to_string());
    }
    let source = source?;
    let start_line = entity.source_line_start?;
    let end_line = entity.source_line_end?;
    if start_line <= 0 || end_line < start_line {
        return None;
    }
    let start = usize::try_from(start_line - 1).ok()?;
    let mut end = usize::try_from(end_line).ok()?;
    let lines = source.lines().collect::<Vec<_>>();
    end = end.min(lines.len());
    if start >= end {
        return None;
    }
    let normalized = lines[start..end].join("\n");
    Some(blake3::hash(normalized.as_bytes()).to_hex().to_string())
}

fn line_range_excerpt(
    source: &str,
    start_line: Option<i64>,
    end_line: Option<i64>,
) -> Option<String> {
    let start_line = start_line?;
    let end_line = end_line?;
    if start_line <= 0 || end_line < start_line {
        return None;
    }
    let start = usize::try_from(start_line - 1).ok()?;
    let end = usize::try_from(end_line).ok()?;
    let lines = source.split_inclusive('\n').collect::<Vec<_>>();
    let end = end.min(lines.len());
    if start >= end {
        return None;
    }
    Some(lines[start..end].concat())
}

fn truncate_excerpt(source: String) -> String {
    if source.len() > 8_000 {
        source.chars().take(8_000).collect()
    } else {
        source
    }
}

fn unresolved_sites_json(sites: &[UnresolvedCallSiteRow]) -> String {
    serde_json::to_string(
        &sites
            .iter()
            .map(|site| {
                json!({
                    "caller_entity_id": site.caller_entity_id,
                    "caller_content_hash": site.caller_content_hash,
                    "site_key": site.site_key,
                    "site_ordinal": site.site_ordinal,
                    "source_file_id": site.source_file_id,
                    "source_byte_start": site.source_byte_start,
                    "source_byte_end": site.source_byte_end,
                    "callee_expr": site.callee_expr
                })
            })
            .collect::<Vec<_>>(),
    )
    .expect("unresolved site JSON serializes")
}

fn entities_json(entities: &[EntityRow]) -> String {
    // Candidate entities for the LLM inference prompt — an internal payload,
    // not a tool return surface, so it carries identity fields only (no `sei`).
    serde_json::to_string(
        &entities
            .iter()
            .map(entity_identity_json)
            .collect::<Vec<_>>(),
    )
    .expect("candidate entity JSON serializes")
}

fn inferred_records_from_result(
    read: &InferredRead,
    result_json: &str,
    max_edges: usize,
) -> Result<Vec<InferredCallEdgeRecord>, InferredDispatchFailure> {
    let parsed: InferredCallsResponse = serde_json::from_str(result_json).map_err(|err| {
        InferredDispatchFailure::new(
            McpErrorCode::LlmInvalidJson,
            &format!("inferred provider returned invalid JSON: {err}"),
            true,
        )
    })?;
    let cache_key = inferred_edge_cache_key_id(&read.key);
    let sites_by_key = read
        .sites
        .iter()
        .map(|site| (site.site_key.as_str(), site))
        .collect::<HashMap<_, _>>();
    let mut records = Vec::new();
    for edge in parsed.edges.into_iter().take(max_edges) {
        if edge.target_id.trim().is_empty() {
            continue;
        }
        let site = match edge.site_key.as_deref() {
            Some(site_key) => sites_by_key.get(site_key).copied(),
            None if read.sites.len() == 1 => read.sites.first(),
            None => None,
        };
        let Some(site) = site else {
            continue;
        };
        let properties = json!({
            "model_id": read.key.model_id,
            "prompt_version": read.key.prompt_version,
            "caller_content_hash": read.key.caller_content_hash,
            "inference_cache_key": cache_key,
            "site_key": site.site_key,
            "model_confidence": edge.confidence,
            "rationale": edge.rationale
        });
        records.push(InferredCallEdgeRecord {
            from_id: read.caller.id.clone(),
            to_id: edge.target_id,
            source_file_id: site.source_file_id.clone(),
            source_byte_start: site.source_byte_start,
            source_byte_end: site.source_byte_end,
            properties_json: properties.to_string(),
        });
    }
    Ok(records)
}

fn stale_semantic(entry: &SummaryCacheEntry, caller_count: i64, fan_out: i64) -> bool {
    entry.stale_semantic
        || count_drifted(entry.caller_count, caller_count)
        || count_drifted(entry.fan_out, fan_out)
}

fn count_drifted(stored: i64, current: i64) -> bool {
    if stored == current {
        return false;
    }
    if stored == 0 {
        return current != 0;
    }
    i128::from((current - stored).abs()) * 2 > i128::from(stored.abs())
}

fn summary_cache_expired(created_at: &str, now: &str, max_age_days: u32) -> bool {
    let Some(created) = timestamp_day_index(created_at) else {
        return false;
    };
    let Some(current) = timestamp_day_index(now) else {
        return false;
    };
    current.saturating_sub(created) > i64::from(max_age_days)
}

fn timestamp_day_index(raw: &str) -> Option<i64> {
    if let Some(seconds) = raw.strip_prefix("unix:") {
        return seconds.parse::<i64>().ok().map(|value| value / 86_400);
    }
    let date = raw.get(..10)?;
    let date = Date::parse(date, format_description!("[year]-[month]-[day]")).ok()?;
    let unix_epoch = Date::from_calendar_date(1970, Month::January, 1)
        .expect("Unix epoch is a valid calendar date");
    Some(i64::from(date.to_julian_day() - unix_epoch.to_julian_day()))
}

/// The `max_output_tokens` ceiling a leaf summary request reserves. Reported by
/// `summary_preview_cost` as the output ceiling (not a length prediction).
const SUMMARY_MAX_OUTPUT_TOKENS: i64 = 512;

/// A provider-free, deterministic input-token estimate for `summary_preview_cost`:
/// roughly four characters per token. Intended only as a pre-spend order-of-
/// magnitude hint, not an exact count (the real count is recorded on the cache
/// row once a summary has actually run).
fn estimate_tokens_from_chars(text: &str) -> i64 {
    let chars = i64::try_from(text.chars().count()).unwrap_or(i64::MAX);
    // ceil(chars / 4) without the unstable i64::div_ceil.
    chars.saturating_add(3) / 4
}

fn default_now_string() -> String {
    let seconds = OffsetDateTime::now_utc().unix_timestamp();
    format!("unix:{seconds}")
}

fn guidance_authored_at_from_clock(raw: &str) -> String {
    const ISO_MILLIS_UTC: &[time::format_description::FormatItem<'_>] =
        format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z");

    let parsed = if let Some(seconds) = raw.strip_prefix("unix:") {
        seconds
            .trim()
            .parse::<i64>()
            .ok()
            .and_then(|seconds| OffsetDateTime::from_unix_timestamp(seconds).ok())
    } else {
        parse_to_unix_seconds(raw)
            .and_then(|seconds| OffsetDateTime::from_unix_timestamp(seconds).ok())
    };

    parsed
        .and_then(|timestamp| timestamp.format(&ISO_MILLIS_UTC).ok())
        .unwrap_or_else(|| raw.to_owned())
}

fn caller_json(
    conn: &rusqlite::Connection,
    edge: &CallEdgeMatch,
) -> Result<Option<Value>, StorageError> {
    Ok(entity_by_id(conn, &edge.from_id)?.map(|entity| {
        json!({
            "entity": entity_json(conn, &entity),
            "edge_confidence": edge.confidence.as_str(),
            "source_byte_start": edge.source_byte_start,
            "source_byte_end": edge.source_byte_end,
            "target_id": edge.to_id,
            "stored_to_id": edge.stored_to_id
        })
    }))
}

fn callee_json(
    conn: &rusqlite::Connection,
    edge: &CallEdgeMatch,
) -> Result<Option<Value>, StorageError> {
    Ok(entity_by_id(conn, &edge.to_id)?.map(|entity| {
        // `stored_to_id` echoes the callee's raw id. A blocked callee now keeps
        // its navigable identity in `entity_json` (clarion-719e7320f5, A3), so
        // this id is no longer secret — it rides through unredacted, matching the
        // exposed `entity.id`.
        json!({
            "entity": entity_json(conn, &entity),
            "edge_confidence": edge.confidence.as_str(),
            "source_byte_start": edge.source_byte_start,
            "source_byte_end": edge.source_byte_end,
            "stored_to_id": edge.stored_to_id
        })
    }))
}

/// Compacted execution-path payload: a deduplicated node table, id-only ranked
/// paths, and whether the path cap trimmed the ranked set.
struct CompactPaths {
    nodes: Vec<Value>,
    paths: Vec<Vec<String>>,
    path_cap_truncated: bool,
}

/// Compact, ranked execution-path payload (clarion-5b3eff9a91). Ranks paths
/// longest-first (deepest reachable flow) with a lexicographic tie-break for
/// determinism, applies the server path cap (clarion-23ae24358c), then emits a
/// deduplicated node table so each entity is serialized once instead of once per
/// path occurrence — the old per-path re-serialization produced responses that
/// blew the transport budget.
fn compact_execution_paths(
    conn: &rusqlite::Connection,
    mut paths: Vec<Vec<String>>,
    path_cap: usize,
) -> Result<CompactPaths, StorageError> {
    paths.sort_by(|a, b| b.len().cmp(&a.len()).then_with(|| a.cmp(b)));
    let path_cap_truncated = paths.len() > path_cap;
    paths.truncate(path_cap);

    let mut node_ids: BTreeSet<String> = BTreeSet::new();
    for path in &paths {
        for id in path {
            node_ids.insert(id.clone());
        }
    }
    // A briefing-blocked node keeps its navigable identity in the node table
    // (clarion-719e7320f5, A3) — flagged `briefing_blocked` — and its id rides
    // through the path arrays unchanged: the secret is the file content, not the
    // structural flow. The sentinel is retained only for the rare high-entropy
    // node whose id is itself secret-like (the guard withholds that one id).
    let mut secret_id: BTreeSet<String> = BTreeSet::new();
    let mut nodes = Vec::new();
    for id in &node_ids {
        if let Some(entity) = entity_by_id(conn, id)? {
            if let Some(reason) = briefing_block_reason(&entity) {
                if name_value_is_secretlike(&entity.id) {
                    secret_id.insert(id.clone());
                } else {
                    nodes.push(compact_blocked_node_json(&entity, &reason));
                }
            } else {
                nodes.push(compact_node_json(conn, &entity));
            }
        }
    }
    if !secret_id.is_empty() {
        for path in &mut paths {
            for id in path.iter_mut() {
                if secret_id.contains(id) {
                    BRIEFING_BLOCKED_PATH_SENTINEL.clone_into(id);
                }
            }
        }
    }
    Ok(CompactPaths {
        nodes,
        paths,
        path_cap_truncated,
    })
}

/// Truncation reason for an execution-path response. `edge-cap` (traversal
/// stopped early, so the graph itself is incomplete) takes precedence over
/// `path-cap` (traversal finished but the ranked output was trimmed for size,
/// clarion-23ae24358c).
fn path_truncation_reason(edge_truncated: bool, path_cap_truncated: bool) -> Option<&'static str> {
    if edge_truncated {
        Some("edge-cap")
    } else if path_cap_truncated {
        Some("path-cap")
    } else {
        None
    }
}

/// A path node trimmed for token economy: identity + location only. The full id
/// already encodes the qualified name; `content_hash` and the redundant `name`
/// are dropped (clarion-5b3eff9a91).
fn compact_node_json(conn: &rusqlite::Connection, entity: &EntityRow) -> Value {
    // REQ-C-04: a path node is the only representation of that entity in the
    // response, so it carries its `sei` (the durable binding key) alongside the
    // mutable `id`/locator. Graceful-degrade to null on a pre-SEI DB.
    json!({
        "id": entity.id,
        "sei": sei_for_locator(conn, &entity.id).ok().flatten(),
        "kind": entity.kind,
        "short_name": entity.short_name,
        "source_file_path": entity.source_file_path,
        "source_line_start": entity.source_line_start,
        "source_line_end": entity.source_line_end
    })
}

/// A path node for a briefing-blocked entity (clarion-719e7320f5, A3): the same
/// navigable shape as [`compact_node_json`] plus the `briefing_blocked` flag.
/// The `sei` cross-tool binding key stays null (no read-time join — and a blocked
/// row is navigable by locator, not bound across siblings). The `short_name`
/// guard re-withholds only the one field if it is itself high-entropy.
fn compact_blocked_node_json(entity: &EntityRow, reason: &str) -> Value {
    json!({
        "id": redact_secretlike(&entity.id),
        "sei": Value::Null,
        "kind": entity.kind,
        "short_name": redact_secretlike(&entity.short_name),
        "source_file_path": entity.source_file_path,
        "source_line_start": entity.source_line_start,
        "source_line_end": entity.source_line_end,
        "briefing_blocked": reason,
    })
}

fn reference_neighbors(
    conn: &rusqlite::Connection,
    entity_id: &str,
    direction: ReferenceDirection,
) -> Result<Vec<Value>, StorageError> {
    edge_neighbors_json(
        conn,
        reference_edges_for_entity(conn, entity_id, direction)?,
    )
}

/// Kind-tagged relation neighbors (`inherits_from` / `decorates` /
/// `implements` / `derives`) for `neighborhood` / `orientation_pack`
/// (clarion-ae5b43ea40). Same entry shape as the references buckets plus the
/// edge `kind`; the cursor-paginated set with anchor evidence is the dedicated
/// `entity_relation_list` tool. Honors the confidence tier (relations are
/// resolved|ambiguous only — stronger claims than `references`, so the
/// resolved default excludes ambiguous rows rather than mixing them in).
fn relation_neighbors(
    conn: &rusqlite::Connection,
    entity_id: &str,
    direction: ReferenceDirection,
    confidence: EdgeConfidence,
) -> Result<Vec<Value>, StorageError> {
    let mut neighbors = Vec::new();
    for edge in relation_edges_for_entity(conn, entity_id, direction, None)? {
        if edge.confidence > confidence {
            continue;
        }
        let neighbor_id = match direction {
            ReferenceDirection::In => &edge.from_id,
            ReferenceDirection::Out => &edge.to_id,
        };
        if let Some(entity) = entity_by_id(conn, neighbor_id)? {
            neighbors.push(json!({
                "kind": edge.kind,
                "entity": entity_json(conn, &entity),
                "edge_confidence": edge.confidence.as_str(),
                "source_byte_start": edge.source_byte_start,
                "source_byte_end": edge.source_byte_end
            }));
        }
    }
    Ok(neighbors)
}

/// `imports`-edge neighbors for a module entity (clarion-79d0ff6e14). Direction
/// `In` is the reverse-import lookup ("who imports this module").
fn import_neighbors(
    conn: &rusqlite::Connection,
    entity_id: &str,
    direction: ReferenceDirection,
) -> Result<Vec<Value>, StorageError> {
    edge_neighbors_json(conn, import_edges_for_entity(conn, entity_id, direction)?)
}

/// Reference neighbors for `neighborhood` / `orientation_pack`, rolled up to
/// file-scope altitude when the entity is a manifest-declared file-scope entity
/// (clarion-79d0ff6e14).
///
/// References are tracked symbol-to-symbol, so a file-scope entity's OWN
/// reference edges are almost always empty — "who imports this module /
/// contract?" used to answer `[]`. For a file-scope entity we instead aggregate
/// the `references` edges of every transitively contained symbol (excluding
/// intra-file wiring) and tag each neighbor with the contained `via` symbol it
/// touches. For any other kind the direct symbol-level edges are returned
/// unchanged (no `via`).
///
/// Returns `(neighbors, rolled_up)`; `rolled_up` is true only for file scopes.
fn reference_neighbors_for(
    conn: &rusqlite::Connection,
    entity: &EntityRow,
    direction: ReferenceDirection,
) -> Result<(Vec<Value>, bool), StorageError> {
    if is_plugin_file_scope_entity(entity) {
        let edges = module_reference_rollup(conn, &entity.id, direction)?;
        Ok((rolled_up_neighbors_json(conn, edges, direction)?, true))
    } else {
        Ok((reference_neighbors(conn, &entity.id, direction)?, false))
    }
}

fn is_plugin_file_scope_entity(entity: &EntityRow) -> bool {
    if entity.plugin_id == "core" {
        return false;
    }
    if let Some(source_file_id) = entity.source_file_id.as_deref()
        && entity.parent_id.as_deref() == Some(source_file_id)
        && source_file_id.starts_with("core:file:")
    {
        return true;
    }
    entity.kind == "module"
        && entity.parent_id.is_none()
        && entity.source_file_path.is_some()
        && entity.source_line_start.is_some()
        && entity.source_line_end.is_some()
}

fn rolled_up_neighbors_json(
    conn: &rusqlite::Connection,
    edges: Vec<RolledUpReferenceEdge>,
    direction: ReferenceDirection,
) -> Result<Vec<Value>, StorageError> {
    let mut neighbors = Vec::new();
    for edge in edges {
        if let Some(entity) = entity_by_id(conn, &edge.neighbor_id)? {
            let via = entity_by_id(conn, &edge.via_id)?;
            let mut object = serde_json::Map::new();
            object.insert("entity".to_owned(), entity_json(conn, &entity));
            object.insert(
                "edge_confidence".to_owned(),
                json!(edge.confidence.as_str()),
            );
            object.insert(
                "source_byte_start".to_owned(),
                json!(edge.source_byte_start),
            );
            object.insert("source_byte_end".to_owned(), json!(edge.source_byte_end));
            // The module-contained symbol this edge actually touches, so a
            // rolled-up "who imports this module" answer names the importer
            // (entity) AND the imported symbol (via).
            object.insert(
                "via".to_owned(),
                via.as_ref()
                    .map_or(Value::Null, |entity| entity_json(conn, entity)),
            );
            // Reverse-import altitude (clarion-79d0ff6e14): the edge is recorded
            // against the importing *symbol*, but "who imports this module /
            // contract" is a module-altitude question, so name the importer's
            // containing module too. Only meaningful for the `In` direction
            // (the neighbor is the importer); `Out` neighbors are referenced
            // symbols, not importers. `null` for an importer with no module
            // ancestor.
            if direction == ReferenceDirection::In {
                let importer_module = match containing_module_id(conn, &edge.neighbor_id)? {
                    Some(module_id) => entity_by_id(conn, &module_id)?,
                    None => None,
                };
                object.insert(
                    "importer_module".to_owned(),
                    importer_module
                        .as_ref()
                        .map_or(Value::Null, |entity| entity_json(conn, entity)),
                );
            }
            neighbors.push(Value::Object(object));
        }
    }
    Ok(neighbors)
}

fn edge_neighbors_json(
    conn: &rusqlite::Connection,
    edges: Vec<ReferenceEdgeMatch>,
) -> Result<Vec<Value>, StorageError> {
    let mut neighbors = Vec::new();
    for edge in edges {
        if let Some(entity) = entity_by_id(conn, &edge.neighbor_id)? {
            neighbors.push(json!({
                "entity": entity_json(conn, &entity),
                "edge_confidence": edge.confidence.as_str(),
                "source_byte_start": edge.source_byte_start,
                "source_byte_end": edge.source_byte_end
            }));
        }
    }
    Ok(neighbors)
}

fn result_response(id: &Value, result: &Value) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": id,
        "result": result
    })
}

fn error_response(id: &Value, code: i64, message: &str) -> Value {
    json!({
        "jsonrpc": "2.0",
        "id": id,
        "error": {
            "code": code,
            "message": message
        }
    })
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::time::Duration;

    use loomweave_core::{CachingModel, LlmProvider, LlmProviderError, LlmRequest, LlmResponse};
    use loomweave_storage::{
        EntityRow, InferredEdgeCacheKey, ReaderPool, UnresolvedCallSiteRow, pragma, schema,
    };
    use rusqlite::Connection;
    use tokio::sync::mpsc;

    use super::{
        InferenceLlmState, InferredRead, McpToolPolicy, RENAME_MAP, ServerState, config::LlmConfig,
        list_tools,
    };
    use loomweave_federation::config::{LlmProviderKind, SemanticProviderKind};

    /// tools/list is loaded into EVERY consuming session, so its size is a
    /// per-session context tax (clarion-e65354898d). Budget: each description
    /// ≤ 350 chars (what it answers, key args, one honesty caveat — depth
    /// belongs in the loomweave-workflow skill, fetched once on demand), and
    /// the serialized catalogue ≤ 23.5 KB (~5.6k tokens at the observed ~4.2
    /// bytes/token; the pre-budget baseline was 46 KB / ~11k tokens). The
    /// budget was raised 22 → 23 KB for weft-ac59e8e730: the two config-set
    /// tools' inputSchemas were bare `{}` per property, and declaring real
    /// types/enums for the surface that can enable writes and live LLM spend
    /// is worth ~0.7 KB. Raised 23 → 23.5 KB for clarion-663aca16aa (A5): the
    /// two heuristic-surface tools (`entity_dead_list`,
    /// `entity_coupling_hotspot_list`) gained the opt-in, validation-bearing
    /// `roots`/`app_only` params (typed schemas + the honesty caveats they
    /// carry), worth ~0.25 KB.
    #[test]
    fn tools_list_fits_the_context_budget() {
        let tools = list_tools();
        for tool in &tools {
            let len = tool.description.chars().count();
            assert!(
                len <= 350,
                "description for {} is {len} chars (budget 350) — move the depth \
                 into the loomweave-workflow skill",
                tool.name
            );
        }
        let serialized = serde_json::to_string(&tools).expect("serialize tools");
        assert!(
            serialized.len() <= 23_500,
            "serialized tools/list is {} bytes (budget 23500 ≈ 5.6k tokens)",
            serialized.len()
        );
    }

    /// The `initialize.instructions` blob is truncated by real clients
    /// (observed live: Claude Code cut it mid-sentence around ~2 KB), so the
    /// write-gating note must come EARLY and the whole blob must stay small
    /// (clarion-e65354898d).
    #[test]
    fn server_instructions_fit_truncating_clients() {
        let instructions = super::server_instructions(McpToolPolicy::read_only());
        assert!(
            instructions.len() <= 2_000,
            "instructions are {} bytes; budget 2000",
            instructions.len()
        );
        let gate = instructions
            .find("rite-gated")
            .expect("write-gating note present in read-only posture");
        assert!(
            gate <= 700,
            "write-gating note starts at byte {gate}; it must sit in the first \
             700 bytes to survive client truncation"
        );
    }

    /// SKILL.md must speak the registered tool dialect (clarion-888434f3ce).
    ///
    /// The skill is the canonical onboarding doc (embedded asset, served via
    /// `prompts/get`, installed by `loomweave install --skills`). An MCP client
    /// exposes only the names in `tools/list`; the `RENAME_MAP` shim rescues
    /// raw JSON-RPC callers but NOT `mcp__loomweave__<old-name>` clients, so a
    /// skill teaching retired names burns failed calls. Three invariants:
    /// no retired alias appears as a backticked token, every
    /// `mcp__loomweave__<name>` mention is registered (or the `*` wildcard),
    /// and every registered tool is documented somewhere in the skill.
    #[test]
    fn skill_md_speaks_the_registered_tool_dialect() {
        let skill = include_str!("../assets/skills/loomweave-workflow/SKILL.md");
        let registered: std::collections::BTreeSet<&str> =
            list_tools().iter().map(|tool| tool.name).collect();

        for &(old, new) in RENAME_MAP {
            if old == new {
                continue;
            }
            assert!(
                !skill.contains(&format!("`{old}`")),
                "SKILL.md teaches retired tool name `{old}` — the registered name is `{new}`"
            );
            assert!(
                !skill.contains(&format!("mcp__loomweave__{old}")),
                "SKILL.md claims mcp__loomweave__{old} exists — clients expose mcp__loomweave__{new}"
            );
        }

        for (idx, _) in skill.match_indices("mcp__loomweave__") {
            let rest = &skill[idx + "mcp__loomweave__".len()..];
            if rest.starts_with('*') {
                continue; // the documented wildcard form
            }
            let name: String = rest
                .chars()
                .take_while(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || *c == '_')
                .collect();
            assert!(
                registered.contains(name.as_str()),
                "SKILL.md mentions mcp__loomweave__{name}, which is not in tools/list"
            );
        }

        for name in &registered {
            assert!(
                skill.contains(&format!("`{name}`")),
                "registered tool `{name}` is undocumented in SKILL.md — document it \
                 (or its containing table row) so the skill covers the live catalogue"
            );
        }
    }

    #[test]
    #[allow(clippy::too_many_lines)] // exhaustively pins all 46 tool docstrings by design
    fn tools_list_exposes_exact_docstrings() {
        let tools = list_tools();

        assert_eq!(tools.len(), 46);
        assert_eq!(tools[0].name, "entity_at");
        assert_eq!(
            tools[0].description,
            "Innermost entity whose source range contains `file`+`line`, with an `entity_context` evidence block (match_reason, containing stack, sub-ranges, alternatives). A line only a module spans reports containing_range — never a fabricated exact match."
        );
        assert_eq!(tools[1].name, "entity_find");
        assert_eq!(
            tools[1].description,
            "Search entities by name, id, summary, and docstring before grepping. Optional `kind`; paginated; scanner-blocked docstrings never match."
        );
        assert_eq!(tools[2].name, "entity_callers_list");
        assert_eq!(
            tools[2].description,
            "List callers. `confidence` resolved (default) | ambiguous | inferred. Bounded. scope_excludes is non-empty ONLY when a candidate was skipped; an empty scope_excludes with traversal_complete:true means every candidate was searched. Skipped sites surface as unresolved_candidates."
        );
        assert_eq!(tools[3].name, "entity_execution_path_list");
        assert_eq!(
            tools[3].description,
            "Bounded calls-only paths from an entity (`max_depth` default 3). Compact: deduplicated `nodes` table + `paths` as node-id arrays, longest-first; truncation reports edge-cap/path-cap. `scope_excludes` names unsearched blind spots (attribute-receiver / unresolved static calls) — empties are not proof."
        );
        assert_eq!(tools[4].name, "entity_summary_get");
        assert_eq!(
            tools[4].description,
            "On-demand cached LLM summary for one entity (leaf scope: module summaries do not aggregate members). Non-JSON LLM output degrades to a cached deterministic structural fallback, so a retry is a cache hit, not a re-billed failure."
        );
        assert_eq!(tools[5].name, "entity_issue_list");
        assert_eq!(
            tools[5].description,
            "Filigree issues attached to an entity (optionally contained entities too), plus an enrich-only `wardline_findings` section. result_kind matched | no_matches | unavailable distinguishes empty-Filigree from unreachable-Filigree; `filigree_endpoint` shows which endpoint answered."
        );
        assert_eq!(tools[6].name, "entity_neighborhood_get");
        assert_eq!(
            tools[6].description,
            "One-hop callers/callees/container/contained/references/imports/relations. Per-bucket limit. scope_excludes is non-empty only when a caller candidate was skipped; empty + traversal_complete:true confirms the callers bucket searched all candidates. Skipped sites surface as unresolved_candidates."
        );
        assert_eq!(tools[7].name, "subsystem_member_list");
        assert_eq!(
            tools[7].description,
            "List the module entities in a subsystem. Bounded: `limit` (default 50, max 100) + numeric-offset `cursor`."
        );
        assert_eq!(tools[8].name, "entity_subsystem_get");
        assert_eq!(
            tools[8].description,
            "The subsystem an entity belongs to (reverse of subsystem_member_list). Any entity id (functions/classes resolve through their containing module); reports the via-module."
        );
        assert_eq!(tools[9].name, "project_status_get");
        assert_eq!(
            tools[9].description,
            "Deterministic diagnostics: repo root, db path, latest run, entity/edge/finding counts, index staleness, per-plugin counts, LLM policy, and the resolved Filigree endpoint."
        );
        assert_eq!(tools[10].name, "llm_config_get");
        assert_eq!(
            tools[10].description,
            "Read LLM/provider/live and MCP write-tool config."
        );
        assert_eq!(tools[11].name, "llm_config_set");
        assert_eq!(
            tools[11].description,
            "Update LLM/provider/live/write-tool config in loomweave.yaml; reconnect."
        );
        assert_eq!(tools[12].name, "semantic_config_get");
        assert_eq!(
            tools[12].description,
            "Read semantic-search config and sidecar diagnostics."
        );
        assert_eq!(tools[13].name, "semantic_config_set");
        assert_eq!(
            tools[13].description,
            "Update semantic-search config in loomweave.yaml; rerun analyze."
        );
        assert_eq!(tools[14].name, "entity_summary_preview_cost_get");
        assert_eq!(
            tools[14].description,
            "Preview what entity_summary_get would cost BEFORE spending: cache_status (hit | expired | miss), real cost on a hit, an estimate on a miss, and live_spend_would_occur. A disabled LLM is reported distinctly from a cache miss. Never invokes the provider."
        );
        assert_eq!(tools[15].name, "entity_source_get");
        assert_eq!(
            tools[15].description,
            "An entity's exact indexed source span plus `context_lines` (default 10) of context, line-numbered and flagged in_entity. source_status reports ok | missing | no_range | no_source_path | binary | drifted (rerun analyze) instead of a misleading stale snippet."
        );
        assert_eq!(tools[16].name, "entity_call_site_list");
        assert_eq!(
            tools[16].description,
            "Show the source sites behind calls/references edges — the evidence for WHY an edge exists. role=caller (default) lists outgoing sites; role=callee lists incoming, INCLUDING `unresolved_sites` (static calls Loomweave could not bind — the recovery surface for suspicious empty caller lists). Bounded."
        );
        assert_eq!(tools[17].name, "entity_orientation_pack_get");
        assert_eq!(
            tools[17].description,
            "One deterministic orientation packet for an `entity` id OR `file`+`line` (exactly one): primary entity, match evidence, one-hop neighbors, execution paths, Filigree issues, Wardline findings, health, suggested next reads. Optional `include` (wardline/findings/issues) folds a `dossier`. Bounded; `omitted` + degraded sections keep empties honest."
        );
        assert_eq!(tools[18].name, "analyze_start");
        assert_eq!(
            tools[18].description,
            "Start a background `loomweave analyze` run; returns run_id immediately (runs can take minutes). One run per project (cross-process lock). Poll analyze_status_get; stop via analyze_cancel. No arguments."
        );
        assert_eq!(tools[19].name, "analyze_status_get");
        assert_eq!(
            tools[19].description,
            "A run's live status: queued | running | completed | failed | cancelled | skipped_no_plugins, with phase, current plugin/file, processed/total files, heartbeat, and progress_observed=false when the heartbeat is stale (run may be wedged)."
        );
        assert_eq!(tools[20].name, "analyze_cancel");
        assert_eq!(
            tools[20].description,
            "Cancel a running analyze: SIGKILLs the run's process group (plugin + pyright) and marks the run cancelled — never left dangling as running. Idempotent. Partial work already written is kept."
        );
        assert_eq!(tools[21].name, "index_diff_get");
        assert_eq!(tools[22].name, "entity_guidance_list");
        assert_eq!(tools[23].name, "propose_guidance");
        assert_eq!(tools[24].name, "promote_guidance");
        assert_eq!(tools[25].name, "entity_finding_list");
        assert_eq!(tools[26].name, "entity_wardline_get");
        assert_eq!(tools[27].name, "entity_tag_list");
        assert_eq!(tools[28].name, "entity_kind_list");
        assert_eq!(tools[29].name, "entity_wardline_list");
        assert_eq!(tools[30].name, "module_circular_import_list");
        assert_eq!(tools[31].name, "entity_coupling_hotspot_list");
        assert_eq!(tools[32].name, "entity_entry_point_list");
        assert_eq!(tools[33].name, "entity_http_route_list");
        assert_eq!(tools[34].name, "entity_data_model_list");
        assert_eq!(tools[35].name, "entity_test_list");
        assert_eq!(tools[36].name, "entity_deprecation_list");
        assert_eq!(tools[37].name, "entity_todo_list");
        assert_eq!(tools[38].name, "entity_test_caller_list");
        assert_eq!(tools[39].name, "entity_high_churn_list");
        assert_eq!(tools[40].name, "entity_recent_change_list");
        assert_eq!(tools[41].name, "entity_dead_list");
        assert_eq!(tools[42].name, "entity_semantic_search_list");
        assert_eq!(tools[43].name, "project_finding_list");
        assert_eq!(tools[44].name, "entity_resolve");
        assert_eq!(tools[45].name, "entity_relation_list");
    }

    #[test]
    fn config_set_tool_schemas_declare_types_for_every_property() {
        // weft-ac59e8e730: the bootstrap config tools shipped with every
        // inputSchema property as a bare `{}`, declaring nothing. Pin that
        // each property is typed, and that the provider enums track the
        // parsers' accepted alias sets.
        for tool_name in ["llm_config_set", "semantic_config_set"] {
            let tools = list_tools();
            let tool = tools
                .iter()
                .find(|tool| tool.name == tool_name)
                .unwrap_or_else(|| panic!("{tool_name} registered"));
            let properties = tool.input_schema["properties"]
                .as_object()
                .expect("schema properties object");
            assert!(!properties.is_empty());
            for (property, schema) in properties {
                assert!(
                    schema["type"].is_string(),
                    "{tool_name}.{property} schema must declare a type, got: {schema}"
                );
            }
            let provider_enum = properties["provider"]["enum"]
                .as_array()
                .unwrap_or_else(|| panic!("{tool_name}.provider must enumerate accepted values"));
            for value in provider_enum {
                let value = value.as_str().expect("enum entries are strings");
                let parsed = match tool_name {
                    "llm_config_set" => LlmProviderKind::parse(value).map(|_| ()),
                    _ => SemanticProviderKind::parse(value).map(|_| ()),
                };
                parsed.unwrap_or_else(|err| {
                    panic!("{tool_name}.provider enum value {value:?} is not parseable: {err}")
                });
            }
        }
    }

    #[test]
    fn server_instructions_enumerate_every_tool() {
        // Single-source guard (clarion-71f0d6c3dd): the `instructions` tool list
        // is derived from list_tools_for_policy under the active policy, so every
        // tool the server actually registers must appear in it — and a write-gated
        // tool must NOT appear when the gate is off (agent-first-feedback §2.5).
        use super::McpToolPolicy;

        // With write tools enabled, every tool is advertised.
        let all = super::server_instructions(McpToolPolicy::allow_write_tools());
        for tool in super::list_tools_for_policy(McpToolPolicy::allow_write_tools()) {
            assert!(
                all.contains(tool.name),
                "instructions omit registered tool {:?}; instructions were:\n{all}",
                tool.name
            );
        }

        // Under the default read-only policy, the advertised list matches the
        // registered list exactly — gated write tools are absent from the list
        // but named in the gate note.
        let read_only = super::server_instructions(McpToolPolicy::default());
        let registered = super::list_tools_for_policy(McpToolPolicy::default());
        for tool in &registered {
            assert!(
                read_only.contains(tool.name),
                "instructions omit registered tool {:?}; instructions were:\n{read_only}",
                tool.name
            );
        }
        assert!(
            registered.len() < super::list_tools().len(),
            "default policy should gate at least one write tool"
        );
        // The gate note names the write tools and how to enable them.
        assert!(read_only.contains("enable_write_tools"), "{read_only}");
        assert!(read_only.contains("entity_summary_get"), "{read_only}");
        // The deliberate bootstrap exemption (PM ruling, weft-ac59e8e730) must
        // be stated loudly: a read-only session can persistently enable writes
        // and live LLM spend through these two tools.
        assert!(read_only.contains("EXEMPT"), "{read_only}");
        assert!(read_only.contains("llm_config_set"), "{read_only}");
        assert!(read_only.contains("semantic_config_set"), "{read_only}");
    }

    #[test]
    fn no_index_initialize_chirps_install_and_analyze() {
        let root = std::path::Path::new("/tmp/demo");
        let request = serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "initialize"});
        let response =
            super::handle_json_rpc_no_index(&request, root).expect("initialize yields a response");
        assert_eq!(
            response["result"]["protocolVersion"],
            super::MCP_PROTOCOL_VERSION
        );
        assert_eq!(response["result"]["serverInfo"]["name"], "loomweave");
        assert!(response["result"]["capabilities"]["tools"].is_object());
        let instructions = response["result"]["instructions"]
            .as_str()
            .expect("instructions present");
        // Both halves of the canonical hook sequence, plus the project path.
        assert!(
            instructions.contains("loomweave install --path /tmp/demo"),
            "instructions: {instructions}"
        );
        assert!(
            instructions.contains("loomweave analyze /tmp/demo"),
            "instructions: {instructions}"
        );
    }

    #[test]
    fn no_index_tools_call_returns_actionable_is_error() {
        let root = std::path::Path::new("/tmp/demo");
        let request = serde_json::json!({
            "jsonrpc": "2.0",
            "id": 2,
            "method": "tools/call",
            "params": {"name": "entity_find", "arguments": {"query": "foo"}}
        });
        let response = super::handle_json_rpc_no_index(&request, root).expect("response");
        // isError is the load-bearing chirp channel — fires the moment the agent
        // touches any tool, regardless of whether the client surfaced instructions.
        assert_eq!(response["result"]["isError"], serde_json::json!(true));
        let text = response["result"]["content"][0]["text"]
            .as_str()
            .expect("tool result text");
        assert!(
            text.contains("loomweave analyze /tmp/demo"),
            "tool chirp text: {text}"
        );
    }

    #[test]
    fn no_index_tools_list_still_advertises_tools() {
        let root = std::path::Path::new("/tmp/demo");
        let request = serde_json::json!({"jsonrpc": "2.0", "id": 3, "method": "tools/list"});
        let response = super::handle_json_rpc_no_index(&request, root).expect("response");
        let tools = response["result"]["tools"].as_array().expect("tools array");
        assert!(
            !tools.is_empty(),
            "degraded tools/list should still advertise the surface"
        );
    }

    #[test]
    fn no_index_ignores_notifications() {
        let root = std::path::Path::new("/tmp/demo");
        // The client sends notifications/initialized right after initialize; it
        // has no id and must draw no response.
        let request = serde_json::json!({"jsonrpc": "2.0", "method": "notifications/initialized"});
        assert!(super::handle_json_rpc_no_index(&request, root).is_none());
    }

    #[test]
    fn serve_stdio_no_index_round_trips_initialize_over_json_line() {
        let root = std::path::Path::new("/tmp/demo");
        let input = b"{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\"}\n";
        let mut reader = std::io::BufReader::new(&input[..]);
        let mut output = Vec::new();
        super::serve_stdio_no_index(root, &mut reader, &mut output).expect("degraded serve");
        let response: serde_json::Value = serde_json::from_slice(&output).expect("framed json");
        let instructions = response["result"]["instructions"]
            .as_str()
            .expect("instructions present");
        assert!(
            instructions.contains("loomweave analyze /tmp/demo"),
            "instructions: {instructions}"
        );
    }

    #[test]
    fn initialize_returns_server_info_and_tools_capability() {
        let response = super::handle_json_rpc(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2025-11-25",
                "capabilities": {},
                "clientInfo": {"name": "test-client", "version": "0.0.0"}
            }
        }))
        .expect("initialize request returns a response");

        assert_eq!(response["jsonrpc"], "2.0");
        assert_eq!(response["id"], 1);
        assert_eq!(
            response["result"]["protocolVersion"],
            super::MCP_PROTOCOL_VERSION
        );
        assert_eq!(response["result"]["serverInfo"]["name"], "loomweave");
        assert!(response["result"]["capabilities"]["tools"].is_object());
        // Orientation instructions present and mention the skill + entity model.
        let instructions = response["result"]["instructions"]
            .as_str()
            .expect("initialize result has instructions");
        assert!(
            instructions.contains("loomweave-workflow"),
            "instructions should point at the skill"
        );
        assert!(
            instructions.contains("entity"),
            "instructions should describe the entity model"
        );
        // The stateless free handler does NOT serve resources/* or prompts/*,
        // so it must not advertise those capabilities (a client enabling those
        // flows against the stateless server would otherwise fail).
        assert!(
            response["result"]["capabilities"]["prompts"].is_null(),
            "stateless initialize must not advertise prompts: {response:?}"
        );
        assert!(
            response["result"]["capabilities"]["resources"].is_null(),
            "stateless initialize must not advertise resources: {response:?}"
        );
    }

    #[test]
    fn optional_confidence_rejects_non_string_json_types() {
        for value in [
            serde_json::Value::Null,
            serde_json::json!(1),
            serde_json::json!(["resolved"]),
            serde_json::json!({"tier": "resolved"}),
        ] {
            let mut arguments = serde_json::Map::new();
            arguments.insert("confidence".to_owned(), value);
            let err = super::optional_confidence(&arguments).expect_err("non-string must reject");
            assert_eq!(err.message, "confidence must be a string");
        }
    }

    #[test]
    fn optional_confidence_rejects_unknown_string() {
        let mut arguments = serde_json::Map::new();
        arguments.insert("confidence".to_owned(), serde_json::json!("all"));
        let err = super::optional_confidence(&arguments).expect_err("unknown string must reject");
        assert_eq!(
            err.message,
            "confidence must be one of resolved, ambiguous, inferred"
        );
    }

    #[tokio::test]
    async fn stateful_initialize_advertises_prompts_and_resources() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-11-25",
                    "capabilities": {},
                    "clientInfo": {"name": "test-client", "version": "0.0.0"}
                }
            }))
            .await
            .expect("initialize request returns a response");

        // The production (ServerState) path serves prompts/* and resources/*,
        // so its initialize must advertise the full capability surface.
        assert!(response["result"]["capabilities"]["tools"].is_object());
        assert!(
            response["result"]["capabilities"]["prompts"].is_object(),
            "stateful initialize must advertise prompts: {response:?}"
        );
        assert!(
            response["result"]["capabilities"]["resources"].is_object(),
            "stateful initialize must advertise resources: {response:?}"
        );
        let instructions = response["result"]["instructions"]
            .as_str()
            .expect("initialize result has instructions");
        assert!(
            instructions.contains("loomweave-workflow"),
            "instructions should point at the skill"
        );
    }

    #[tokio::test]
    async fn resources_list_includes_loomweave_context() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 1,
                "method": "resources/list",
                "params": {}
            }))
            .await
            .expect("response");

        let resources = response["result"]["resources"].as_array().unwrap();
        assert!(
            resources.iter().any(|r| r["uri"] == "loomweave://context"),
            "loomweave://context not listed: {resources:?}"
        );
    }

    #[tokio::test]
    async fn resources_read_returns_context_snapshot_json() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
            conn.execute(
                "INSERT INTO entities \
                 (id, plugin_id, kind, name, short_name, properties, created_at, updated_at) \
                 VALUES ('python:module:m','python','module','m','m','{}', \
                         '2026-01-01T00:00:00.000Z','2026-01-01T00:00:00.000Z')",
                [],
            )
            .unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 7,
                "method": "resources/read",
                "params": {"uri": "loomweave://context"}
            }))
            .await
            .expect("response");

        let text = response["result"]["contents"][0]["text"]
            .as_str()
            .expect("snapshot text");
        let parsed: serde_json::Value = serde_json::from_str(text).unwrap();
        assert_eq!(parsed["db_present"], true);
        assert_eq!(parsed["entity_count"], 1);
        assert_eq!(parsed["staleness"], "never_analyzed");
        // A healthy read carries `degraded: false`, distinguishing it from the
        // reader-error fallback which sets `degraded: true`.
        assert_eq!(
            parsed["degraded"], false,
            "healthy snapshot must not be degraded"
        );
    }

    #[tokio::test]
    async fn resources_read_rejects_unknown_uri() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 8,
                "method": "resources/read",
                "params": {"uri": "loomweave://nope"}
            }))
            .await
            .expect("response");
        assert!(response["error"].is_object(), "expected an error envelope");
        assert_eq!(response["error"]["code"], -32602, "{response:?}");
    }

    #[tokio::test]
    async fn prompts_get_rejects_unknown_name() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 11,
                "method": "prompts/get",
                "params": {"name": "nope"}
            }))
            .await
            .expect("response");
        assert!(response["error"].is_object(), "expected an error envelope");
        assert_eq!(response["error"]["code"], -32602, "{response:?}");
    }

    #[tokio::test]
    async fn prompts_get_returns_skill_text() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 9,
                "method": "prompts/get",
                "params": {"name": "loomweave-workflow"}
            }))
            .await
            .expect("response");
        let text = response["result"]["messages"][0]["content"]["text"]
            .as_str()
            .unwrap();
        assert!(
            text.contains("name: loomweave-workflow"),
            "not the skill text"
        );
    }

    #[tokio::test]
    async fn prompts_list_includes_loomweave_workflow() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 10,
                "method": "prompts/list",
                "params": {}
            }))
            .await
            .expect("response");
        let prompts = response["result"]["prompts"].as_array().unwrap();
        assert!(prompts.iter().any(|p| p["name"] == "loomweave-workflow"));
    }

    #[test]
    fn tools_list_request_wraps_read_only_tools_by_default() {
        let response = super::handle_json_rpc(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": "tools-1",
            "method": "tools/list",
            "params": {}
        }))
        .expect("tools/list request returns a response");

        assert_eq!(response["jsonrpc"], "2.0");
        assert_eq!(response["id"], "tools-1");
        let tools = response["result"]["tools"].as_array().unwrap();
        let tool_names: Vec<&str> = tools
            .iter()
            .filter_map(|tool| tool.get("name").and_then(serde_json::Value::as_str))
            .collect();
        assert!(tool_names.contains(&"entity_at"));
        assert!(!tool_names.contains(&"analyze_start"));
        assert!(!tool_names.contains(&"propose_guidance"));
        assert!(!tool_names.contains(&"promote_guidance"));
        assert_eq!(response["result"]["tools"][0]["name"], "entity_at");
        assert_eq!(
            response["result"]["tools"][0]["metadata"]["read_only"],
            true
        );
        // False flags are omitted on the wire (absent ⇒ false).
        assert!(
            response["result"]["tools"][0]["metadata"]
                .get("writes_local_state")
                .is_none()
        );
        assert!(
            tool_names.contains(&"subsystem_member_list"),
            "read-only list should include subsystem_member_list: {tool_names:?}"
        );
    }

    #[tokio::test]
    async fn server_policy_hides_and_blocks_write_tools_by_default() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers);

        let listed = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "tools-readonly",
                "method": "tools/list",
                "params": {}
            }))
            .await
            .expect("tools/list response");
        let names: Vec<&str> = listed["result"]["tools"]
            .as_array()
            .unwrap()
            .iter()
            .filter_map(|tool| tool["name"].as_str())
            .collect();
        assert!(names.contains(&"entity_at"));
        assert!(names.contains(&"llm_config_get"));
        assert!(names.contains(&"llm_config_set"));
        assert!(names.contains(&"semantic_config_get"));
        assert!(names.contains(&"semantic_config_set"));
        assert!(!names.contains(&"analyze_start"));
        assert!(!names.contains(&"entity_summary_get"));

        let blocked = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "blocked",
                "method": "tools/call",
                "params": {"name": "analyze_start", "arguments": {}}
            }))
            .await
            .expect("tools/call response");
        assert_eq!(blocked["error"]["code"], -32601);
        assert!(
            blocked["error"]["message"]
                .as_str()
                .unwrap()
                .contains("tool disabled by MCP tool policy")
        );
    }

    #[tokio::test]
    async fn llm_config_set_bootstraps_provider_and_write_tools_under_read_only_policy() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::read_only());

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "llm-config-set",
                "method": "tools/call",
                "params": {
                    "name": "llm_config_set",
                    "arguments": {
                        "enabled": true,
                        "provider": "codex_sidecar",
                        "allow_live_provider": true,
                        "enable_write_tools": true,
                        "codex_model": "gpt-5-codex"
                    }
                }
            }))
            .await
            .expect("tools/call response");
        assert_eq!(response["result"]["isError"], false, "{response}");
        let text = response["result"]["content"][0]["text"].as_str().unwrap();
        let envelope: serde_json::Value = serde_json::from_str(text).unwrap();
        assert_eq!(envelope["ok"], true, "{envelope}");
        assert_eq!(envelope["result"]["llm"]["provider"], "codex_cli");
        assert_eq!(
            envelope["result"]["serve"]["mcp"]["enable_write_tools"],
            true
        );
        assert_eq!(
            envelope["result"]["active_session"]["enable_write_tools"], false,
            "the current server policy should not mutate mid-session"
        );
        assert_eq!(
            envelope["result"]["active_session"]["restart_required"],
            true
        );

        let saved =
            loomweave_federation::config::McpConfig::from_path(&dir.path().join("loomweave.yaml"))
                .unwrap();
        assert!(saved.llm.enabled);
        assert!(saved.llm.allow_live_provider);
        assert_eq!(
            saved.llm.provider,
            loomweave_federation::config::LlmProviderKind::CodexCli
        );
        assert_eq!(saved.llm.codex_cli.model.as_deref(), Some("gpt-5-codex"));
        assert!(saved.serve.mcp.enable_write_tools);
    }

    #[tokio::test]
    async fn semantic_config_set_bootstraps_local_embeddings_under_read_only_policy() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::read_only());

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "semantic-config-set",
                "method": "tools/call",
                "params": {
                    "name": "semantic_config_set",
                    "arguments": {
                        "enabled": true,
                        "provider": "local_openai",
                        "endpoint_url": "http://127.0.0.1:11434/v1",
                        "model_id": "nomic-embed-text",
                        "dimensions": 768
                    }
                }
            }))
            .await
            .expect("tools/call response");
        assert_eq!(response["result"]["isError"], false, "{response}");
        let text = response["result"]["content"][0]["text"].as_str().unwrap();
        let envelope: serde_json::Value = serde_json::from_str(text).unwrap();
        assert_eq!(envelope["ok"], true, "{envelope}");
        assert_eq!(
            envelope["result"]["semantic_search"]["provider"],
            "local_openai"
        );
        assert_eq!(
            envelope["result"]["semantic_search"]["provider_available"],
            true
        );
        assert_eq!(
            envelope["result"]["active_session"]["restart_required"],
            true
        );

        let saved =
            loomweave_federation::config::McpConfig::from_path(&dir.path().join("loomweave.yaml"))
                .unwrap();
        assert!(saved.semantic_search.enabled);
        assert_eq!(
            saved.semantic_search.provider,
            loomweave_federation::config::SemanticProviderKind::LocalOpenAi
        );
        assert_eq!(saved.semantic_search.dimensions, 768);
    }

    #[test]
    fn semantic_status_sidecar_path_follows_store_dir_override() {
        // The semantic status probe must report the sidecar at the configured
        // `[loomweave].store_dir`, not the hardcoded default — otherwise an
        // operator with a relocated store sees a populated sidecar as absent
        // (read-vs-status parity with `EmbeddingStore::open_in_store_dir`).
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path();
        std::fs::write(
            root.join(loomweave_core::store::WEFT_TOML),
            "[loomweave]\nstore_dir = \"custom/store\"\n",
        )
        .unwrap();

        // Populate a sidecar at the override location.
        std::fs::create_dir_all(loomweave_core::store::store_dir(root)).unwrap();
        let store = loomweave_storage::EmbeddingStore::open_in_store_dir(root).unwrap();
        store
            .upsert(
                &loomweave_storage::EmbeddingKey {
                    entity_id: "python:function:m.f".to_owned(),
                    content_hash: "h".to_owned(),
                    model_id: "model".to_owned(),
                },
                &[0.1, 0.2, 0.3],
                0.0,
                0,
                "2026-06-15T00:00:00Z",
            )
            .unwrap();
        drop(store);

        let status = super::semantic_config_status_json(
            &root.join("loomweave.yaml"),
            root,
            true,
            &loomweave_federation::config::McpConfig::default(),
        );
        let sidecar = &status["embeddings_sidecar"];
        assert_eq!(
            sidecar["path"],
            root.join("custom/store/embeddings.db")
                .display()
                .to_string(),
            "status must probe the overridden store_dir, not the default path"
        );
        assert_eq!(sidecar["present"], true);
        assert_eq!(sidecar["vector_count"], 1);
    }

    #[tokio::test]
    async fn server_policy_can_advertise_write_tools() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::allow_write_tools());

        let listed = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "tools-write",
                "method": "tools/list",
                "params": {}
            }))
            .await
            .expect("tools/list response");
        let tools = listed["result"]["tools"].as_array().unwrap();
        let analyze = tools
            .iter()
            .find(|tool| tool["name"] == "analyze_start")
            .expect("analyze_start advertised when write tools enabled");
        // False flags are omitted on the wire (absent ⇒ false).
        assert!(analyze["metadata"].get("read_only").is_none());
        assert_eq!(analyze["metadata"]["writes_local_state"], true);
        assert_eq!(analyze["metadata"]["spawns_process"], true);
        assert!(tools.iter().any(|tool| tool["name"] == "propose_guidance"));
    }

    #[tokio::test]
    async fn stateful_tools_list_filters_write_tools_when_policy_disables_them() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::read_only());

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "policy-list",
                "method": "tools/list",
                "params": {}
            }))
            .await
            .expect("response");
        let tools = response["result"]["tools"].as_array().unwrap();
        let tool_names: Vec<&str> = tools
            .iter()
            .filter_map(|tool| tool.get("name").and_then(serde_json::Value::as_str))
            .collect();
        assert!(tool_names.contains(&"project_status_get"));
        assert!(tool_names.contains(&"entity_callers_list"));
        assert!(!tool_names.contains(&"analyze_start"));
        assert!(!tool_names.contains(&"analyze_cancel"));
        assert!(!tool_names.contains(&"entity_summary_get"));
        assert!(!tool_names.contains(&"propose_guidance"));
        assert!(!tool_names.contains(&"promote_guidance"));
        let callers = tools
            .iter()
            .find(|tool| tool["name"] == "entity_callers_list")
            .expect("callers tool still listed");
        assert_eq!(callers["metadata"]["read_only"], true);
        assert_eq!(callers["metadata"]["may_call_llm"], true);
    }

    #[tokio::test]
    async fn stateful_tools_call_rejects_disabled_write_tool() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::read_only());

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "policy-call",
                "method": "tools/call",
                "params": {"name": "analyze_start", "arguments": {}}
            }))
            .await
            .expect("response");
        assert_eq!(response["error"]["code"], -32601);
        assert!(
            response["error"]["message"]
                .as_str()
                .unwrap()
                .contains("disabled by MCP tool policy"),
            "{response}"
        );
    }

    #[tokio::test]
    async fn stateful_tools_call_rejects_inferred_confidence_when_write_tools_disabled() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::read_only());

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "policy-inferred",
                "method": "tools/call",
                "params": {
                    "name": "entity_callers_list",
                    "arguments": {"id": "python:function:demo.f", "confidence": "inferred"}
                }
            }))
            .await
            .expect("response");
        assert_eq!(response["error"]["code"], -32602);
        assert!(
            response["error"]["message"]
                .as_str()
                .unwrap()
                .contains("confidence=inferred"),
            "{response}"
        );
    }

    #[tokio::test]
    async fn stateful_tools_call_rejects_malformed_confidence_arguments() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::allow_write_tools());

        for (idx, (value, expected)) in [
            (serde_json::Value::Null, "confidence must be a string"),
            (serde_json::json!(1), "confidence must be a string"),
            (
                serde_json::json!(["resolved"]),
                "confidence must be a string",
            ),
            (
                serde_json::json!({"tier": "resolved"}),
                "confidence must be a string",
            ),
            (
                serde_json::json!("all"),
                "confidence must be one of resolved, ambiguous, inferred",
            ),
        ]
        .into_iter()
        .enumerate()
        {
            let response = state
                .handle_json_rpc(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": format!("bad-confidence-{idx}"),
                    "method": "tools/call",
                    "params": {
                        "name": "entity_callers_list",
                        "arguments": {
                            "id": "python:function:demo.f",
                            "confidence": value
                        }
                    }
                }))
                .await
                .expect("response");
            assert_eq!(response["error"]["code"], -32602, "{response}");
            assert_eq!(response["error"]["message"], expected, "{response}");
        }
    }

    #[tokio::test]
    async fn stateful_tools_call_rejects_unknown_arguments_from_strict_schema() {
        let dir = tempfile::tempdir().unwrap();
        let db = dir.path().join("loomweave.db");
        {
            let mut conn = rusqlite::Connection::open(&db).unwrap();
            pragma::apply_write_pragmas(&conn).unwrap();
            schema::apply_migrations(&mut conn).unwrap();
        }
        let readers = ReaderPool::open(&db, 4).unwrap();
        let state = ServerState::new(dir.path().to_path_buf(), readers)
            .with_tool_policy(McpToolPolicy::allow_write_tools());

        let response = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": "unknown-argument",
                "method": "tools/call",
                "params": {
                    "name": "entity_summary_get",
                    "arguments": {
                        "id": "python:function:demo.f",
                        "safety_override": true
                    }
                }
            }))
            .await
            .expect("response");
        assert_eq!(response["error"]["code"], -32602, "{response}");
        assert_eq!(
            response["error"]["message"],
            "unknown argument for entity_summary_get: safety_override",
            "{response}"
        );
    }

    #[test]
    fn unknown_method_is_json_rpc_method_not_found() {
        let response = super::handle_json_rpc(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 7,
            "method": "not/real",
            "params": {}
        }))
        .expect("unknown request returns a JSON-RPC error response");

        assert_eq!(response["error"]["code"], -32601);
    }

    #[test]
    fn json_rpc_notification_does_not_return_response() {
        let response = super::handle_json_rpc(&serde_json::json!({
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
            "params": {}
        }));

        assert!(response.is_none());
    }

    #[test]
    fn stateless_call_tool_requires_server_state_before_tool_validation() {
        let response = super::handle_json_rpc(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 8,
            "method": "tools/call",
            "params": {"name": "not_a_tool", "arguments": {}}
        }))
        .expect("tools/call request returns a response");

        assert_eq!(response["error"]["code"], -32601);
        assert_eq!(
            response["error"]["message"],
            "tools/call requires ServerState::handle_json_rpc"
        );
    }

    #[test]
    fn stateless_json_rpc_does_not_fake_tool_calls() {
        let response = super::handle_json_rpc(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 88,
            "method": "tools/call",
            "params": {"name": "summary", "arguments": {"id": "python:function:demo.entry"}}
        }))
        .expect("tools/call request returns a response");

        assert_eq!(response["error"]["code"], -32601);
        assert!(
            response["error"]["message"]
                .as_str()
                .unwrap()
                .contains("ServerState")
        );
    }

    #[test]
    fn stateless_call_tool_with_invalid_params_requires_server_state() {
        let response = super::handle_json_rpc(&serde_json::json!({
            "jsonrpc": "2.0",
            "id": 9,
            "method": "tools/call",
            "params": {"arguments": {}}
        }))
        .expect("tools/call request returns a response");

        assert_eq!(response["error"]["code"], -32601);
        assert_eq!(
            response["error"]["message"],
            "tools/call requires ServerState::handle_json_rpc"
        );
    }

    #[test]
    fn frame_dispatch_decodes_and_reencodes_json_rpc() {
        let frame = loomweave_core::plugin::Frame {
            body: serde_json::to_vec(&serde_json::json!({
                "jsonrpc": "2.0",
                "id": 10,
                "method": "tools/list",
                "params": {}
            }))
            .unwrap(),
        };

        let response = super::handle_frame(&frame)
            .unwrap()
            .expect("request frame returns a response");
        let decoded: serde_json::Value = serde_json::from_slice(&response.body).unwrap();

        assert_eq!(decoded["jsonrpc"], "2.0");
        assert_eq!(decoded["id"], 10);
        let tools = decoded["result"]["tools"].as_array().unwrap();
        let tool_names: Vec<&str> = tools
            .iter()
            .filter_map(|tool| tool.get("name").and_then(serde_json::Value::as_str))
            .collect();
        assert!(tool_names.contains(&"entity_at"));
        assert!(!tool_names.contains(&"propose_guidance"));
        assert!(!tool_names.contains(&"promote_guidance"));
    }

    #[test]
    fn frame_dispatch_returns_none_for_json_rpc_notifications() {
        let frame = loomweave_core::plugin::Frame {
            body: serde_json::to_vec(&serde_json::json!({
                "jsonrpc": "2.0",
                "method": "notifications/initialized",
                "params": {}
            }))
            .unwrap(),
        };

        let response = super::handle_frame(&frame).unwrap();

        assert!(response.is_none());
    }

    #[test]
    fn serve_stdio_handles_multiple_content_length_frames() {
        let mut input = Vec::new();
        loomweave_core::plugin::write_frame(
            &mut input,
            &loomweave_core::plugin::Frame {
                body: serde_json::to_vec(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": 11,
                    "method": "initialize",
                    "params": {
                        "protocolVersion": "2025-11-25",
                        "capabilities": {},
                        "clientInfo": {"name": "test-client", "version": "0.0.0"}
                    }
                }))
                .unwrap(),
            },
        )
        .unwrap();
        loomweave_core::plugin::write_frame(
            &mut input,
            &loomweave_core::plugin::Frame {
                body: serde_json::to_vec(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": 12,
                    "method": "tools/list",
                    "params": {}
                }))
                .unwrap(),
            },
        )
        .unwrap();

        let mut reader = std::io::BufReader::new(std::io::Cursor::new(input));
        let mut output = Vec::new();

        super::serve_stdio(&mut reader, &mut output).unwrap();

        let mut response_reader = std::io::BufReader::new(std::io::Cursor::new(output));
        let first = loomweave_core::plugin::read_frame(
            &mut response_reader,
            loomweave_core::plugin::ContentLengthCeiling::new(usize::MAX),
        )
        .unwrap();
        let second = loomweave_core::plugin::read_frame(
            &mut response_reader,
            loomweave_core::plugin::ContentLengthCeiling::new(usize::MAX),
        )
        .unwrap();
        let first_json: serde_json::Value = serde_json::from_slice(&first.body).unwrap();
        let second_json: serde_json::Value = serde_json::from_slice(&second.body).unwrap();

        assert_eq!(first_json["id"], 11);
        assert_eq!(first_json["result"]["serverInfo"]["name"], "loomweave");
        assert_eq!(second_json["id"], 12);
        let tools = second_json["result"]["tools"].as_array().unwrap();
        let tool_names: Vec<&str> = tools
            .iter()
            .filter_map(|tool| tool.get("name").and_then(serde_json::Value::as_str))
            .collect();
        assert!(tool_names.contains(&"entity_at"));
        assert!(!tool_names.contains(&"propose_guidance"));
        assert!(!tool_names.contains(&"promote_guidance"));
    }

    #[test]
    fn serve_stdio_ignores_json_rpc_notifications() {
        let input = notification_sequence_input(13, 14);
        let mut reader = std::io::BufReader::new(std::io::Cursor::new(input));
        let mut output = Vec::new();

        super::serve_stdio(&mut reader, &mut output).unwrap();
        assert_notification_sequence_responses(output, 13, 14);
    }

    #[test]
    fn serve_stdio_with_state_ignores_json_rpc_notifications() {
        let project = tempfile::tempdir().expect("temp project");
        let db_path = project.path().join("loomweave.db");
        let mut conn = Connection::open(&db_path).expect("open sqlite");
        pragma::apply_write_pragmas(&conn).expect("write pragmas");
        schema::apply_migrations(&mut conn).expect("apply migrations");
        drop(conn);

        let readers = ReaderPool::open(&db_path, 1).expect("reader pool");
        let state = ServerState::new(project.path().to_path_buf(), readers);
        let input = notification_sequence_input(15, 16);
        let mut reader = std::io::BufReader::new(std::io::Cursor::new(input));
        let mut output = Vec::new();

        super::serve_stdio_with_state(&state, &mut reader, &mut output).unwrap();
        assert_notification_sequence_responses(output, 15, 16);
    }

    #[test]
    fn serve_stdio_with_state_uses_json_line_transport_for_json_line_requests() {
        let project = tempfile::tempdir().expect("temp project");
        let db_path = project.path().join("loomweave.db");
        let mut conn = Connection::open(&db_path).expect("open sqlite");
        pragma::apply_write_pragmas(&conn).expect("write pragmas");
        schema::apply_migrations(&mut conn).expect("apply migrations");
        drop(conn);

        let readers = ReaderPool::open(&db_path, 1).expect("reader pool");
        let state = ServerState::new(project.path().to_path_buf(), readers);
        let input = notification_sequence_json_lines(17, 18);
        let mut reader = std::io::BufReader::new(std::io::Cursor::new(input));
        let mut output = Vec::new();

        super::serve_stdio_with_state(&state, &mut reader, &mut output).unwrap();
        assert_notification_sequence_json_lines(output, 17, 18);
    }

    fn notification_sequence_input(initialize_id: u64, tools_list_id: u64) -> Vec<u8> {
        let mut input = Vec::new();
        loomweave_core::plugin::write_frame(
            &mut input,
            &loomweave_core::plugin::Frame {
                body: serde_json::to_vec(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": initialize_id,
                    "method": "initialize",
                    "params": {
                        "protocolVersion": "2025-11-25",
                        "capabilities": {},
                        "clientInfo": {"name": "test-client", "version": "0.0.0"}
                    }
                }))
                .unwrap(),
            },
        )
        .unwrap();
        loomweave_core::plugin::write_frame(
            &mut input,
            &loomweave_core::plugin::Frame {
                body: serde_json::to_vec(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "method": "notifications/initialized",
                    "params": {}
                }))
                .unwrap(),
            },
        )
        .unwrap();
        loomweave_core::plugin::write_frame(
            &mut input,
            &loomweave_core::plugin::Frame {
                body: serde_json::to_vec(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": tools_list_id,
                    "method": "tools/list",
                    "params": {}
                }))
                .unwrap(),
            },
        )
        .unwrap();
        input
    }

    fn assert_notification_sequence_responses(
        output: Vec<u8>,
        initialize_id: u64,
        tools_list_id: u64,
    ) {
        let mut response_reader = std::io::BufReader::new(std::io::Cursor::new(output));
        let first = loomweave_core::plugin::read_frame(
            &mut response_reader,
            loomweave_core::plugin::ContentLengthCeiling::new(usize::MAX),
        )
        .unwrap();
        let second = loomweave_core::plugin::read_frame(
            &mut response_reader,
            loomweave_core::plugin::ContentLengthCeiling::new(usize::MAX),
        )
        .unwrap();
        let first_json: serde_json::Value = serde_json::from_slice(&first.body).unwrap();
        let second_json: serde_json::Value = serde_json::from_slice(&second.body).unwrap();

        assert_eq!(first_json["id"], initialize_id);
        assert_eq!(second_json["id"], tools_list_id);
        assert!(
            loomweave_core::plugin::read_frame(
                &mut response_reader,
                loomweave_core::plugin::ContentLengthCeiling::new(usize::MAX),
            )
            .is_err(),
            "notifications must not produce JSON-RPC response frames"
        );
    }

    fn notification_sequence_json_lines(initialize_id: u64, tools_list_id: u64) -> Vec<u8> {
        let messages = [
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": initialize_id,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2025-11-25",
                    "capabilities": {},
                    "clientInfo": {"name": "test-client", "version": "0.0.0"}
                }
            }),
            serde_json::json!({
                "jsonrpc": "2.0",
                "method": "notifications/initialized",
                "params": {}
            }),
            serde_json::json!({
                "jsonrpc": "2.0",
                "id": tools_list_id,
                "method": "tools/list",
                "params": {}
            }),
        ];
        let mut input = Vec::new();
        for message in messages {
            serde_json::to_writer(&mut input, &message).expect("serialize json line");
            input.push(b'\n');
        }
        input
    }

    fn assert_notification_sequence_json_lines(
        output: Vec<u8>,
        initialize_id: u64,
        tools_list_id: u64,
    ) {
        let output = String::from_utf8(output).expect("json lines are utf8");
        let lines: Vec<_> = output.lines().collect();
        assert_eq!(
            lines.len(),
            2,
            "notifications must not produce JSON-RPC response lines"
        );
        let first_json: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
        let second_json: serde_json::Value = serde_json::from_str(lines[1]).unwrap();

        assert_eq!(first_json["id"], initialize_id);
        assert_eq!(second_json["id"], tools_list_id);
    }

    #[tokio::test]
    async fn inferred_inflight_entry_is_removed_when_leader_future_is_aborted() {
        let project = tempfile::tempdir().expect("temp project");
        let db_path = project.path().join("loomweave.db");
        let mut conn = Connection::open(&db_path).expect("open sqlite");
        pragma::apply_write_pragmas(&conn).expect("write pragmas");
        schema::apply_migrations(&mut conn).expect("apply migrations");
        drop(conn);

        let readers = ReaderPool::open(&db_path, 1).expect("reader pool");
        let state = Arc::new(ServerState::new(project.path().to_path_buf(), readers));
        let key = inferred_test_key();
        let read = inferred_test_read(key.clone());
        let (writer, _rx) = mpsc::channel(1);
        let (release_tx, release_rx) = tokio::sync::mpsc::channel(1);
        let llm = InferenceLlmState {
            writer,
            config: LlmConfig::default(),
            provider: Arc::new(BlockingProvider {
                release: tokio::sync::Mutex::new(release_rx),
                started: None,
            }),
        };

        let leader_state = Arc::clone(&state);
        let leader_key = key.clone();
        let handle = tokio::spawn(async move {
            leader_state
                .coalesced_inferred_dispatch(leader_key, read, llm)
                .await
        });
        assert_inferred_inflight_contains(&state, &key).await;

        handle.abort();
        let _ = handle.await;
        let removed = wait_until_inferred_inflight_removed(&state, &key).await;
        let _ = release_tx.send(()).await;

        assert!(
            removed,
            "aborted inferred-dispatch leader left stale in-flight key"
        );
    }

    #[tokio::test]
    async fn cancellation_notification_aborts_in_flight_llm_request() {
        let project = tempfile::tempdir().expect("temp project");
        let source_path = project.path().join("demo.py");
        std::fs::write(&source_path, "def target():\n    return 1\n").expect("write source");
        let content_hash = blake3::hash("def target():\n    return 1".as_bytes())
            .to_hex()
            .to_string();
        let db_path = project.path().join("loomweave.db");
        let mut conn = Connection::open(&db_path).expect("open sqlite");
        pragma::apply_write_pragmas(&conn).expect("write pragmas");
        schema::apply_migrations(&mut conn).expect("apply migrations");
        conn.execute(
            "INSERT INTO entities (
                id, plugin_id, kind, name, short_name, source_file_path,
                source_line_start, source_line_end, properties, content_hash,
                created_at, updated_at
             ) VALUES (
                'python:function:demo.target', 'python', 'function', 'target', 'target', ?1,
                1, 2, '{}', ?2,
                strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
             )",
            rusqlite::params![source_path.display().to_string(), content_hash],
        )
        .expect("insert entity");
        drop(conn);

        let readers = ReaderPool::open(&db_path, 1).expect("reader pool");
        let (writer, _rx) = mpsc::channel(1);
        let (started_tx, mut started_rx) = tokio::sync::mpsc::channel(1);
        let (_release_tx, release_rx) = tokio::sync::mpsc::channel(1);
        let config = LlmConfig {
            enabled: true,
            allow_live_provider: true,
            ..LlmConfig::default()
        };
        let state = Arc::new(
            ServerState::new(project.path().to_path_buf(), readers)
                .with_tool_policy(McpToolPolicy::allow_write_tools())
                .with_summary_llm(
                    writer,
                    config,
                    Arc::new(BlockingProvider {
                        started: Some(started_tx),
                        release: tokio::sync::Mutex::new(release_rx),
                    }),
                ),
        );

        let request_state = Arc::clone(&state);
        let handle = tokio::spawn(async move {
            request_state
                .handle_json_rpc(&serde_json::json!({
                    "jsonrpc": "2.0",
                    "id": 99,
                    "method": "tools/call",
                    "params": {
                        "name": "entity_summary_get",
                        "arguments": {"id": "python:function:demo.target"}
                    }
                }))
                .await
                .expect("cancelled response")
        });
        wait_until_active_request(&state, "n:99").await;
        tokio::time::timeout(Duration::from_secs(2), started_rx.recv())
            .await
            .expect("provider should start before cancellation")
            .expect("provider start channel should remain open");

        let notification = state
            .handle_json_rpc(&serde_json::json!({
                "jsonrpc": "2.0",
                "method": "notifications/cancelled",
                "params": {"requestId": 99}
            }))
            .await;
        assert!(notification.is_none());

        let response = tokio::time::timeout(Duration::from_secs(2), handle)
            .await
            .expect("cancelled request should complete promptly")
            .expect("request task should not panic");
        assert_eq!(response["id"], 99);
        assert_eq!(response["error"]["code"], -32800);
        assert_eq!(response["error"]["message"], "request cancelled");
        wait_until_inactive_request(&state, "n:99").await;
    }

    async fn assert_inferred_inflight_contains(state: &ServerState, key: &InferredEdgeCacheKey) {
        for _ in 0..50 {
            if state.inferred_inflight.lock().await.contains_key(key) {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("inferred-dispatch leader never registered in-flight key");
    }

    async fn wait_until_inferred_inflight_removed(
        state: &ServerState,
        key: &InferredEdgeCacheKey,
    ) -> bool {
        for _ in 0..50 {
            if !state.inferred_inflight.lock().await.contains_key(key) {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        false
    }

    async fn wait_until_active_request(state: &ServerState, request_id: &str) {
        for _ in 0..50 {
            if state.active_requests.lock().await.contains(request_id) {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("request {request_id} never became active");
    }

    async fn wait_until_inactive_request(state: &ServerState, request_id: &str) {
        for _ in 0..50 {
            if !state.active_requests.lock().await.contains(request_id)
                && !state.cancelled_requests.lock().await.contains(request_id)
            {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("request {request_id} stayed active after cancellation");
    }

    fn inferred_test_key() -> InferredEdgeCacheKey {
        InferredEdgeCacheKey {
            caller_entity_id: "python:function:demo.dynamic".to_owned(),
            caller_content_hash: "hash-caller".to_owned(),
            model_id: "test-model".to_owned(),
            prompt_version: "test-prompt".to_owned(),
        }
    }

    fn inferred_test_read(key: InferredEdgeCacheKey) -> InferredRead {
        InferredRead {
            caller: entity_row(&key.caller_entity_id, "dynamic", Some("hash-caller")),
            sites: vec![UnresolvedCallSiteRow {
                caller_entity_id: key.caller_entity_id.clone(),
                caller_content_hash: key.caller_content_hash.clone(),
                site_key: "site-1".to_owned(),
                site_ordinal: 0,
                source_file_id: Some("python:module:demo".to_owned()),
                source_byte_start: 0,
                source_byte_end: 8,
                callee_expr: "target()".to_owned(),
            }],
            candidates: vec![entity_row(
                "python:function:demo.target",
                "target",
                Some("hash-target"),
            )],
            key,
            cached: None,
        }
    }

    fn entity_row(id: &str, name: &str, content_hash: Option<&str>) -> EntityRow {
        EntityRow {
            id: id.to_owned(),
            plugin_id: "python".to_owned(),
            kind: "function".to_owned(),
            name: name.to_owned(),
            short_name: name.to_owned(),
            parent_id: Some("python:module:demo".to_owned()),
            source_file_id: Some("python:module:demo".to_owned()),
            source_file_path: None,
            source_byte_start: Some(0),
            source_byte_end: Some(8),
            source_line_start: Some(1),
            source_line_end: Some(1),
            properties_json: "{}".to_owned(),
            content_hash: content_hash.map(str::to_owned),
            summary_json: None,
        }
    }

    #[test]
    fn blocked_entity_stub_preserves_navigable_identity() {
        // clarion-719e7320f5 (A3): the briefing-block projection keeps the
        // navigable identity (id/kind/name/short_name/path/lines/hash) and rides
        // the `briefing_blocked` flag; only the cross-tool `sei` stays null.
        let mut entity = entity_row("python:function:app.login", "app.login", Some("abc123"));
        entity.short_name = "login".to_owned();
        entity.source_file_path = Some("app.py".to_owned());
        entity.source_line_start = Some(10);
        entity.source_line_end = Some(20);

        let projection = super::blocked_entity_stub(&entity, "secret_present");
        assert_eq!(projection["id"], "python:function:app.login");
        assert_eq!(projection["kind"], "function");
        assert_eq!(projection["name"], "app.login");
        assert_eq!(projection["short_name"], "login");
        assert_eq!(projection["source_file_path"], "app.py");
        assert_eq!(projection["source_line_start"], 10);
        assert_eq!(projection["source_line_end"], 20);
        assert_eq!(projection["content_hash"], "abc123");
        assert_eq!(projection["briefing_blocked"], "secret_present");
        assert!(
            projection["sei"].is_null(),
            "SEI must stay null: {projection}"
        );
        // The secret-bearing content is never part of this projection.
        for content in ["summary", "source", "docstring"] {
            assert!(
                projection.get(content).is_none(),
                "{content} leaked: {projection}"
            );
        }
    }

    #[test]
    fn blocked_entity_stub_re_withholds_high_entropy_name_only() {
        // The guard: a high-entropy name/id is itself secret-like, so that one
        // field is re-withheld while the rest of the identity still rides along.
        let secret = "fn_aGVsbG8gd29ybGQgc2VjcmV0IGtleSBhYmMxMjP8x9z";
        let id = format!("python:function:{secret}");
        let mut entity = entity_row(&id, secret, Some("abc123"));
        entity.short_name = secret.to_owned();
        entity.source_file_path = Some("g.py".to_owned());

        let projection = super::blocked_entity_stub(&entity, "secret_present");
        assert!(
            projection["id"].is_null(),
            "high-entropy id must be withheld"
        );
        assert!(projection["name"].is_null());
        assert!(projection["short_name"].is_null());
        // Non-secret structural identity still rides along.
        assert_eq!(projection["kind"], "function");
        assert_eq!(projection["source_file_path"], "g.py");
        assert_eq!(projection["content_hash"], "abc123");
        assert_eq!(projection["briefing_blocked"], "secret_present");
    }

    #[test]
    fn name_value_is_secretlike_flags_secrets_not_ordinary_identifiers() {
        // Ordinary identifiers (even long dotted qualnames) stay below the
        // threshold; a base64/hex secret blob trips the guard.
        assert!(!super::name_value_is_secretlike("LibraryService"));
        assert!(!super::name_value_is_secretlike("app.auth.login"));
        assert!(!super::name_value_is_secretlike("python:function:demo.mid"));
        // Short-but-varied names are not flagged (length floor).
        assert!(!super::name_value_is_secretlike("aB3xZ9"));
        // Long *descriptive* identifiers stay navigable: verbose snake_case,
        // Rust `::` paths, and deep dotted qualnames all measure ~4.0–4.2 and
        // must survive the guard (the A3 over-redaction regression).
        assert!(!super::name_value_is_secretlike(
            "get_user_by_id_and_organization_with_permissions_check_v2"
        ));
        assert!(!super::name_value_is_secretlike(
            "rust:function:crate::module::submodule::function_with_a_long_name"
        ));
        assert!(!super::name_value_is_secretlike(
            "python:class:my.very.long.module.path.deeply.nested.HandlerClass"
        ));
        // A real base64 secret blob still trips the guard...
        assert!(super::name_value_is_secretlike(
            "fn_aGVsbG8gd29ybGQgc2VjcmV0IGtleSBhYmMxMjP8x9z"
        ));
        // ...and so does a blob embedded as ONE segment of an otherwise valid
        // qualified locator: per-segment entropy is not diluted by the prefix
        // (the qualified-name hole, clarion review P1).
        assert!(super::name_value_is_secretlike(
            "python:function:pkg.aGVsbG8gd29ybGQgc2VjcmV0IGtleSBhYmMxMjP8x9z"
        ));
    }

    #[test]
    fn source_for_entity_blocks_briefing_blocked_file_without_leaking_bytes() {
        // An entity whose source file the pre-ingest scanner marked
        // briefing_blocked must never have its bytes returned by
        // source_for_entity — that path would otherwise bypass the
        // secret-redaction policy the scanner enforces.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("secret.py");
        std::fs::write(&path, "API_KEY = 'super-secret-value'\n").unwrap();
        let mut entity = entity_row("python:function:demo.secret", "secret", None);
        entity.source_file_path = Some(path.to_string_lossy().into_owned());
        entity.source_line_start = Some(1);
        entity.source_line_end = Some(1);
        entity.properties_json =
            r#"{"briefing_blocked":"secret detected by pre-ingest scanner"}"#.to_owned();

        // A migrated connection satisfies the SEI read-time join
        // (`entity_json`); with no bindings seeded, `sei` degrades to null,
        // which is irrelevant to this briefing-block assertion. WAL (ADR-011)
        // requires a file-backed DB, so use the tempdir rather than :memory:.
        let db_path = dir.path().join("loomweave.db");
        let mut conn = Connection::open(&db_path).expect("open sqlite");
        pragma::apply_write_pragmas(&conn).expect("write pragmas");
        schema::apply_migrations(&mut conn).expect("apply migrations");

        let out = super::source_for_entity_json(&conn, &entity, 10);

        assert_eq!(out["source_status"], "briefing_blocked");
        assert_eq!(
            out["briefing_blocked"],
            "secret detected by pre-ingest scanner"
        );
        assert!(
            out.get("lines").is_none(),
            "must not return source lines: {out}"
        );
        assert!(
            !out.to_string().contains("super-secret-value"),
            "leaked briefing-blocked bytes: {out}"
        );
    }

    #[test]
    fn source_for_entity_blocks_when_only_the_file_anchor_is_briefing_blocked() {
        // The child entity itself carries no briefing_blocked flag, but its
        // enclosing source file does. source_for_entity must still refuse to
        // read or return bytes: resolving the file anchor (here via
        // `source_file_id`) and honouring its briefing_blocked flag is what
        // stops an agent from reaching secret-bearing bytes through an
        // individually-unmarked child entity.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("secret.py");
        std::fs::write(&path, "API_KEY = 'super-secret-value'\n").unwrap();

        let db_path = dir.path().join("loomweave.db");
        let mut conn = Connection::open(&db_path).expect("open sqlite");
        pragma::apply_write_pragmas(&conn).expect("write pragmas");
        schema::apply_migrations(&mut conn).expect("apply migrations");

        // Seed the file anchor the pre-ingest scanner marked briefing_blocked.
        conn.execute(
            "INSERT INTO entities (
                id, plugin_id, kind, name, short_name, source_file_path, properties,
                content_hash, created_at, updated_at
             ) VALUES (
                ?1, 'core', 'file', 'secret.py', 'secret.py', ?2, ?3, ?4,
                strftime('%Y-%m-%dT%H:%M:%fZ', 'now'),
                strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
             )",
            rusqlite::params![
                "core:file:secret.py",
                path.to_string_lossy(),
                r#"{"briefing_blocked":"secret detected by pre-ingest scanner"}"#,
                "hash-anchor",
            ],
        )
        .expect("seed briefing-blocked file anchor");

        // The child entity is clean at the entity level (properties "{}"); the
        // block must come from the resolved file anchor, not the entity itself.
        let mut entity = entity_row("python:function:demo.secret", "secret", None);
        entity.source_file_path = Some(path.to_string_lossy().into_owned());
        entity.source_file_id = Some("core:file:secret.py".to_owned());
        entity.source_line_start = Some(1);
        entity.source_line_end = Some(1);

        let out = super::source_for_entity_json(&conn, &entity, 10);

        assert_eq!(out["source_status"], "briefing_blocked");
        assert_eq!(
            out["briefing_blocked"],
            "secret detected by pre-ingest scanner"
        );
        assert!(out.get("lines").is_none(), "must not return lines: {out}");
        assert!(
            !out.to_string().contains("super-secret-value"),
            "leaked briefing-blocked bytes via file anchor: {out}"
        );
    }

    struct BlockingProvider {
        release: tokio::sync::Mutex<tokio::sync::mpsc::Receiver<()>>,
        started: Option<tokio::sync::mpsc::Sender<()>>,
    }

    #[async_trait::async_trait]
    impl LlmProvider for BlockingProvider {
        fn name(&self) -> &'static str {
            "blocking"
        }

        async fn invoke(&self, _request: LlmRequest) -> Result<LlmResponse, LlmProviderError> {
            if let Some(started) = &self.started {
                let _ = started.send(()).await;
            }
            let mut rx = self.release.lock().await;
            let _ = rx.recv().await;
            Ok(LlmResponse {
                model_id: "test-model".to_owned(),
                output_json: r#"{"edges":[]}"#.to_owned(),
                input_tokens: 1,
                cached_input_tokens: 0,
                output_tokens: 1,
                total_tokens: 2,
                cost_usd: 0.0,
            })
        }

        fn estimate_tokens(&self, _request: &LlmRequest) -> u64 {
            1
        }

        fn tier_to_model(&self, _tier: &str) -> Option<&str> {
            Some("test-model")
        }

        fn caching_model(&self) -> CachingModel {
            CachingModel::OpenAiChatCompletions
        }
    }
}
