//! The `coder.*` JSON-RPC surface — session registry, orchestration, fanout.
//!
//! Transport-thin: `handle_coder_*` functions parse params and delegate to
//! orchestration functions that are generation-injectable (the same seam as
//! the loops), so the full start→confirm→run→approve flow is testable with a
//! scripted model and a temp git repo.
//!
//! ## Event fanout
//!
//! Each session owns an [`EventSink`] whose emitter feeds an unbounded
//! channel; one drain task per session appends to the replay buffer and
//! forwards `coder.event` notification frames to every subscribed WS channel.
//! `coder.subscribe` replays from a `seq` cursor while holding the buffer
//! lock, then registers — same no-gap/no-dup discipline as `runs.subscribe`.
//! Lock order: `events` buffer → `coder_subscribers`; never the reverse.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use serde::Deserialize;
use serde_json::{json, Value};

use crate::handler::JsonRpcMessage;
use crate::parslee_tools::ParsleeToolExecutor;
use crate::session::{ClientSession, ServerState, WsChannel};

use super::config::CoderConfig;
use super::contract::{derive_contract, OutcomeContract};
use super::external_loop::{run_external_loop, ExternalLoopConfig};
use super::merge::{publish_branch, stage_and_diff};
use super::native_loop::{run_native_loop, AskUser, LoopOutcome, NativeLoopConfig, TurnGenerator};
use super::router::{detect_ready_agents, resolve_engine, EngineChoice};
use super::session::{
    default_state_dir, CancelFlag, CoderEvent, CoderEventKind, CoderSession, CoderState,
    EventEmitter, EventSink, UserInputGate,
};
use super::shell_tool::WorktreeExecutor;
use super::skill_memory::RepairMemory;

/// One live session in the daemon's registry.
pub struct CoderSessionEntry {
    pub session: Arc<tokio::sync::Mutex<CoderSession>>,
    /// Replay buffer for `coder.subscribe { from_seq }` after reconnects.
    pub events: Arc<tokio::sync::Mutex<Vec<CoderEvent>>>,
    pub cancel: CancelFlag,
    pub sink: Arc<EventSink>,
    /// The model seam the loops run on (production: the shared
    /// `InferenceEngine`; tests: a script).
    pub generator: Arc<dyn TurnGenerator>,
    /// Durable repair learning for the native loop. Cloned from the embedder's
    /// `shared_memgine`; a no-op store when the daemon runs standalone.
    pub memory: RepairMemory,
    /// The daemon's MCP URL (e.g. `"http://127.0.0.1:9102/mcp"`), captured at
    /// session start from [`ServerState::mcp_url`]. Threaded into the external
    /// and foreman delegation engines so the CLI's CAR-namespace tool calls
    /// (`memory_*`, `verify`, `skill_*`) route back through the daemon's policy
    /// + memgine — gated and audited. `None` when the daemon has no MCP
    /// listener (`--mcp-bind disabled`); delegation degrades to ungoverned
    /// CAR-namespace calls (the CLI's own built-in tools are ungoverned either
    /// way — the residual upstream stage-4b limitation).
    pub mcp_endpoint: Option<String>,
    /// Mid-session user-input rendezvous: the native loop parks a oneshot here
    /// when it asks a question (via the `ask_user` tool); `coder.respond`
    /// fulfills it. Cancellation clears it so a waiting question unblocks.
    pub user_input: Arc<UserInputGate>,
    /// The running loop task, present from confirm until terminal.
    pub task: std::sync::Mutex<Option<tokio::task::JoinHandle<()>>>,
}

/// Where session snapshots, journals, and worktrees live.
/// `CAR_CODER_STATE_DIR` overrides for tests and embedders.
pub fn coder_state_dir() -> Result<PathBuf, String> {
    if let Some(dir) = std::env::var_os("CAR_CODER_STATE_DIR") {
        return Ok(PathBuf::from(dir));
    }
    default_state_dir()
}

fn now_event_frame(event: &CoderEvent) -> Option<String> {
    serde_json::to_string(&json!({
        "jsonrpc": "2.0",
        "method": "coder.event",
        "params": event,
    }))
    .ok()
}

async fn send_frame(channel: &WsChannel, frame: &str) {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;
    let _ = channel
        .write
        .lock()
        .await
        .send(Message::Text(frame.to_string().into()))
        .await;
}

/// Cheap repo orientation for the contract-derivation prompt: top-level
/// listing plus recognizable build files.
fn summarize_repo(root: &Path) -> String {
    let mut names: Vec<String> = std::fs::read_dir(root)
        .map(|entries| {
            entries
                .flatten()
                .filter_map(|e| e.file_name().into_string().ok())
                .filter(|n| n != ".git")
                .collect()
        })
        .unwrap_or_default();
    names.sort();
    names.truncate(40);
    let build_hints: Vec<&str> = [
        ("Cargo.toml", "Rust (cargo)"),
        ("package.json", "Node (npm)"),
        ("pyproject.toml", "Python (pyproject)"),
        ("go.mod", "Go"),
        ("Makefile", "make"),
        ("Package.swift", "Swift (SwiftPM)"),
    ]
    .iter()
    .filter(|(f, _)| root.join(f).exists())
    .map(|(_, hint)| *hint)
    .collect();
    format!(
        "Top-level entries: {}\nBuild systems detected: {}",
        names.join(", "),
        if build_hints.is_empty() {
            "none recognized".to_string()
        } else {
            build_hints.join(", ")
        }
    )
}

fn is_git_repo(path: &Path) -> bool {
    std::process::Command::new("git")
        .arg("-C")
        .arg(path)
        .args(["rev-parse", "--is-inside-work-tree"])
        .output()
        .map(|o| o.status.success())
        .unwrap_or(false)
}

/// Register the per-session drain task: buffer every event and forward it to
/// current subscribers. Ends when the sink (and its emitter) drops.
fn spawn_event_drain(
    state: Arc<ServerState>,
    session_id: String,
    events: Arc<tokio::sync::Mutex<Vec<CoderEvent>>>,
) -> EventEmitter {
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<CoderEvent>();
    tokio::spawn(async move {
        while let Some(event) = rx.recv().await {
            let frame = now_event_frame(&event);
            // Hold the buffer lock across the sends: subscribe replays and
            // registers under this same lock, so a subscriber sees every
            // event exactly once (no gap between replay and live).
            let mut buffer = events.lock().await;
            buffer.push(event);
            if let Some(frame) = &frame {
                let subscribers: Vec<Arc<WsChannel>> = state
                    .coder_subscribers
                    .lock()
                    .await
                    .iter()
                    .filter(|((sid, _), _)| *sid == session_id)
                    .map(|(_, ch)| ch.clone())
                    .collect();
                for channel in subscribers {
                    send_frame(&channel, frame).await;
                }
            }
            drop(buffer);
        }
    });
    Arc::new(move |event| {
        let _ = tx.send(event);
    })
}

/// How long a model's `ask_user` request waits for the human before the loop
/// gives up and feeds a timeout error back to the model. Bounded so a wedged
/// session can never hang forever waiting on input that isn't coming.
const ASK_USER_TIMEOUT_SECS: u64 = 600;
/// Cancel-flag poll granularity while parked on a user answer.
const ASK_USER_CANCEL_POLL_MS: u64 = 200;

/// The native loop's [`AskUser`] handler: emits `UserInputRequested`, parks a
/// oneshot on the session's [`UserInputGate`], and awaits the reply while
/// honoring the cancel flag and a hard timeout. `coder.respond` fulfills the
/// oneshot from another task.
struct GateAsker {
    sink: Arc<EventSink>,
    gate: Arc<UserInputGate>,
    cancel: CancelFlag,
}

#[async_trait::async_trait]
impl AskUser for GateAsker {
    async fn ask(&self, prompt: &str) -> Result<String, String> {
        let mut rx = self.gate.park();
        self.sink.emit(CoderEventKind::UserInputRequested {
            prompt: prompt.to_string(),
        });
        let deadline =
            tokio::time::Instant::now() + std::time::Duration::from_secs(ASK_USER_TIMEOUT_SECS);
        let poll = std::time::Duration::from_millis(ASK_USER_CANCEL_POLL_MS);
        loop {
            if self.cancel.load(std::sync::atomic::Ordering::SeqCst) {
                // Cancellation: drop the parked sender and unblock the model.
                self.gate.clear();
                return Err("cancelled while awaiting user input".to_string());
            }
            tokio::select! {
                res = &mut rx => {
                    return match res {
                        Ok(answer) => Ok(answer),
                        // Sender dropped (cleared by cancel/teardown) without a
                        // value: treat as no answer rather than hanging.
                        Err(_) => Err("user-input request was cleared before an answer arrived".to_string()),
                    };
                }
                _ = tokio::time::sleep(poll) => {
                    if tokio::time::Instant::now() >= deadline {
                        self.gate.clear();
                        return Err(format!(
                            "no user response within {ASK_USER_TIMEOUT_SECS}s; proceeding without it"
                        ));
                    }
                }
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Orchestration (generation-injectable, transport-free)
// ---------------------------------------------------------------------------

pub struct StartArgs {
    pub repo: PathBuf,
    pub intent: String,
    pub engine: EngineChoice,
    /// `None` falls back to the operator config's `default_max_iterations`
    /// (`~/.car/coder.toml`), resolved inside `start_session` against the
    /// config it already loads — so the file is read once per start, and the
    /// preference / keep-on-failure / iteration defaults can't drift.
    pub max_iterations: Option<u32>,
    pub state_dir: PathBuf,
    /// When set, this session works on a CAR-managed project (`repo` is the
    /// project's repo path). Drives commit-to-main delivery and, for `Agent`
    /// projects, the scenario contract + agent registration. `None` =
    /// raw-repo session.
    pub project: Option<(String, super::project::ProjectKind)>,
}

/// Provision worktree + derive contract + register the session. Returns the
/// start response value.
pub async fn start_session(
    state: &Arc<ServerState>,
    args: StartArgs,
    generator: Arc<dyn TurnGenerator>,
) -> Result<Value, String> {
    let repo = args
        .repo
        .canonicalize()
        .map_err(|e| format!("repo path {}: {e}", args.repo.display()))?;
    if !is_git_repo(&repo) {
        return Err(format!(
            "{} is not inside a git repository — the coder works in git worktrees",
            repo.display()
        ));
    }

    // Operator config (`~/.car/coder.toml`): delegation preference + keep-on-
    // failure. Tolerant — a missing file yields documented defaults.
    let config = CoderConfig::load();

    // Resolve the engine up front so the user confirms the contract knowing
    // who will execute it. The configured `engine_preference` decides which
    // ready external CLI wins under `auto`/`external`/`foreman`.
    let resolved = match &args.engine {
        EngineChoice::Native => super::router::ResolvedEngine {
            engine: EngineChoice::Native,
            reason: "explicitly requested".into(),
        },
        other => {
            let detected = detect_ready_agents().await;
            resolve_engine(other, &args.intent, &detected, &config.preference_refs())?
        }
    };

    // Durable repair learning rides on the embedder's shared memgine when
    // present; standalone daemons get a no-op store (never a hard dependency).
    let memory = RepairMemory::new(state.shared_memgine.clone());

    // The daemon's MCP URL, when its listener is bound. Threaded into the
    // external/foreman engines so the CLI's CAR-namespace tool calls route
    // back through the daemon's policy + memgine. `None` degrades cleanly.
    let mcp_endpoint = state.mcp_url.get().cloned();

    // Omitted max_iterations falls back to the same config instance, so the
    // file is parsed once per start (no second load in the RPC handler).
    let max_iterations = args.max_iterations.unwrap_or(config.default_max_iterations);
    let mut session = CoderSession::new(
        &repo,
        &args.intent,
        resolved.engine.clone(),
        max_iterations,
        Some(args.state_dir.clone()),
    );
    if let Some((slug, kind)) = &args.project {
        session = session.with_project(slug.clone(), *kind);
    }
    session.keep_workspace_on_failure = config.keep_workspace_on_failure;
    let worktree = session.provision_workspace()?;
    let session_id = session.id.clone();

    let events = Arc::new(tokio::sync::Mutex::new(Vec::new()));
    let emitter = spawn_event_drain(state.clone(), session_id.clone(), events.clone());
    let sink = Arc::new(EventSink::new(
        &session_id,
        Some(emitter),
        Some(args.state_dir.join(format!("{session_id}.events.jsonl"))),
    ));

    sink.emit(CoderEventKind::EngineSelected {
        engine: resolved.engine.label(),
        reason: resolved.reason,
    });

    // Agent projects don't derive a shell contract — their "definition of
    // done" is "the built agent passes its own scenarios", which the agent
    // build loop verifies in-daemon (run_session_loop). Synthesize a contract
    // for the confirmation UX; the real verification is the scenario run.
    let is_agent_project = matches!(args.project, Some((_, super::project::ProjectKind::Agent)));
    let contract = if is_agent_project {
        Ok(OutcomeContract {
            description: format!(
                "Build an in-daemon agent for: {}. It must pass its own acceptance scenarios.",
                args.intent.trim()
            ),
            checks: vec![super::contract::ContractCheck {
                name: "agent_scenarios_pass".into(),
                command: "(in-daemon scenario evaluation)".into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 600,
            }],
        })
    } else {
        derive_app_contract(&generator, &args.intent, &worktree).await
    };

    let contract = match contract {
        Ok(c) => c,
        Err(e) => {
            session.error = Some(e.clone());
            let _ = session.transition(CoderState::Failed, &sink);
            let entry = Arc::new(CoderSessionEntry {
                session: Arc::new(tokio::sync::Mutex::new(session)),
                events,
                cancel: Arc::new(std::sync::atomic::AtomicBool::new(false)),
                sink,
                generator,
                memory: memory.clone(),
                mcp_endpoint: mcp_endpoint.clone(),
                user_input: Arc::new(UserInputGate::new()),
                task: std::sync::Mutex::new(None),
            });
            state
                .coder_sessions
                .lock()
                .await
                .insert(session_id.clone(), entry);
            return Err(format!("contract derivation failed: {e}"));
        }
    };

    session.contract = Some(contract.clone());
    session.transition(CoderState::ContractProposed, &sink)?;
    sink.emit(CoderEventKind::ContractProposed {
        contract: contract.clone(),
    });

    let response = json!({
        "session_id": session_id,
        "state": session.state.as_str(),
        "engine": session.engine.label(),
        "worktree": session.workspace_path,
        "contract": contract,
    });

    let entry = Arc::new(CoderSessionEntry {
        session: Arc::new(tokio::sync::Mutex::new(session)),
        events,
        cancel: Arc::new(std::sync::atomic::AtomicBool::new(false)),
        sink,
        generator,
        memory,
        mcp_endpoint,
        user_input: Arc::new(UserInputGate::new()),
        task: std::sync::Mutex::new(None),
    });
    state.coder_sessions.lock().await.insert(session_id, entry);
    Ok(response)
}

/// Derive an App project / raw-repo session's shell contract from the intent
/// (the model path). Agent projects synthesize their contract instead.
async fn derive_app_contract(
    generator: &Arc<dyn TurnGenerator>,
    intent: &str,
    worktree: &Path,
) -> Result<OutcomeContract, String> {
    let summary = summarize_repo(worktree);
    let gen_for_derive = generator.clone();
    derive_contract(
        move |prompt| {
            let generator = gen_for_derive.clone();
            async move {
                generator
                    .generate(car_inference::GenerateRequest {
                        prompt,
                        params: car_inference::GenerateParams {
                            temperature: 0.0,
                            // Structured JSON extraction, not open reasoning:
                            // force thinking OFF (hybrid models otherwise burn
                            // the budget in an unclosed `<think>` and return
                            // empty text) and give room for the object.
                            max_tokens: 2048,
                            thinking: car_inference::tasks::generate::ThinkingMode::Off,
                            ..Default::default()
                        },
                        // `require: [Code]` is a HARD filter so a tiny non-code
                        // local model is excluded when a capable one exists,
                        // instead of winning on cost and emitting garbage.
                        intent: Some(car_inference::IntentHint {
                            task: Some(car_inference::TaskHint::Code),
                            require: vec![car_inference::ModelCapability::Code],
                            // Deriving a good contract is quality-critical and
                            // happens once per session — prefer the most capable
                            // code model over the cheapest.
                            prefer_quality: true,
                            ..Default::default()
                        }),
                        ..Default::default()
                    })
                    .await
                    .map(|r| r.text)
            }
        },
        intent,
        &summary,
        3,
    )
    .await
}

/// Confirm (optionally replacing) the contract and spawn the work loop.
pub async fn confirm_session(
    state: &Arc<ServerState>,
    session_id: &str,
    contract_override: Option<OutcomeContract>,
) -> Result<Value, String> {
    let entry = get_entry(state, session_id).await?;
    {
        let mut session = entry.session.lock().await;
        if session.state != CoderState::ContractProposed {
            return Err(format!(
                "session is {}, expected contract_proposed",
                session.state.as_str()
            ));
        }
        if let Some(contract) = contract_override {
            let issues = contract.validate();
            if !issues.is_empty() {
                return Err(format!("edited contract is invalid: {}", issues.join("; ")));
            }
            entry.sink.emit(CoderEventKind::ContractProposed {
                contract: contract.clone(),
            });
            session.contract = Some(contract);
        }
        session.transition(CoderState::ContractConfirmed, &entry.sink)?;
        session.transition(CoderState::Running, &entry.sink)?;
    }

    let task_entry = entry.clone();
    let handle = tokio::spawn(async move {
        run_session_loop(task_entry).await;
    });
    *entry.task.lock().expect("task slot poisoned") = Some(handle);

    Ok(json!({ "state": "running" }))
}

/// The spawned work loop: engine → (fallback) → verify → diff → gate.
async fn run_session_loop(entry: Arc<CoderSessionEntry>) {
    let (engine, intent, contract, worktree, max_iterations, project_kind) = {
        let session = entry.session.lock().await;
        let Some(contract) = session.contract.clone() else {
            return; // unreachable: confirm requires a contract
        };
        let Some(worktree) = session.workspace_path.clone() else {
            return;
        };
        (
            session.engine.clone(),
            session.intent.clone(),
            contract,
            worktree,
            session.max_iterations,
            session.project_kind,
        )
    };

    // Parslee platform tools are available as a delegate; the coder→agent loop
    // advertises them so generated agents can allowlist them, and scenario eval
    // can execute them.
    let executor = WorktreeExecutor::new(&worktree).with_delegate(
        Arc::new(ParsleeToolExecutor),
        ParsleeToolExecutor::tool_defs(),
    );

    // Agent projects don't use the engine/shell loop at all: the work is
    // "build a declarative agent that passes its own scenarios", run entirely
    // in-daemon. On success the spec is written to the worktree (so
    // commit_to_main captures it) and stashed for registration on approve.
    if matches!(project_kind, Some(super::project::ProjectKind::Agent)) {
        let outcome = run_agent_build(&entry, &intent, &worktree, &executor, max_iterations).await;
        finalize_outcome(&entry, &worktree, outcome).await;
        return;
    }

    let native_cfg = NativeLoopConfig {
        max_iterations,
        ..Default::default()
    };
    // The native loop's mid-session question handler. Only the native loop can
    // ask (the external/foreman CLIs own their own interaction model), so it is
    // threaded into every native call below.
    let asker = GateAsker {
        sink: entry.sink.clone(),
        gate: entry.user_input.clone(),
        cancel: entry.cancel.clone(),
    };

    let outcome: LoopOutcome = match &engine {
        EngineChoice::External(agent_id) if !agent_id.is_empty() => {
            run_external_with_native_fallback(
                &entry,
                agent_id,
                &intent,
                &contract,
                &executor,
                &native_cfg,
                &asker,
            )
            .await
        }
        EngineChoice::Foreman(agent_id) if !agent_id.is_empty() => {
            // Foreman-first ladder: verified parallel farm-out → (decline)
            // single-session external → (spawn failure) native. A red
            // contract AFTER foreman applied its verified union also falls
            // to native, which then repairs on top of foreman's work.
            match super::foreman_loop::run_foreman_loop(
                agent_id,
                &intent,
                &contract,
                &executor,
                &entry.sink,
                &entry.cancel,
                &entry.generator,
                entry.mcp_endpoint.as_deref(),
            )
            .await
            {
                Ok(outcome) if outcome.passed || outcome.error.is_some() => outcome,
                Ok(_red) => {
                    entry.sink.emit(CoderEventKind::EngineFallback {
                        from: format!("foreman:{agent_id}"),
                        to: "native".into(),
                        reason: "contract not satisfied after foreman's verified union; \
                                 repairing natively on top of it"
                            .into(),
                    });
                    run_native_loop(
                        entry.generator.as_ref(),
                        &executor,
                        &intent,
                        &contract,
                        &entry.sink,
                        &entry.cancel,
                        &native_cfg,
                        &entry.memory,
                        Some(&asker),
                    )
                    .await
                }
                Err(fallback) => {
                    entry.sink.emit(CoderEventKind::EngineFallback {
                        from: format!("foreman:{agent_id}"),
                        to: format!("external:{agent_id}"),
                        reason: fallback.reason(),
                    });
                    run_external_with_native_fallback(
                        &entry,
                        agent_id,
                        &intent,
                        &contract,
                        &executor,
                        &native_cfg,
                        &asker,
                    )
                    .await
                }
            }
        }
        _ => {
            run_native_loop(
                entry.generator.as_ref(),
                &executor,
                &intent,
                &contract,
                &entry.sink,
                &entry.cancel,
                &native_cfg,
                &entry.memory,
                Some(&asker),
            )
            .await
        }
    };

    finalize_outcome(&entry, &worktree, outcome).await;
}

/// Fold a loop outcome into the session: green → diff + `NeedsApproval`;
/// red → `Failed` (or `Abandoned` on cancel). Shared by the engine paths and
/// the agent-build path.
async fn finalize_outcome(entry: &Arc<CoderSessionEntry>, worktree: &Path, outcome: LoopOutcome) {
    let mut session = entry.session.lock().await;
    session.iterations = outcome.iterations;
    session.last_check_results = outcome.last_results.clone();

    if outcome.passed {
        match stage_and_diff(worktree) {
            Ok((stat, patch_truncated)) => {
                entry.sink.emit(CoderEventKind::DiffReady {
                    stat,
                    patch_truncated,
                });
            }
            Err(e) => {
                entry.sink.emit(CoderEventKind::Error {
                    message: format!("diff generation failed: {e}"),
                });
            }
        }
        let _ = session.transition(CoderState::NeedsApproval, &entry.sink);
    } else {
        session.error = Some(outcome.error.unwrap_or_else(|| {
            format!(
                "contract not satisfied after {} iteration(s)",
                outcome.iterations
            )
        }));
        let to = if session.error.as_deref() == Some("cancelled") {
            CoderState::Abandoned
        } else {
            CoderState::Failed
        };
        if to == CoderState::Failed && session.keep_workspace_on_failure {
            if let Some(path) = &session.workspace_path {
                entry.sink.emit(CoderEventKind::Error {
                    message: format!(
                        "session failed; worktree retained for postmortem at {} \
                         (keep_workspace_on_failure)",
                        path.display()
                    ),
                });
            }
        }
        let _ = session.transition(to, &entry.sink);
    }
}

/// The coder→agent build loop for an Agent project: generate a declarative
/// agent spec from the intent, drive its scenarios green in-daemon, write the
/// spec to the worktree (so commit_to_main captures it), and stash it on the
/// session for registration on approve.
async fn run_agent_build(
    entry: &Arc<CoderSessionEntry>,
    intent: &str,
    worktree: &Path,
    executor: &WorktreeExecutor,
    max_iterations: u32,
) -> LoopOutcome {
    use super::declarative::{build_agent, BuildAgentConfig};

    if entry.cancel.load(std::sync::atomic::Ordering::SeqCst) {
        return LoopOutcome {
            passed: false,
            iterations: 0,
            last_results: Vec::new(),
            error: Some("cancelled".into()),
        };
    }

    let agent_id = {
        let session = entry.session.lock().await;
        session
            .project
            .clone()
            .unwrap_or_else(|| session.short_id().to_string())
    };
    let mut available_tools: Vec<String> = WorktreeExecutor::tool_defs()
        .iter()
        .filter_map(|d| d.get("name").and_then(Value::as_str).map(String::from))
        .collect();
    // Offer the Parslee platform tools to generated agents (they're wired as a
    // delegate on the executor above, so allowlisting them makes them callable).
    available_tools.extend(ParsleeToolExecutor::tool_names());

    entry.sink.emit(CoderEventKind::PlanText {
        text: "Designing the agent and checking it against its scenarios…".into(),
    });

    let cfg = BuildAgentConfig {
        agent_id,
        available_tools,
        max_attempts: max_iterations.max(3),
    };
    let built = build_agent(intent, entry.generator.as_ref(), executor, &cfg).await;

    if !built.passed {
        return LoopOutcome {
            passed: false,
            iterations: built.attempts,
            last_results: Vec::new(),
            error: Some(if built.issues.is_empty() {
                "could not build an agent that passes its scenarios".into()
            } else {
                format!(
                    "agent did not pass its scenarios: {}",
                    built.issues.join("; ")
                )
            }),
        };
    }

    let spec = built.spec.expect("passed build has a spec");
    // Write the spec + scenarios into the worktree so the commit captures them.
    let agent_json = serde_json::to_string_pretty(&spec).unwrap_or_default();
    let scenarios_json = serde_json::to_string_pretty(&spec.scenarios).unwrap_or_default();
    if let Err(e) = std::fs::write(worktree.join("agent.json"), agent_json)
        .and_then(|_| std::fs::write(worktree.join("scenarios.json"), scenarios_json))
    {
        return LoopOutcome {
            passed: false,
            iterations: built.attempts,
            last_results: Vec::new(),
            error: Some(format!("failed to write the agent spec: {e}")),
        };
    }

    entry.sink.emit(CoderEventKind::PlanText {
        text: format!(
            "Built agent '{}' — {} scenario(s) pass. Tools: {}.",
            spec.name,
            spec.scenarios.len(),
            if spec.tools.is_empty() {
                "none".into()
            } else {
                spec.tools.join(", ")
            }
        ),
    });

    entry.session.lock().await.built_agent = Some(spec);

    LoopOutcome {
        passed: true,
        iterations: built.attempts,
        last_results: Vec::new(),
        error: None,
    }
}

/// One external-CLI session with native fallback on spawn/transport failure
/// (red checks and cancellation are not fallbacks — they end the attempt).
async fn run_external_with_native_fallback(
    entry: &Arc<CoderSessionEntry>,
    agent_id: &str,
    intent: &str,
    contract: &OutcomeContract,
    executor: &WorktreeExecutor,
    native_cfg: &NativeLoopConfig,
    asker: &GateAsker,
) -> LoopOutcome {
    let external = run_external_loop(
        agent_id,
        intent,
        contract,
        executor,
        &entry.sink,
        &entry.cancel,
        &ExternalLoopConfig::default(),
        entry.mcp_endpoint.as_deref(),
    )
    .await;
    let spawn_failed = external.error.as_deref().is_some_and(|e| e != "cancelled");
    if spawn_failed {
        entry.sink.emit(CoderEventKind::EngineFallback {
            from: format!("external:{agent_id}"),
            to: "native".into(),
            reason: external.error.clone().unwrap_or_default(),
        });
        run_native_loop(
            entry.generator.as_ref(),
            executor,
            intent,
            contract,
            &entry.sink,
            &entry.cancel,
            native_cfg,
            &entry.memory,
            Some(asker),
        )
        .await
    } else {
        external
    }
}

/// Approve (publish branch) or deny (abandon) a session awaiting merge.
pub async fn approve_merge_session(
    state: &Arc<ServerState>,
    session_id: &str,
    approve: bool,
) -> Result<Value, String> {
    let entry = get_entry(state, session_id).await?;
    let mut session = entry.session.lock().await;
    if session.state != CoderState::NeedsApproval {
        return Err(format!(
            "session is {}, expected needs_approval",
            session.state.as_str()
        ));
    }
    if !approve {
        session.transition(CoderState::Abandoned, &entry.sink)?;
        return Ok(json!({ "state": "abandoned" }));
    }
    let worktree = session
        .workspace_path
        .clone()
        .ok_or("session has no worktree")?;
    let contract = session.contract.clone().ok_or("session has no contract")?;

    // Managed projects commit straight to `main` (the project is fully
    // CAR-owned — no separate user working tree to protect); raw repos get a
    // `car/coder/<id>` branch. Both showed the diff before this gate.
    let branch = if session.project.is_some() {
        super::merge::commit_to_main(&session.repo, &worktree, &session.intent, &contract)?;
        "main".to_string()
    } else {
        publish_branch(
            &session.repo,
            &worktree,
            session.short_id(),
            &session.intent,
            &contract,
        )?
    };
    session.result_branch = Some(branch.clone());

    // Agent projects: register the built declarative agent so it shows in
    // agents.list and is runnable in-daemon. Registration failure is surfaced
    // but does not undo the commit (the spec is in the repo either way).
    let mut registered_agent: Option<String> = None;
    if let Some(spec) = session.built_agent.clone() {
        match state.declagents().and_then(|r| r.upsert(spec.clone())) {
            Ok(()) => {
                registered_agent = Some(spec.id.clone());
                entry.sink.emit(CoderEventKind::PlanText {
                    text: format!(
                        "Agent '{}' added to your agents and ready to run.",
                        spec.name
                    ),
                });
            }
            Err(e) => {
                entry.sink.emit(CoderEventKind::Error {
                    message: format!("agent built and saved, but registration failed: {e}"),
                });
            }
        }
    }

    entry.sink.emit(CoderEventKind::MergeCompleted {
        branch: branch.clone(),
    });
    session.transition(CoderState::Merged, &entry.sink)?;
    Ok(json!({ "state": "merged", "branch": branch, "agent_id": registered_agent }))
}

/// Cancel a session: flag the loop, abort its task, abandon the state.
pub async fn cancel_session(state: &Arc<ServerState>, session_id: &str) -> Result<Value, String> {
    let entry = get_entry(state, session_id).await?;
    entry
        .cancel
        .store(true, std::sync::atomic::Ordering::SeqCst);
    // Unblock any model question parked on the gate: dropping the sender closes
    // the waiter's receiver, so it returns immediately instead of waiting out
    // the timeout (the cancel flag is also set, so the loop exits next turn).
    entry.user_input.clear();
    if let Some(handle) = entry.task.lock().expect("task slot poisoned").take() {
        // The loop checks the flag between turns; abort cuts long-running
        // inference/shell awaits. kill_on_drop reaps any spawned shell.
        handle.abort();
    }
    let mut session = entry.session.lock().await;
    if !session.state.is_terminal() {
        session.transition(CoderState::Abandoned, &entry.sink)?;
    }
    Ok(json!({ "state": session.state.as_str() }))
}

async fn get_entry(
    state: &Arc<ServerState>,
    session_id: &str,
) -> Result<Arc<CoderSessionEntry>, String> {
    state
        .coder_sessions
        .lock()
        .await
        .get(session_id)
        .cloned()
        .ok_or_else(|| format!("no live coder session '{session_id}'"))
}

fn session_summary(session: &CoderSession, live: bool) -> Value {
    json!({
        "session_id": session.id,
        "state": session.state.as_str(),
        "intent": session.intent,
        "repo": session.repo,
        "engine": session.engine.label(),
        "iterations": session.iterations,
        "updated_at": session.updated_at,
        "live": live,
        "error": session.error,
    })
}

// ---------------------------------------------------------------------------
// JSON-RPC handlers (thin parsing wrappers)
// ---------------------------------------------------------------------------

#[derive(Deserialize)]
struct StartParams {
    /// A raw git repo path. Exactly one of `repo` / `project` must be set.
    #[serde(default)]
    repo: Option<PathBuf>,
    /// A CAR-managed project slug (resolved under `~/.car/projects/`). The
    /// non-dev path — no repo to pick.
    #[serde(default)]
    project: Option<String>,
    intent: String,
    #[serde(default)]
    engine: Option<String>,
    #[serde(default)]
    max_iterations: Option<u32>,
}

pub async fn handle_coder_start(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: StartParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let engine = EngineChoice::parse(params.engine.as_deref().unwrap_or("auto"))?;
    let generator: Arc<dyn TurnGenerator> = crate::handler::get_inference_engine(state).clone();

    // Exactly one of repo / project. A project resolves to its managed repo
    // path and tags the session so delivery commits to main + (for Agent
    // projects) registers the agent.
    let (repo, project) = match (params.repo, params.project) {
        (Some(_), Some(_)) => {
            return Err("provide exactly one of `repo` or `project`, not both".into())
        }
        (None, None) => {
            return Err(
                "provide one of `repo` (a git path) or `project` (a managed project)".into(),
            )
        }
        (Some(repo), None) => (repo, None),
        (None, Some(slug)) => {
            let proj = super::project::load_project(&slug)?;
            (proj.repo_path, Some((proj.slug, proj.kind)))
        }
    };

    // `max_iterations` is passed through as-is; `start_session` resolves the
    // None fallback from the config it loads, so coder.toml is read once.
    start_session(
        state,
        StartArgs {
            repo,
            intent: params.intent,
            engine,
            max_iterations: params.max_iterations,
            state_dir: coder_state_dir()?,
            project,
        },
        generator,
    )
    .await
}

#[derive(Deserialize)]
struct ProjectsCreateParams {
    name: String,
    #[serde(default)]
    kind: Option<String>,
}

pub async fn handle_coder_projects_create(
    req: &JsonRpcMessage,
    _state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: ProjectsCreateParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let kind = super::project::ProjectKind::parse(params.kind.as_deref().unwrap_or("app"))?;
    let project = super::project::resolve_or_create_project(&params.name, kind)?;
    serde_json::to_value(&project).map_err(|e| e.to_string())
}

pub async fn handle_coder_projects_list(_state: &Arc<ServerState>) -> Result<Value, String> {
    Ok(json!({ "projects": super::project::list_projects() }))
}

#[derive(Deserialize)]
struct ProjectsGetParams {
    slug: String,
}

pub async fn handle_coder_projects_get(
    req: &JsonRpcMessage,
    _state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: ProjectsGetParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let project = super::project::load_project(&params.slug)?;
    serde_json::to_value(&project).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct ConfirmParams {
    session_id: String,
    #[serde(default)]
    contract: Option<OutcomeContract>,
}

pub async fn handle_coder_confirm_contract(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: ConfirmParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    confirm_session(state, &params.session_id, params.contract).await
}

pub async fn handle_coder_list(state: &Arc<ServerState>) -> Result<Value, String> {
    let mut out: Vec<Value> = Vec::new();
    let mut live_ids = std::collections::HashSet::new();
    {
        let sessions = state.coder_sessions.lock().await;
        for entry in sessions.values() {
            let session = entry.session.lock().await;
            live_ids.insert(session.id.clone());
            out.push(session_summary(&session, true));
        }
    }
    // Persisted sessions from prior daemon lifetimes (orphans included).
    if let Ok(dir) = coder_state_dir() {
        for session in CoderSession::list(&dir) {
            if !live_ids.contains(&session.id) {
                out.push(session_summary(&session, false));
            }
        }
    }
    out.sort_by_key(|v| std::cmp::Reverse(v["updated_at"].as_u64().unwrap_or(0)));
    Ok(json!({ "sessions": out }))
}

#[derive(Deserialize)]
struct SessionIdParams {
    session_id: String,
}

pub async fn handle_coder_get(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: SessionIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    if let Ok(entry) = get_entry(state, &params.session_id).await {
        let session = entry.session.lock().await;
        let mut value = serde_json::to_value(&*session).map_err(|e| e.to_string())?;
        value["live"] = json!(true);
        value["next_seq"] = json!(entry.events.lock().await.len() as u64);
        return Ok(value);
    }
    // Fall back to the persisted snapshot (prior daemon lifetime).
    let dir = coder_state_dir()?;
    let session = CoderSession::load(&dir.join(format!("{}.json", params.session_id)))?;
    let mut value = serde_json::to_value(&session).map_err(|e| e.to_string())?;
    value["live"] = json!(false);
    Ok(value)
}

#[derive(Deserialize)]
struct SubscribeParams {
    session_id: String,
    #[serde(default)]
    from_seq: u64,
}

pub async fn handle_coder_subscribe(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: SubscribeParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let entry = get_entry(state, &params.session_id).await?;

    // Replay + register under the buffer lock (see module docs).
    let buffer = entry.events.lock().await;
    let mut replayed = 0u64;
    for event in buffer.iter().filter(|e| e.seq >= params.from_seq) {
        if let Some(frame) = now_event_frame(event) {
            send_frame(&session.channel, &frame).await;
            replayed += 1;
        }
    }
    state.coder_subscribers.lock().await.insert(
        (params.session_id.clone(), session.client_id.clone()),
        session.channel.clone(),
    );
    drop(buffer);

    let current_state = entry.session.lock().await.state.as_str().to_string();
    Ok(json!({ "state": current_state, "events_replayed": replayed }))
}

pub async fn handle_coder_unsubscribe(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<ClientSession>,
) -> Result<Value, String> {
    let params: SessionIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    state
        .coder_subscribers
        .lock()
        .await
        .remove(&(params.session_id, session.client_id.clone()));
    Ok(json!({ "ok": true }))
}

#[derive(Deserialize)]
struct RespondParams {
    session_id: String,
    /// The user's reply to the session's pending `UserInputRequested`.
    text: String,
}

/// Fulfill a session's pending mid-session user-input request (the native loop's
/// `ask_user` tool). Returns `{ok:true}` when a request was waiting and got the
/// answer; a clear error when nothing is pending or the waiter already gave up.
pub async fn handle_coder_respond(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: RespondParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let entry = get_entry(state, &params.session_id).await?;
    entry.user_input.fulfill(params.text)?;
    Ok(json!({ "ok": true }))
}

#[derive(Deserialize)]
struct ApproveParams {
    session_id: String,
    approve: bool,
}

pub async fn handle_coder_approve_merge(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: ApproveParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    approve_merge_session(state, &params.session_id, params.approve).await
}

pub async fn handle_coder_cancel(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: SessionIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    cancel_session(state, &params.session_id).await
}

/// Drop a disconnecting client's coder subscriptions (called from
/// `remove_session`).
pub async fn drop_subscriptions_for_client(state: &ServerState, client_id: &str) {
    state
        .coder_subscribers
        .lock()
        .await
        .retain(|(_, cid), _| cid != client_id);
}

// Keep HashMap import alive for the registry type alias used by ServerState.
pub type CoderSessionMap = HashMap<String, Arc<CoderSessionEntry>>;

// ---------------------------------------------------------------------------
// declagents.* — declarative (in-daemon) agents
// ---------------------------------------------------------------------------

/// Render a declarative spec as an `agents.list`-style row (tagged
/// `kind:"declarative"`, carrying `enabled` rather than process status).
fn declarative_row(spec: &car_registry::declarative::DeclarativeAgentSpec) -> Value {
    json!({
        "id": spec.id,
        "name": spec.name,
        "kind": "declarative",
        "enabled": spec.enabled,
        "tools": spec.tools,
        "scenarios": spec.scenarios.len(),
    })
}

/// Declarative agents as `agents.list` rows, for the unified host view.
/// Returns an empty list (never errors) so a missing registry never breaks
/// `agents.list`.
pub async fn declarative_agent_rows(state: &Arc<ServerState>) -> Vec<Value> {
    match state.declagents() {
        Ok(reg) => reg.list().iter().map(declarative_row).collect(),
        Err(_) => Vec::new(),
    }
}

pub async fn handle_declagents_list(state: &Arc<ServerState>) -> Result<Value, String> {
    let reg = state.declagents()?;
    Ok(json!({ "agents": reg.list().iter().map(declarative_row).collect::<Vec<_>>() }))
}

#[derive(Deserialize)]
struct DeclAgentIdParams {
    id: String,
}

pub async fn handle_declagents_get(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DeclAgentIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let reg = state.declagents()?;
    let spec = reg
        .get(&params.id)
        .ok_or_else(|| format!("no declarative agent '{}'", params.id))?;
    serde_json::to_value(&spec).map_err(|e| e.to_string())
}

pub async fn handle_declagents_remove(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DeclAgentIdParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let reg = state.declagents()?;
    Ok(json!({ "removed": reg.remove(&params.id)? }))
}

#[derive(Deserialize)]
struct DeclAgentEnableParams {
    id: String,
    enabled: bool,
}

pub async fn handle_declagents_set_enabled(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DeclAgentEnableParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let reg = state.declagents()?;
    reg.set_enabled(&params.id, params.enabled)?;
    Ok(json!({ "ok": true }))
}

#[derive(Deserialize)]
struct DeclAgentInvokeParams {
    id: String,
    input: String,
}

/// Run a declarative agent on `input`, in-daemon (no process). Shared by
/// `declagents.invoke` (caller names the agent) and `declagents.route`
/// (the runtime picks the agent by capability similarity).
async fn run_declarative(
    spec: &car_registry::declarative::DeclarativeAgentSpec,
    input: &str,
    state: &Arc<ServerState>,
) -> Result<super::declarative::AgentRunResult, String> {
    let generator: Arc<dyn TurnGenerator> = crate::handler::get_inference_engine(state).clone();
    // Ephemeral scratch workspace for any file tools the agent uses. Parslee
    // platform tools are available as a delegate (subject to the spec allowlist).
    let scratch = tempfile::tempdir().map_err(|e| format!("scratch dir: {e}"))?;
    let executor = WorktreeExecutor::new(scratch.path()).with_delegate(
        Arc::new(ParsleeToolExecutor),
        ParsleeToolExecutor::tool_defs(),
    );
    let runner =
        super::declarative::DeclarativeAgentRunner::new(spec, generator.as_ref(), &executor);
    Ok(runner.run(input).await)
}

fn run_result_json(result: &super::declarative::AgentRunResult) -> Value {
    json!({
        "output": result.output,
        "turns": result.turns,
        "tool_calls": result.tool_calls,
        "error": result.error,
    })
}

/// A run counts as a success for routing-prior purposes when it completed
/// without an error and produced non-empty output.
fn run_succeeded(result: &super::declarative::AgentRunResult) -> bool {
    result.error.is_none() && !result.output.trim().is_empty()
}

/// Whether a run's outcome should teach the routing store at all. A run that
/// errored without taking a single turn never reached the model — that's infra
/// noise (admission starvation, model load failure), not the agent's
/// competence. Recording it would let bad luck depress a capable agent's prior
/// and starve it from future routing, so such runs are left unlearned.
fn run_is_recordable(result: &super::declarative::AgentRunResult) -> bool {
    !(result.turns == 0 && result.error.is_some())
}

/// Feed a run's outcome into the routing learning store. Best-effort: a store
/// failure (or unresolved home dir) must never fail the routed/invoked call —
/// routing just stays cold.
fn record_routing_outcome(
    state: &Arc<ServerState>,
    agent_id: &str,
    result: &super::declarative::AgentRunResult,
) {
    if !run_is_recordable(result) {
        return;
    }
    if let Ok(store) = state.routing() {
        let _ = store.record_outcome(agent_id, run_succeeded(result));
    }
}

/// Reinforce or weaken the directed forward edge `from → to` by a run's
/// outcome. Best-effort, same as [`record_routing_outcome`].
fn record_routing_edge(state: &Arc<ServerState>, from: &str, to: &str, ok: bool) {
    if let Ok(store) = state.routing() {
        let _ = store.record_edge(from, to, ok);
    }
}

/// Fold the need's embedding into the agent's learned capability centroid after
/// a successful run. Best-effort.
fn record_routing_capability(state: &Arc<ServerState>, agent: &str, task_emb: &[f32]) {
    if let Ok(store) = state.routing() {
        let _ = store.record_capability(agent, task_emb);
    }
}

/// Run a registered declarative agent on an input, in-daemon (no process).
/// Returns `{ output, turns, tool_calls, error? }`.
pub async fn handle_declagents_invoke(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DeclAgentInvokeParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let reg = state.declagents()?;
    let spec = reg
        .get(&params.id)
        .ok_or_else(|| format!("no declarative agent '{}'", params.id))?;
    if !spec.enabled {
        return Err(format!("agent '{}' is disabled", params.id));
    }
    let result = run_declarative(&spec, &params.input, state).await?;
    record_routing_outcome(state, &spec.id, &result);
    Ok(run_result_json(&result))
}

// --- declagents.route — capability-similarity routing (AgentNet milestone) ---
//
// AgentNet (arXiv:2504.00587) routes a task to the agent whose capability
// vector best matches the task: `argmax_i sim(c_task, c_i)`. This is the
// smallest in-repo slice of that idea — see
// docs/proposals/agentnet-self-organization.md. The capability vector is a
// cold-start embedding of the agent's identity + standing goal + tools (no
// learned history yet); the task vector is a query-side embedding of the
// need. Routing only *proposes* the agent; invocation (when requested) still
// flows through the governed declarative runner — tool allowlist + policy.

/// The text we embed to represent an agent's capability surface. Cold-start:
/// derived from the static spec (identity, goal, tools), not yet from observed
/// routing outcomes (the EMA-updated `c_i` of the full AgentNet design).
fn capability_text(spec: &car_registry::declarative::DeclarativeAgentSpec) -> String {
    let mut text = format!("{}. {}", spec.name, spec.identity);
    if !spec.standing_goal.is_empty() {
        text.push_str(&format!(" Goal: {}.", spec.standing_goal));
    }
    if !spec.tools.is_empty() {
        text.push_str(&format!(" Tools: {}.", spec.tools.join(", ")));
    }
    text
}

/// Cosine similarity. Returns 0.0 for a zero-norm vector (no NaN leaks into
/// the ranking) and for mismatched lengths — a query and document embedded by
/// different models/endpoints could disagree on dimension; scoring over a
/// silently truncated prefix (what `zip` would do) is worse than declining.
fn cosine(a: &[f32], b: &[f32]) -> f32 {
    if a.len() != b.len() {
        return 0.0;
    }
    let dot: f32 = a.iter().zip(b).map(|(x, y)| x * y).sum();
    let na: f32 = a.iter().map(|x| x * x).sum::<f32>().sqrt();
    let nb: f32 = b.iter().map(|x| x * x).sum::<f32>().sqrt();
    if na == 0.0 || nb == 0.0 {
        0.0
    } else {
        dot / (na * nb)
    }
}

/// Weight on embedding similarity vs. the learned success prior when ranking.
/// Similarity dominates so cold-start correctness holds; the prior nudges
/// toward agents that actually complete routed work.
const ROUTE_SIMILARITY_WEIGHT: f32 = 0.7;

/// Exploration constant for the unified success-prior UCB
/// (`car_memgine::utility::UtilityPosterior::ucb`). 0.0 = pure exploitation:
/// the prior is the Beta(success+1, fail+1) posterior *mean*, whose uniform
/// cold-start value is exactly [`car_registry::routing::NEUTRAL_PRIOR`] (0.5)
/// — a never-tried service keeps the documented neutral prior instead of an
/// inflated uncertainty bonus. Routing deliberately does not explore on
/// uncertainty (unlike memory retrieval, where the caller opts in): a routed
/// need runs on ONE service, and similarity already gives cold candidates a
/// fair shot. Turning exploration on later is this one constant.
const ROUTE_PRIOR_EXPLORATION: f64 = 0.0;

/// The success prior for ranking — H2 Part 2's ONE scoring substrate
/// (`docs/proposals/h2-builder-discovery-acceptance.md`). Folds the raw
/// `successes`/`failures` that `~/.car/routing.json` persists under each of
/// `keys` into a single Beta(success+1, fail+1) posterior
/// (`car_memgine::utility::UtilityPosterior`) scored by the deterministic UCB.
/// Multiple keys exist because a declarative agent learns under its agent id
/// (`declagents.route`/`invoke` outcomes) *and* under its
/// `agentdns://local/agent/<id>` identifier (`discovery.report` outcomes) —
/// summing the counts makes it one agent, one score, on both surfaces. The
/// legacy EMA field remains persisted for display (`declagents.routing_stats`)
/// but no longer drives ranking.
fn posterior_success_prior(
    routing: &car_registry::routing::RoutingSnapshot,
    keys: &[&str],
) -> f32 {
    let (mut successes, mut failures) = (0u64, 0u64);
    for key in keys {
        let (s, f) = routing.outcome_counts(key);
        successes += s;
        failures += f;
    }
    car_memgine::utility::UtilityPosterior::from_counts(successes, failures)
        .ucb(ROUTE_PRIOR_EXPLORATION) as f32
}

/// The `agentdns://local/agent/<id>` identifier a declarative agent surfaces
/// under in `discovery.resolve` — the second routing-store key its outcomes may
/// be recorded against (via `discovery.report`). None only if the id somehow
/// isn't identifier-safe (registry ids are filename-safe ⊆ the identifier
/// charset, so this is defensive).
fn declarative_discovery_key(agent_id: &str) -> Option<String> {
    car_connectors::discovery::ServiceIdentifier::local("agent", agent_id)
        .ok()
        .map(|i| i.to_string())
}

/// [`posterior_success_prior`] over a declarative agent's two routing keys:
/// its agent id and its discovery identifier. Shared by `rank_agents`
/// (`declagents.route`) and `score_service` (`discovery.resolve`) so a
/// declarative agent carries the SAME prior on both surfaces.
fn declarative_success_prior(
    routing: &car_registry::routing::RoutingSnapshot,
    agent_id: &str,
) -> f32 {
    match declarative_discovery_key(agent_id) {
        Some(ident) => posterior_success_prior(routing, &[agent_id, &ident]),
        None => posterior_success_prior(routing, &[agent_id]),
    }
}

/// Weight on a learned forward edge when a delegating agent (`from`) is routing
/// onward. Additive on top of the similarity/prior blend, so a proven
/// delegation path re-ranks peers without overriding a much stronger match.
const ROUTE_EDGE_WEIGHT: f32 = 0.2;

/// Maximum agents on one routing path before the DAG guard refuses to forward
/// further — bounds the Forward chain and guarantees termination.
const MAX_ROUTE_HOPS: usize = 4;

/// Weight on learned similarity (need vs the agent's reinforced capability
/// centroid) vs. cold-start similarity (need vs static capability text) once an
/// agent has a learned vector. Below 0.5 so the static description still anchors
/// ranking and a few lucky successes can't fully capture an agent.
const LEARNED_SIM_WEIGHT: f32 = 0.4;

/// Blend cold-start similarity with learned-centroid similarity. Falls back to
/// pure cold-start until the agent has succeeded at least once (no centroid).
fn blended_similarity(coldstart: f32, learned: Option<f32>) -> f32 {
    match learned {
        Some(l) => (1.0 - LEARNED_SIM_WEIGHT) * coldstart + LEARNED_SIM_WEIGHT * l,
        None => coldstart,
    }
}

/// Blend embedding similarity with an agent's learned success prior into one
/// ranking score. Cosine is clamped at 0 so an anti-correlated agent can't post
/// a negative score that an unrelated-but-unproven agent (prior 0.5) would beat
/// on the prior term alone.
fn blended_score(similarity: f32, success_prior: f32) -> f32 {
    let sim = similarity.max(0.0);
    ROUTE_SIMILARITY_WEIGHT * sim + (1.0 - ROUTE_SIMILARITY_WEIGHT) * success_prior
}

/// Final routing score: the similarity/prior blend plus a learned forward-edge
/// boost. `edge_weight` is 0 at network entry (no delegating agent) or when no
/// edge has been learned yet, so this reduces to [`blended_score`] in the cold
/// case and only the learned topology pulls it away. This is an unbounded
/// *ranking* score (a fully-forwarded agent can exceed 1.0), not a probability —
/// only its order across candidates is meaningful.
fn route_score(similarity: f32, success_prior: f32, edge_weight: f32) -> f32 {
    blended_score(similarity, success_prior) + ROUTE_EDGE_WEIGHT * edge_weight
}

/// An agent is excluded as a forward target when it is the delegator itself or
/// is already on the routing path (cycle guard — Forward must preserve the DAG).
fn is_excluded(id: &str, from: Option<&str>, visited: &[String]) -> bool {
    from == Some(id) || visited.iter().any(|v| v == id)
}

/// Rank `agents` for a need, given the need's query embedding and each agent's
/// pre-computed capability-doc embedding (positionally aligned with `agents`).
/// Returns `(index, score, similarity, success_prior, edge_weight)` sorted by
/// score descending, ties broken by agent id for restart-determinism. Shared by
/// single-need routing and per-subtask Split routing so both score identically.
fn rank_agents(
    need_emb: &[f32],
    agent_embs: &[Vec<f32>],
    agents: &[car_registry::declarative::DeclarativeAgentSpec],
    routing: &car_registry::routing::RoutingSnapshot,
    from: Option<&str>,
) -> Vec<(usize, f32, f32, f32, f32)> {
    let mut ranked: Vec<(usize, f32, f32, f32, f32)> = agent_embs
        .iter()
        .enumerate()
        .map(|(i, e)| {
            let coldstart = cosine(need_emb, e);
            // Learned-centroid similarity, if the agent has succeeded before.
            let learned = routing
                .learned_capability(&agents[i].id)
                .map(|c| cosine(need_emb, c));
            let similarity = blended_similarity(coldstart, learned);
            let prior = declarative_success_prior(routing, &agents[i].id);
            // Learned forward edge from the delegating agent, if any.
            let edge = from.map_or(0.0, |f| routing.edge_weight(f, &agents[i].id));
            (
                i,
                route_score(similarity, prior, edge),
                similarity,
                prior,
                edge,
            )
        })
        .collect();
    // Descending score; ties broken by agent id so the pick is deterministic
    // across restarts (registry iteration order is not).
    ranked.sort_by(|a, b| {
        b.1.total_cmp(&a.1)
            .then_with(|| agents[a.0].id.cmp(&agents[b.0].id))
    });
    ranked
}

#[derive(Deserialize)]
struct DeclAgentRouteParams {
    /// Natural-language description of the task to route.
    need: String,
    /// If true, also run the top-ranked agent on `need` and include its result.
    #[serde(default)]
    invoke: bool,
    /// The agent forwarding this need onward (AgentNet's Forward op). Excluded
    /// from candidates; on invoke, the directed edge `from → chosen` is
    /// reinforced or weakened by the outcome. Absent at network entry.
    #[serde(default)]
    from: Option<String>,
    /// Agents already on this routing path — the DAG/cycle guard. Excluded from
    /// candidates; the caller accumulates this as it walks a Forward chain.
    #[serde(default)]
    visited: Vec<String>,
}

/// Number of ranked candidates returned to the caller.
const ROUTE_TOP_K: usize = 3;

/// Route a need to the best-matching declarative agent. Ranks by a blend of
/// embedding similarity (need vs. each agent's capability surface) and the
/// agent's learned success prior. Returns `{ chosen, candidates: [{ id, name,
/// score, similarity, success_rate }], invoked, result? }`. With `invoke: true`,
/// the top agent is run on `need` and its `{ output, turns, tool_calls, error? }`
/// lands in `result`.
pub async fn handle_declagents_route(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DeclAgentRouteParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;

    // An empty need embeds to noise and would route (and with invoke, run) an
    // essentially random agent — then pollute its prior. Refuse up front.
    if params.need.trim().is_empty() {
        return Err("need must be a non-empty task description".to_string());
    }

    // DAG guard: a Forward chain must terminate. Refuse once the path is at the
    // hop limit (the caller accumulates `visited` as it walks).
    if params.visited.len() >= MAX_ROUTE_HOPS {
        return Err(format!(
            "routing path exceeded {MAX_ROUTE_HOPS} hops (cycle or runaway forward)"
        ));
    }

    let from = params.from.as_deref();
    let reg = state.declagents()?;
    // Eligible forward targets: enabled, and neither the delegator nor any
    // agent already on the path (cycle guard).
    let agents: Vec<_> = reg
        .list()
        .into_iter()
        .filter(|s| s.enabled && !is_excluded(&s.id, from, &params.visited))
        .collect();
    if agents.is_empty() {
        return Err("no eligible declarative agents to route to".to_string());
    }

    // The embedder is asymmetric (Qwen3-Embedding): the need is a query (gets
    // the Instruct/Query prefix), the capability docs are embedded raw. So two
    // calls, not one batch — under a single admission permit. Embeds load
    // model weights, so share the generation gate (same as `handle_embed`) to
    // keep a burst from bypassing the concurrency cap.
    let engine = crate::handler::get_inference_engine(state);
    let _permit = state.admission.acquire().await;
    let need_embs = engine
        .embed(car_inference::EmbedRequest {
            texts: vec![params.need.clone()],
            model: None,
            instruction: Some("Match this task to the agent best able to perform it".to_string()),
            is_query: true,
        })
        .await
        .map_err(|e| format!("embed failed: {e}"))?;
    let agent_embs = engine
        .embed(car_inference::EmbedRequest {
            texts: agents.iter().map(capability_text).collect(),
            model: None,
            instruction: None,
            is_query: false,
        })
        .await
        .map_err(|e| format!("embed failed: {e}"))?;
    drop(_permit);

    let need_emb = need_embs
        .first()
        .ok_or_else(|| "embedder returned no vectors".to_string())?;

    // Learned priors (one snapshot, read once). Absent store ⇒ cold-start
    // neutral priors for everyone, so ranking falls back to pure similarity.
    let routing = state.routing().map(|s| s.snapshot()).unwrap_or_default();

    let ranked = rank_agents(need_emb, &agent_embs, &agents, &routing, from);

    let candidates: Vec<Value> = ranked
        .iter()
        .take(ROUTE_TOP_K)
        .map(|(i, score, similarity, prior, edge)| {
            json!({
                "id": agents[*i].id,
                "name": agents[*i].name,
                "score": score,
                "similarity": similarity,
                "success_rate": prior,
                "edge_weight": edge,
            })
        })
        .collect();

    let chosen = &agents[ranked[0].0];
    let result = if params.invoke {
        let run = run_declarative(chosen, &params.need, state).await?;
        record_routing_outcome(state, &chosen.id, &run);
        if run_is_recordable(&run) {
            // On a genuine success, fold this need into the agent's capability
            // centroid so similar future needs favor it (c_i reinforcement).
            if run_succeeded(&run) {
                record_routing_capability(state, &chosen.id, need_emb);
            }
            // Reinforce the forward edge that brought us here (Forward learning).
            if let Some(f) = from {
                record_routing_edge(state, f, &chosen.id, run_succeeded(&run));
            }
        }
        Some(run_result_json(&run))
    } else {
        None
    };

    // The path the caller should carry into the next Forward hop. Echoing it
    // (rather than trusting the caller to reconstruct it) keeps the DAG/hop-cap
    // guard reliable: every hop strictly grows `visited`, so MAX_ROUTE_HOPS
    // always fires and cycles through prior delegators can't reopen.
    let mut next_visited = params.visited.clone();
    next_visited.push(chosen.id.clone());

    Ok(json!({
        "chosen": chosen.id,
        "candidates": candidates,
        "invoked": params.invoke,
        "result": result,
        "next_visited": next_visited,
    }))
}

// --- declagents.route_split — Split op: decompose a need, fan out the parts ---
//
// AgentNet's Split decomposes a task into subtasks and routes each. Here it's a
// fan-out primitive: a planner model breaks `need` into independent subtasks,
// each is routed by the same capability-similarity ranking as `route`, and
// (optionally) run. Decomposition is the one model-driven step — it only
// *proposes* the split; every subtask still routes deterministically and runs
// on the governed declarative runner. Any decomposition failure falls back to
// treating the whole need as a single subtask, so Split never does worse than
// `route`.

/// Default / hard cap on the number of subtasks a need is split into.
const DEFAULT_MAX_SUBTASKS: usize = 5;
const MAX_SUBTASKS_CAP: usize = 10;
const DEFAULT_SAD_HINTS: usize = 15;
const MAX_SAD_HINTS: usize = 50;
const DEFAULT_SAD_ITERATIONS: usize = 1;
const MAX_SAD_ITERATIONS: usize = 3;
const DEFAULT_SAD_CONVERGENCE_JACCARD: f64 = 0.6;
const DEFAULT_CANDIDATES_PER_STEP: usize = 5;
const MAX_CANDIDATES_PER_STEP: usize = 10;

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum DecompositionMode {
    Vanilla,
    Sad,
}

impl Default for DecompositionMode {
    fn default() -> Self {
        Self::Vanilla
    }
}

#[derive(Debug, Clone)]
struct SadConfig {
    mode: DecompositionMode,
    hints: usize,
    iterations: usize,
    convergence_jaccard: f64,
}

impl SadConfig {
    fn new(
        mode: DecompositionMode,
        hints: Option<usize>,
        iterations: Option<usize>,
        convergence_jaccard: Option<f64>,
    ) -> Self {
        Self {
            mode,
            hints: hints.unwrap_or(DEFAULT_SAD_HINTS).clamp(1, MAX_SAD_HINTS),
            iterations: iterations
                .unwrap_or(DEFAULT_SAD_ITERATIONS)
                .clamp(1, MAX_SAD_ITERATIONS),
            convergence_jaccard: convergence_jaccard
                .unwrap_or(DEFAULT_SAD_CONVERGENCE_JACCARD)
                .clamp(0.0, 1.0),
        }
    }
}

#[derive(Debug, Clone)]
struct DecompositionTrace {
    mode: DecompositionMode,
    rounds: usize,
    initial_subtasks: Vec<String>,
    final_subtasks: Vec<String>,
    hints: Vec<String>,
    hint_jaccard: Option<f64>,
}

/// Parse a planner model's JSON reply into a clean subtask list. Tolerant by
/// design: anything malformed, empty, or missing the `subtasks` array falls
/// back to `[need]` so Split degrades to a single route rather than failing.
fn parse_subtasks(raw: &str, need: &str, max: usize) -> Vec<String> {
    let subs: Vec<String> = serde_json::from_str::<Value>(raw)
        .ok()
        .and_then(|v| v.get("subtasks").and_then(|s| s.as_array()).cloned())
        .into_iter()
        .flatten()
        .filter_map(|v| v.as_str().map(|s| s.trim().to_string()))
        .filter(|s| !s.is_empty())
        .take(max)
        .collect();
    if subs.is_empty() {
        vec![need.to_string()]
    } else {
        subs
    }
}

fn decomposition_prompt(need: &str, max: usize, hints: &[String]) -> String {
    if hints.is_empty() {
        return format!(
            "You are a task planner. Decompose the request below into at most {max} \
         INDEPENDENT subtasks, each handleable by a separate specialist agent. \
         If the request is already atomic, return it as a single subtask. \
         Respond with JSON only: {{\"subtasks\": [\"...\", \"...\"]}}.\n\n\
         Request: {need}"
        );
    }
    format!(
        "You are a task planner. Decompose the request below into at most {max} \
         INDEPENDENT subtasks, each handleable by exactly one available skill or \
         service. Use the available skills only as vocabulary hints; do not add \
         steps that the request does not require. If the request is already \
         atomic, return it as a single subtask. Respond with JSON only: \
         {{\"subtasks\": [\"...\", \"...\"]}}.\n\n\
         Available skills that may be relevant: {}\n\nRequest: {need}",
        hints.join(", ")
    )
}

/// Ask a planner model to decompose `need` into independent subtasks. Always
/// returns at least one (falls back to `[need]` on any inference/parse failure).
async fn decompose_need_with_hints(
    state: &Arc<ServerState>,
    need: &str,
    max: usize,
    hints: &[String],
) -> Vec<String> {
    let prompt = decomposition_prompt(need, max, hints);
    let engine = crate::handler::get_inference_engine(state);
    let _permit = state.admission.acquire().await;
    let raw = engine
        .generate(car_inference::GenerateRequest {
            prompt,
            response_format: Some(car_inference::ResponseFormat::JsonObject),
            ..Default::default()
        })
        .await;
    drop(_permit);
    match raw {
        Ok(text) => parse_subtasks(&text, need, max),
        Err(_) => vec![need.to_string()],
    }
}

async fn decompose_need(state: &Arc<ServerState>, need: &str, max: usize) -> Vec<String> {
    decompose_need_with_hints(state, need, max, &[]).await
}

fn hint_jaccard(a: &[String], b: &[String]) -> f64 {
    let left: HashSet<&str> = a.iter().map(String::as_str).collect();
    let right: HashSet<&str> = b.iter().map(String::as_str).collect();
    if left.is_empty() && right.is_empty() {
        return 1.0;
    }
    let intersection = left.intersection(&right).count() as f64;
    let union = left.union(&right).count() as f64;
    if union == 0.0 {
        1.0
    } else {
        intersection / union
    }
}

fn truncate_hint(s: &str, max: usize) -> String {
    let mut out: String = s.chars().take(max).collect();
    if out.len() < s.len() {
        out.push_str("...");
    }
    out
}

fn build_agent_hints(
    subtasks: &[String],
    sub_embs: &[Vec<f32>],
    agent_embs: &[Vec<f32>],
    agents: &[car_registry::declarative::DeclarativeAgentSpec],
    routing: &car_registry::routing::RoutingSnapshot,
    limit: usize,
) -> Vec<String> {
    let mut hints = BTreeMap::new();
    for (i, _sub) in subtasks.iter().enumerate() {
        let Some(emb) = sub_embs.get(i) else {
            continue;
        };
        for (idx, ..) in rank_agents(emb, agent_embs, agents, routing, None)
            .into_iter()
            .take(limit)
        {
            let agent = &agents[idx];
            hints.entry(agent.id.clone()).or_insert_with(|| {
                truncate_hint(&format!("{}: {}", agent.name, capability_text(agent)), 180)
            });
            if hints.len() >= limit {
                break;
            }
        }
        if hints.len() >= limit {
            break;
        }
    }
    hints.into_values().collect()
}

async fn embed_query_texts(
    state: &Arc<ServerState>,
    texts: Vec<String>,
    instruction: &str,
) -> Result<Vec<Vec<f32>>, String> {
    let engine = crate::handler::get_inference_engine(state);
    let _permit = state.admission.acquire().await;
    let out = engine
        .embed(car_inference::EmbedRequest {
            texts,
            model: None,
            instruction: Some(instruction.to_string()),
            is_query: true,
        })
        .await
        .map_err(|e| format!("embed failed: {e}"))?;
    drop(_permit);
    Ok(out)
}

async fn decompose_with_agent_sad(
    state: &Arc<ServerState>,
    need: &str,
    max: usize,
    config: &SadConfig,
    agents: &[car_registry::declarative::DeclarativeAgentSpec],
    agent_embs: &[Vec<f32>],
    routing: &car_registry::routing::RoutingSnapshot,
) -> Result<DecompositionTrace, String> {
    let initial = decompose_need(state, need, max).await;
    if config.mode == DecompositionMode::Vanilla {
        return Ok(DecompositionTrace {
            mode: config.mode,
            rounds: 1,
            initial_subtasks: initial.clone(),
            final_subtasks: initial,
            hints: Vec::new(),
            hint_jaccard: None,
        });
    }

    let mut current = initial.clone();
    let mut previous_hints: Option<Vec<String>> = None;
    let mut last_hints = Vec::new();
    let mut last_jaccard = None;
    let mut rounds = 1;
    for _ in 0..config.iterations {
        let sub_embs = embed_query_texts(
            state,
            current.clone(),
            "Match this task to the agent best able to perform it",
        )
        .await?;
        let hints = build_agent_hints(
            &current,
            &sub_embs,
            agent_embs,
            agents,
            routing,
            config.hints,
        );
        if let Some(prev) = previous_hints.as_ref() {
            let j = hint_jaccard(prev, &hints);
            last_jaccard = Some(j);
            if j >= config.convergence_jaccard {
                last_hints = hints;
                break;
            }
        }
        let refined = decompose_need_with_hints(state, need, max, &hints).await;
        rounds += 1;
        current = refined;
        previous_hints = Some(hints.clone());
        last_hints = hints;
    }
    Ok(DecompositionTrace {
        mode: config.mode,
        rounds,
        initial_subtasks: initial,
        final_subtasks: current,
        hints: last_hints,
        hint_jaccard: last_jaccard,
    })
}

#[derive(Deserialize)]
struct DeclAgentSplitParams {
    /// The composite need to decompose and fan out.
    need: String,
    /// If true, run each subtask's chosen agent and include its result.
    #[serde(default)]
    invoke: bool,
    /// Cap on the number of subtasks (clamped to [1, 10]). Default 5.
    #[serde(default)]
    max_subtasks: Option<usize>,
    #[serde(default)]
    decomposition_mode: DecompositionMode,
    #[serde(default)]
    sad_hints: Option<usize>,
    #[serde(default)]
    sad_iterations: Option<usize>,
    #[serde(default)]
    sad_convergence_jaccard: Option<f64>,
}

/// Split a composite need into subtasks and route each to its best-matching
/// agent. Returns `{ subtasks: [{ subtask, chosen, score, result? }], count,
/// invoked }`. With `invoke: true`, each subtask's chosen agent runs on that
/// subtask (governed path) and outcomes/capability are recorded; a per-subtask
/// infra failure is captured into that subtask's `result.error` and the rest
/// of the fan-out continues.
///
/// Cost note: `invoke: true` runs up to `max_subtasks` full agent loops
/// **sequentially** within one call — potentially long wall-clock. Callers
/// wanting bounded latency should keep `max_subtasks` small or route subtasks
/// themselves (`invoke: false` returns the routing decisions to drive).
pub async fn handle_declagents_route_split(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DeclAgentSplitParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    if params.need.trim().is_empty() {
        return Err("need must be a non-empty task description".to_string());
    }
    let max = params
        .max_subtasks
        .unwrap_or(DEFAULT_MAX_SUBTASKS)
        .clamp(1, MAX_SUBTASKS_CAP);

    let reg = state.declagents()?;
    let agents: Vec<_> = reg.list().into_iter().filter(|s| s.enabled).collect();
    if agents.is_empty() {
        return Err("no enabled declarative agents to route to".to_string());
    }

    // Embed the capability docs once (shared across SAD and final routing).
    let engine = crate::handler::get_inference_engine(state);
    let _permit = state.admission.acquire().await;
    let agent_embs = engine
        .embed(car_inference::EmbedRequest {
            texts: agents.iter().map(capability_text).collect(),
            model: None,
            instruction: None,
            is_query: false,
        })
        .await
        .map_err(|e| format!("embed failed: {e}"))?;
    drop(_permit);

    // One snapshot for the whole split — subtasks rank against a consistent
    // view; learning from earlier subtasks lands for the next route, not
    // mid-split (avoids re-reading the store per subtask).
    let routing = state.routing().map(|s| s.snapshot()).unwrap_or_default();

    let sad = SadConfig::new(
        params.decomposition_mode,
        params.sad_hints,
        params.sad_iterations,
        params.sad_convergence_jaccard,
    );
    let decomposition = decompose_with_agent_sad(
        state,
        &params.need,
        max,
        &sad,
        &agents,
        &agent_embs,
        &routing,
    )
    .await?;
    let subtasks = decomposition.final_subtasks.clone();

    let sub_embs = embed_query_texts(
        state,
        subtasks.clone(),
        "Match this task to the agent best able to perform it",
    )
    .await?;

    let mut routed = Vec::with_capacity(subtasks.len());
    for (i, sub) in subtasks.iter().enumerate() {
        let Some(need_emb) = sub_embs.get(i) else {
            continue;
        };
        let ranked = rank_agents(need_emb, &agent_embs, &agents, &routing, None);
        let (idx, score, ..) = ranked[0]; // agents non-empty ⇒ ranked non-empty
        let chosen = &agents[idx];
        let result = if params.invoke {
            match run_declarative(chosen, sub, state).await {
                Ok(run) => {
                    record_routing_outcome(state, &chosen.id, &run);
                    if run_is_recordable(&run) && run_succeeded(&run) {
                        record_routing_capability(state, &chosen.id, need_emb);
                    }
                    Some(run_result_json(&run))
                }
                // Best-effort fan-out: an infra failure on one subtask must not
                // discard the rest — earlier subtasks may already have run with
                // irreversible side effects. Capture it and carry on, matching
                // the tolerant parse/recording paths.
                Err(e) => Some(json!({ "error": e })),
            }
        } else {
            None
        };
        routed.push(json!({
            "subtask": sub,
            "chosen": chosen.id,
            "score": score,
            "result": result,
        }));
    }

    Ok(json!({
        "subtasks": routed,
        // routed.len() rather than subtasks.len(): invariant-correct regardless
        // of the embedder's per-text contract.
        "count": routed.len(),
        "invoked": params.invoke,
        "decomposition_mode": decomposition.mode,
        "rounds": decomposition.rounds,
        "initial_subtasks": decomposition.initial_subtasks,
        "final_subtasks": decomposition.final_subtasks,
        "hints": decomposition.hints,
        "hint_jaccard": decomposition.hint_jaccard,
    }))
}

/// Read-only view of the learned routing topology: per-agent success stats and
/// directed agent→agent edge weights. Returns `{ agents: { id: { successes,
/// failures, ema_success_rate, learned } }, edges: { from: { to: weight } } }`.
/// `learned` is a bool — the capability centroid itself is omitted (it's a
/// large embedding, noise for observability). Empty when nothing has routed.
pub async fn handle_declagents_routing_stats(state: &Arc<ServerState>) -> Result<Value, String> {
    let snapshot = state.routing()?.snapshot();
    let agents: serde_json::Map<String, Value> = snapshot
        .agents
        .iter()
        .map(|(id, s)| {
            (
                id.clone(),
                json!({
                    "successes": s.successes,
                    "failures": s.failures,
                    "ema_success_rate": s.ema_success_rate,
                    "learned": !s.learned_vector.is_empty(),
                }),
            )
        })
        .collect();
    Ok(json!({ "agents": agents, "edges": snapshot.edges }))
}

// --- discovery.resolve — AgentDNS-style service discovery -------------------
//
// AgentDNS (arXiv:2505.22368) resolves a natural-language need into specific
// service identifiers across vendors. This is the LOCAL resolver: it resolves
// against CAR's own registered services, naming each under the
// `agentdns://organization/category/name` scheme. Providers, all behind one
// `services` record shape: declarative agents (ranked by the same capability
// similarity as `declagents.route`, so discovery rides the AgentNet learning —
// success priors + capability centroids — for free), observe-only registry
// services (`~/.car/registry/`, the dashboard-registered local services),
// connected MCP connector tools, installed external CLIs, A2A peer skills, and
// the opt-in remote Parslee root server (the cross-vendor case). Only
// declarative agents carry routing learning; the rest rank on cold-start
// similarity.

const DISCOVERY_DEFAULT_LIMIT: usize = 5;
const DISCOVERY_MAX_LIMIT: usize = 50;

/// Per-provider bound so a slow provider degrades discovery to whatever else
/// resolved rather than wedging the call: a hung remote MCP server (the first
/// `discovery.resolve` may trigger a cold connector dial with no HTTP timeout of
/// its own), or external-agent detection spawning `--version` subprocesses.
const DISCOVERY_PROVIDER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// TTL for the cached external-agent detection — `detect()` spawns a
/// `--version` subprocess per installed CLI, far too costly to run on every
/// `discovery.resolve`. Installed CLIs change rarely, so a minute is ample.
const EXTERNAL_DETECT_TTL: std::time::Duration = std::time::Duration::from_secs(60);

#[derive(Deserialize)]
struct DiscoveryResolveParams {
    /// Natural-language description of the capability being sought.
    need: String,
    /// Max services to return (clamped to [1, 50]). Default 5.
    #[serde(default)]
    limit: Option<usize>,
}

/// One candidate service surfaced by a discovery provider, before ranking.
#[derive(Clone)]
struct DiscoveredService {
    /// Formatted `agentdns://…` identifier.
    identifier: String,
    name: String,
    /// Service kind — `&'static` for the local providers, but owned because the
    /// remote-root provider carries vendor-defined kinds/protocols.
    kind: String,
    protocol: String,
    /// Text embedded (as a doc) and matched against the need.
    capability_text: String,
    /// Declarative agent id when this service carries AgentNet routing learning
    /// (success prior + capability centroid). None for other kinds.
    agent_id: Option<String>,
    /// Concrete network endpoint a caller can reach the service at, when the
    /// kind has one (e.g. a registry service's dashboard URL). Carried so
    /// `route_compose` can emit an actionable `invoke_target`. None for kinds
    /// invoked through a governed surface keyed off the identifier instead.
    endpoint: Option<String>,
}

async fn gather_discovered_services(
    state: &Arc<ServerState>,
    need: &str,
    remote_limit: usize,
) -> Vec<DiscoveredService> {
    // Local providers (declarative agents, registry) are synchronous bounded
    // filesystem/in-memory reads — they can't hang, so they run unwrapped. The
    // network providers below each get DISCOVERY_PROVIDER_TIMEOUT because they
    // can block on a remote socket or a subprocess; one slow vendor degrades
    // discovery to whatever else resolved rather than wedging the whole call.
    let mut services = declarative_services(state);
    services.extend(registry_services());
    match tokio::time::timeout(DISCOVERY_PROVIDER_TIMEOUT, connector_services(state)).await {
        Ok(connectors) => services.extend(connectors),
        Err(_) => {
            tracing::warn!("discovery: connector provider timed out; skipping")
        }
    }
    match tokio::time::timeout(DISCOVERY_PROVIDER_TIMEOUT, external_agent_services()).await {
        Ok(external) => services.extend(external),
        Err(_) => {
            tracing::warn!("discovery: external-agent provider timed out; skipping")
        }
    }
    match tokio::time::timeout(DISCOVERY_PROVIDER_TIMEOUT, a2a_peer_services()).await {
        Ok(peers) => services.extend(peers),
        Err(_) => {
            tracing::warn!("discovery: a2a-peer provider timed out; skipping")
        }
    }
    match tokio::time::timeout(
        DISCOVERY_PROVIDER_TIMEOUT,
        remote_root_services(state, need, remote_limit),
    )
    .await
    {
        Ok(remote) => services.extend(remote),
        Err(_) => {
            tracing::warn!("discovery: remote-root provider timed out; skipping")
        }
    }
    let mut seen = HashSet::new();
    services.retain(|s| seen.insert(s.identifier.clone()));
    services
}

/// Provider: enabled declarative agents. These carry routing learning, so they
/// rank with the blended similarity + success prior; others use a neutral prior.
fn declarative_services(state: &Arc<ServerState>) -> Vec<DiscoveredService> {
    let Ok(reg) = state.declagents() else {
        return Vec::new();
    };
    reg.list()
        .into_iter()
        .filter(|s| s.enabled)
        .filter_map(|s| {
            // Agent ids are filename-safe (⊆ identifier charset); skip on the
            // off chance one isn't rather than fail the whole resolution.
            let identifier =
                car_connectors::discovery::ServiceIdentifier::local("agent", &s.id).ok()?;
            let capability = capability_text(&s);
            Some(DiscoveredService {
                identifier: identifier.to_string(),
                name: s.name,
                kind: "declarative".to_string(),
                protocol: "in-daemon".to_string(),
                capability_text: capability,
                agent_id: Some(s.id),
                endpoint: None,
            })
        })
        .collect()
}

/// Discovery treats a registry entry as routable only if its heartbeat is this
/// recent. Mirrors the registry reaper's default (`reap_stale(60)`, run by the
/// menubar ~every 30s): a healthy agent heartbeats every 20s, so two missed
/// beats means dead. Discovery enforces the bound *itself* rather than trust the
/// reaper because a headless daemon may have no menubar reaping the directory —
/// without this, a crashed-but-unreaped entry would still read `Running` and a
/// route would target its dead port.
const REGISTRY_STALE_AFTER_SECS: u64 = 60;

/// Whether a registry entry's heartbeat is recent enough to route to. `now_secs`
/// is UNIX seconds; passing `0` (a clock-read failure) fails open — better to
/// surface a possibly-stale service than to blank discovery on a clock glitch.
fn registry_entry_is_fresh(entry: &car_registry::AgentEntry, now_secs: u64) -> bool {
    now_secs.saturating_sub(entry.last_heartbeat_at) <= REGISTRY_STALE_AFTER_SECS
}

/// Map one observe-only registry entry to a discoverable service. Pure so the
/// status filter, capability-text composition, and endpoint wiring are unit
/// testable without touching `~/.car/registry/`. Returns None for a service
/// that isn't routable (stopping/errored) or whose name can't form an
/// identifier.
fn registry_entry_to_service(entry: car_registry::AgentEntry) -> Option<DiscoveredService> {
    // Only running/idle services are routable. A stopping or errored entry is
    // about to vanish (or can't serve), so surfacing it would route work to a
    // dead endpoint.
    if !matches!(
        entry.status,
        car_registry::AgentStatus::Running | car_registry::AgentStatus::Idle
    ) {
        return None;
    }
    // Registry names are validated to the identifier charset on `register`, but
    // skip rather than fail the rest if one somehow isn't.
    let identifier = car_connectors::discovery::ServiceIdentifier::local("service", &entry.name)
        .ok()?
        .to_string();
    let label = entry.display_name.clone().unwrap_or_else(|| entry.name.clone());
    // Capability text drives ranking. With a description, "<label>. <cap>";
    // without one, the bare label (the service still resolves, just ranks on
    // its name — the pre-schema baseline).
    let capability_text = match entry.capability.as_deref().map(str::trim) {
        Some(cap) if !cap.is_empty() => format!("{label}. {cap}"),
        _ => label.clone(),
    };
    Some(DiscoveredService {
        identifier,
        name: label,
        kind: "registry".to_string(),
        protocol: "http".to_string(),
        capability_text,
        agent_id: None,
        endpoint: Some(entry.dashboard_url),
    })
}

/// Provider: locally-running services that announced themselves to the
/// observe-only file registry (`~/.car/registry/`, written by `register_agent` /
/// the supervisor). These are the dashboard-registered services the menubar
/// lists; surfacing them here makes a heartbeating local service routable
/// instead of invisible to discovery (#374-follow-up). No routing learning
/// (agent_id=None) — they rank on cold-start similarity against their
/// `capability` text. Synchronous filesystem read like `declarative_services`,
/// so it isn't wrapped in the per-provider network timeout.
fn registry_services() -> Vec<DiscoveredService> {
    let Ok(reg) = car_registry::AgentRegistry::user_default() else {
        return Vec::new();
    };
    let Ok(entries) = reg.list() else {
        return Vec::new();
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    entries
        .into_iter()
        .filter(|e| registry_entry_is_fresh(e, now))
        .filter_map(registry_entry_to_service)
        .collect()
}

/// Provider: enabled tools of connected remote MCP connectors. Best-effort —
/// a disconnected connector, an uncached tool list, or a tool whose name can't
/// form an identifier is simply skipped, so a flaky connector never fails
/// discovery of everything else.
async fn connector_services(state: &Arc<ServerState>) -> Vec<DiscoveredService> {
    state.ensure_connectors_loaded().await;
    let mgr = state.connectors();
    let mut out = Vec::new();
    for status in mgr.list().await {
        if !status.connected {
            continue;
        }
        let Ok(tools) = mgr.tools(&status.slug).await else {
            continue;
        };
        for t in tools {
            if !t.enabled {
                continue;
            }
            // agentdns://<connector-slug>/tool/<tool-name>.
            let Ok(identifier) = car_connectors::discovery::ServiceIdentifier::new(
                status.slug.clone(),
                [String::from("tool")],
                t.name.clone(),
            ) else {
                continue;
            };
            let capability_text = if t.description.is_empty() {
                t.name.clone()
            } else {
                format!("{}. {}", t.name, t.description)
            };
            out.push(DiscoveredService {
                identifier: identifier.to_string(),
                name: t.canonical,
                kind: "connector".to_string(),
                protocol: "mcp".to_string(),
                capability_text,
                agent_id: None,
                endpoint: None,
            });
        }
    }
    out
}

/// Capability text for an installed external agent CLI — its label plus the
/// features it advertises (the spec carries no free-text description).
fn external_capability_text(spec: &car_external_agents::ExternalAgentSpec) -> String {
    let c = &spec.capabilities;
    let feats: Vec<&str> = [
        (c.tool_use, "tool use"),
        (c.mcp, "MCP"),
        (c.hooks, "hooks"),
        (c.sessions, "sessions"),
        (c.streaming, "streaming"),
    ]
    .into_iter()
    .filter_map(|(on, label)| on.then_some(label))
    .collect();
    let mut text = format!("{}. Agentic coding CLI.", spec.display_name);
    if !feats.is_empty() {
        text.push_str(&format!(" Capabilities: {}.", feats.join(", ")));
    }
    text
}

/// Process-global TTL cache for external-agent detection. External CLIs are a
/// machine-level fact, not session-scoped, so one cache serves all callers.
fn external_detect_cache() -> &'static tokio::sync::Mutex<
    Option<(
        std::time::Instant,
        Vec<car_external_agents::ExternalAgentSpec>,
    )>,
> {
    static CACHE: std::sync::OnceLock<
        tokio::sync::Mutex<
            Option<(
                std::time::Instant,
                Vec<car_external_agents::ExternalAgentSpec>,
            )>,
        >,
    > = std::sync::OnceLock::new();
    CACHE.get_or_init(|| tokio::sync::Mutex::new(None))
}

/// Provider: installed external agentic CLIs (Claude Code, Codex, Gemini) on
/// `$PATH`. Detection is cached for [`EXTERNAL_DETECT_TTL`] to avoid re-spawning
/// `--version` per CLI on every resolve. No routing learning (agent_id=None).
async fn external_agent_services() -> Vec<DiscoveredService> {
    let specs = {
        let mut guard = external_detect_cache().lock().await;
        let fresh = guard
            .as_ref()
            .is_some_and(|(at, _)| at.elapsed() < EXTERNAL_DETECT_TTL);
        if !fresh {
            *guard = Some((
                std::time::Instant::now(),
                car_external_agents::detect().await,
            ));
        }
        guard.as_ref().map(|(_, s)| s.clone()).unwrap_or_default()
    };
    specs
        .into_iter()
        .filter_map(|spec| {
            // Adapter ids ("claude-code", "codex", "gemini") are charset-safe.
            let identifier = car_connectors::discovery::ServiceIdentifier::new(
                "external",
                [String::from("agent")],
                spec.id.clone(),
            )
            .ok()?;
            Some(DiscoveredService {
                identifier: identifier.to_string(),
                capability_text: external_capability_text(&spec),
                name: spec.display_name,
                kind: "external".to_string(),
                protocol: "cli".to_string(),
                agent_id: None,
                endpoint: None,
            })
        })
        .collect()
}

/// Per-peer A2A agent-card fetch timeout — a slow/unreachable peer is skipped.
const A2A_CARD_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
/// TTL for a cached peer card — peer skills change rarely, and re-fetching every
/// registered peer's card on every resolve would hammer them with HTTP.
const A2A_CARD_TTL: std::time::Duration = std::time::Duration::from_secs(60);

/// Process-global TTL cache of fetched A2A peer cards, keyed by peer URL.
fn a2a_card_cache() -> &'static tokio::sync::Mutex<
    std::collections::HashMap<String, (std::time::Instant, car_a2a::AgentCard)>,
> {
    static CACHE: std::sync::OnceLock<
        tokio::sync::Mutex<
            std::collections::HashMap<String, (std::time::Instant, car_a2a::AgentCard)>,
        >,
    > = std::sync::OnceLock::new();
    CACHE.get_or_init(|| tokio::sync::Mutex::new(std::collections::HashMap::new()))
}

/// Fetch a peer's agent card, TTL-cached. None on timeout/unreachable/error —
/// the lock is never held across the network fetch.
async fn peer_card_cached(url: &str) -> Option<car_a2a::AgentCard> {
    if let Some((at, card)) = a2a_card_cache().lock().await.get(url) {
        if at.elapsed() < A2A_CARD_TTL {
            return Some(card.clone());
        }
    }
    let fetched = tokio::time::timeout(
        A2A_CARD_TIMEOUT,
        car_a2a::A2aClient::new(url.to_string()).agent_card(),
    )
    .await;
    let card = match fetched {
        Ok(Ok(c)) => c,
        _ => return None,
    };
    a2a_card_cache()
        .lock()
        .await
        .insert(url.to_string(), (std::time::Instant::now(), card.clone()));
    Some(card)
}

/// Provider: skills advertised by registered remote A2A peers. Each peer's card
/// is fetched concurrently (per-peer timeout + TTL cache); an unreachable peer
/// is skipped. A skill becomes a service identified `agentdns://<slug>/skill/<id>`.
async fn a2a_peer_services() -> Vec<DiscoveredService> {
    let Ok(reg) = car_a2a::peers::PeerRegistry::user_default() else {
        return Vec::new();
    };
    let peers = reg.list();
    // Evict cached cards for peers that are no longer registered so the cache
    // stays bounded to the current peer set (it otherwise only grows).
    {
        let live: std::collections::HashSet<&str> = peers.iter().map(|p| p.url.as_str()).collect();
        a2a_card_cache()
            .lock()
            .await
            .retain(|url, _| live.contains(url.as_str()));
    }
    let fetched = futures::future::join_all(
        peers
            .into_iter()
            .map(|peer| async move { peer_card_cached(&peer.url).await.map(|card| (peer, card)) }),
    )
    .await;
    let mut out = Vec::new();
    for (peer, card) in fetched.into_iter().flatten() {
        for skill in card.skills {
            // Skill ids come from arbitrary peers; skip one that can't form an
            // identifier rather than fail the peer's other skills.
            let identifier = match car_connectors::discovery::ServiceIdentifier::new(
                peer.slug.clone(),
                [String::from("skill")],
                skill.id.clone(),
            ) {
                Ok(id) => id,
                Err(_) => {
                    tracing::debug!(
                        peer = %peer.slug,
                        skill = %skill.id,
                        "discovery: skipping a2a skill with non-identifier id"
                    );
                    continue;
                }
            };
            let capability_text = if skill.description.is_empty() {
                skill.name.clone()
            } else {
                format!("{}. {}", skill.name, skill.description)
            };
            out.push(DiscoveredService {
                identifier: identifier.to_string(),
                name: skill.name,
                kind: "a2a".to_string(),
                protocol: "a2a".to_string(),
                capability_text,
                agent_id: None,
                endpoint: None,
            });
        }
    }
    out
}

/// Env var that enables and points at the remote AgentDNS root server. Unset =
/// the remote provider is inactive (the cross-vendor backend isn't deployed
/// yet — see `docs/agentdns-root-contract.md`). Opt-in keeps discovery from
/// making outbound calls to a root nobody configured.
const AGENTDNS_ROOT_URL_ENV: &str = "CAR_AGENTDNS_ROOT_URL";

/// Hard cap on records accepted from a remote root before embedding — a
/// malicious/buggy root must not be able to blow up the embed batch (`limit` in
/// the request is advisory; the root controls the response).
const MAX_REMOTE_RECORDS: usize = 100;
/// Cap on a remote service's embedded capability text — bounds per-record cost.
const MAX_REMOTE_TEXT_CHARS: usize = 2000;

/// The Parslee API host (where the access token is minted) — the only host the
/// bearer may be sent to.
fn parslee_api_host() -> Option<String> {
    let base = std::env::var(crate::parslee_auth::API_BASE_KEY)
        .unwrap_or_else(|_| crate::parslee_auth::DEFAULT_API_BASE.to_string());
    reqwest::Url::parse(&base)
        .ok()
        .and_then(|u| u.host_str().map(str::to_string))
}

/// Whether a root URL is safe to send the Parslee bearer to: HTTPS **and** the
/// same host that minted the token (the Parslee API).
fn root_host_is_trusted(root_url: &str) -> bool {
    let Ok(url) = reqwest::Url::parse(root_url) else {
        return false;
    };
    url.scheme() == "https" && url.host_str() == parslee_api_host().as_deref()
}

/// The bearer to send to a root, only when [`root_host_is_trusted`]. A
/// third-party / cleartext root gets no token — the contract serves public
/// results unauthenticated — so a mis-set `CAR_AGENTDNS_ROOT_URL` can never
/// exfiltrate the Parslee credential.
async fn trusted_root_bearer(root_url: &str, _state: &Arc<ServerState>) -> Option<String> {
    if !root_host_is_trusted(root_url) {
        return None;
    }
    // Mint a freshly-refreshed bearer instead of the `parslee_session` OnceLock
    // token captured once at boot. That token expires ~1h into daemon uptime,
    // after which the remote root 401'd and `discovery.resolve` silently
    // dropped all remote-root services until restart (#317).
    car_auth::access_token_refreshing().await
}

fn truncate_chars(s: &str, max: usize) -> String {
    s.chars().take(max).collect()
}

/// Provider: a remote AgentDNS root server's cross-vendor registry. Gated on
/// `CAR_AGENTDNS_ROOT_URL`; sends the Parslee bearer only to the trusted Parslee
/// host (see [`trusted_root_bearer`]). Records are folded into local ranking via
/// their `description` (the root's own ordering is advisory), capped in count
/// and length. Best-effort: any error yields no remote services.
async fn remote_root_services(
    state: &Arc<ServerState>,
    need: &str,
    limit: usize,
) -> Vec<DiscoveredService> {
    let Some(base) = std::env::var_os(AGENTDNS_ROOT_URL_ENV) else {
        return Vec::new();
    };
    let base = base.to_string_lossy().into_owned();
    let token = trusted_root_bearer(&base, state).await;
    let root = car_connectors::discovery::RemoteRoot::new(base, token);
    let records = match root.resolve(need, limit).await {
        Ok(r) => r,
        Err(e) => {
            tracing::warn!(error = %e, "discovery.resolve: remote root resolve failed; skipping");
            return Vec::new();
        }
    };
    records
        .into_iter()
        .take(MAX_REMOTE_RECORDS)
        .filter_map(|rec| {
            // Validate the root-provided identifier; drop a malformed one rather
            // than surface an unparseable name.
            let identifier = car_connectors::discovery::ServiceIdentifier::parse(&rec.identifier)
                .ok()?
                .to_string();
            let raw = if rec.description.is_empty() {
                rec.name.clone()
            } else {
                format!("{}. {}", rec.name, rec.description)
            };
            Some(DiscoveredService {
                identifier,
                name: truncate_chars(&rec.name, MAX_REMOTE_TEXT_CHARS),
                kind: truncate_chars(&rec.kind, 64),
                protocol: truncate_chars(&rec.protocol, 64),
                capability_text: truncate_chars(&raw, MAX_REMOTE_TEXT_CHARS),
                agent_id: None,
                endpoint: None,
            })
        })
        .collect()
}

/// Score a discovered service against the need embedding. EVERY provider kind
/// carries a learned success prior — the unified Beta(success+1, fail+1)
/// posterior over the routing-store history keyed by the service's
/// `agentdns://` identifier (fed by `discovery.report`), which for a
/// declarative agent also folds the history under its agent id (fed by
/// `declagents.route`/`invoke`) — the same [`posterior_success_prior`]
/// substrate `rank_agents` uses, so both surfaces score identically (H2
/// Part 2). Declarative agents additionally blend their learned capability
/// centroid; other kinds rank on cold-start similarity (their centroid never
/// learns — only declarative runs record capability vectors). Returns
/// `(score, similarity)`.
fn score_service(
    service: &DiscoveredService,
    need_emb: &[f32],
    cap_emb: &[f32],
    routing: &car_registry::routing::RoutingSnapshot,
) -> (f32, f32) {
    let coldstart = cosine(need_emb, cap_emb);
    let (learned, prior) = match &service.agent_id {
        Some(id) => (
            routing.learned_capability(id).map(|c| cosine(need_emb, c)),
            posterior_success_prior(routing, &[id, &service.identifier]),
        ),
        None => (
            None,
            posterior_success_prior(routing, &[&service.identifier]),
        ),
    };
    let similarity = blended_similarity(coldstart, learned);
    (route_score(similarity, prior, 0.0), similarity)
}

async fn embed_service_docs(
    state: &Arc<ServerState>,
    services: &[DiscoveredService],
) -> Result<Vec<Vec<f32>>, String> {
    let engine = crate::handler::get_inference_engine(state);
    let _permit = state.admission.acquire().await;
    let cap_embs = engine
        .embed(car_inference::EmbedRequest {
            texts: services.iter().map(|s| s.capability_text.clone()).collect(),
            model: None,
            instruction: None,
            is_query: false,
        })
        .await
        .map_err(|e| format!("embed failed: {e}"))?;
    drop(_permit);
    if cap_embs.len() != services.len() {
        return Err(format!(
            "embedder returned {} vectors for {} services",
            cap_embs.len(),
            services.len()
        ));
    }
    Ok(cap_embs)
}

fn rank_services(
    need_emb: &[f32],
    cap_embs: &[Vec<f32>],
    services: &[DiscoveredService],
    routing: &car_registry::routing::RoutingSnapshot,
) -> Vec<(usize, f32, f32)> {
    let mut ranked: Vec<(usize, f32, f32)> = cap_embs
        .iter()
        .enumerate()
        .map(|(i, e)| {
            let (score, similarity) = score_service(&services[i], need_emb, e, routing);
            (i, score, similarity)
        })
        .collect();
    ranked.sort_by(|a, b| {
        b.1.total_cmp(&a.1)
            .then_with(|| services[a.0].identifier.cmp(&services[b.0].identifier))
    });
    ranked
}

fn build_service_hints(
    subtasks: &[String],
    sub_embs: &[Vec<f32>],
    cap_embs: &[Vec<f32>],
    services: &[DiscoveredService],
    routing: &car_registry::routing::RoutingSnapshot,
    limit: usize,
) -> Vec<String> {
    let mut hints = BTreeMap::new();
    for (i, _sub) in subtasks.iter().enumerate() {
        let Some(emb) = sub_embs.get(i) else {
            continue;
        };
        for (idx, ..) in rank_services(emb, cap_embs, services, routing)
            .into_iter()
            .take(limit)
        {
            let svc = &services[idx];
            hints.entry(svc.identifier.clone()).or_insert_with(|| {
                truncate_hint(&format!("{}: {}", svc.name, svc.capability_text), 180)
            });
            if hints.len() >= limit {
                break;
            }
        }
        if hints.len() >= limit {
            break;
        }
    }
    hints.into_values().collect()
}

async fn decompose_with_service_sad(
    state: &Arc<ServerState>,
    need: &str,
    max: usize,
    config: &SadConfig,
    services: &[DiscoveredService],
    cap_embs: &[Vec<f32>],
    routing: &car_registry::routing::RoutingSnapshot,
) -> Result<DecompositionTrace, String> {
    let initial = decompose_need(state, need, max).await;
    if config.mode == DecompositionMode::Vanilla {
        return Ok(DecompositionTrace {
            mode: config.mode,
            rounds: 1,
            initial_subtasks: initial.clone(),
            final_subtasks: initial,
            hints: Vec::new(),
            hint_jaccard: None,
        });
    }
    let mut current = initial.clone();
    let mut previous_hints: Option<Vec<String>> = None;
    let mut last_hints = Vec::new();
    let mut last_jaccard = None;
    let mut rounds = 1;
    for _ in 0..config.iterations {
        let sub_embs = embed_query_texts(
            state,
            current.clone(),
            "Match this need to the service best able to perform it",
        )
        .await?;
        let hints = build_service_hints(
            &current,
            &sub_embs,
            cap_embs,
            services,
            routing,
            config.hints,
        );
        if let Some(prev) = previous_hints.as_ref() {
            let j = hint_jaccard(prev, &hints);
            last_jaccard = Some(j);
            if j >= config.convergence_jaccard {
                last_hints = hints;
                break;
            }
        }
        current = decompose_need_with_hints(state, need, max, &hints).await;
        rounds += 1;
        previous_hints = Some(hints.clone());
        last_hints = hints;
    }
    Ok(DecompositionTrace {
        mode: config.mode,
        rounds,
        initial_subtasks: initial,
        final_subtasks: current,
        hints: last_hints,
        hint_jaccard: last_jaccard,
    })
}

fn invoke_kind_and_target(service: &DiscoveredService) -> (&'static str, String) {
    match service.kind.as_str() {
        "declarative" => (
            "declagents.invoke",
            service.agent_id.clone().unwrap_or_default(),
        ),
        "connector" => ("tool", service.name.clone()),
        "external" => (
            "agents.invoke_external",
            service
                .identifier
                .rsplit('/')
                .next()
                .unwrap_or(service.name.as_str())
                .to_string(),
        ),
        "a2a" => ("a2a_dispatch", service.identifier.clone()),
        // Registry services are plain HTTP endpoints (their dashboard URL); the
        // caller reaches them directly, not through a governed in-daemon surface.
        "registry" => (
            "http",
            service.endpoint.clone().unwrap_or_else(|| service.identifier.clone()),
        ),
        _ => ("manual", service.identifier.clone()),
    }
}

fn infer_plan_edges(subtasks: &[String]) -> Vec<Value> {
    let sequential_markers = [
        " then ",
        " after ",
        " next ",
        " before ",
        " transform",
        " convert",
        " summarize",
        " report",
        " visualize",
        " upload",
        " send",
    ];
    let mut edges = Vec::new();
    for i in 1..subtasks.len() {
        let prev = subtasks[i - 1].to_lowercase();
        let cur = subtasks[i].to_lowercase();
        let marker = sequential_markers
            .iter()
            .any(|m| cur.contains(m.trim()) || prev.contains(m.trim()));
        let overlap = prev
            .split(|c: char| !c.is_alphanumeric())
            .filter(|s| s.len() > 3)
            .any(|tok| cur.contains(tok));
        if marker || overlap || subtasks.len() <= 3 {
            edges.push(json!({
                "from": format!("step_{}", i),
                "to": format!("step_{}", i + 1),
                "reason": if marker { "sequence_marker" } else if overlap { "term_overlap" } else { "conservative_chain" },
            }));
        }
    }
    edges
}

async fn rerank_service_candidates(
    state: &Arc<ServerState>,
    subtask: &str,
    candidates: &[Value],
) -> Option<usize> {
    if candidates.len() < 2 {
        return None;
    }
    let mut lines = Vec::new();
    for (i, c) in candidates.iter().enumerate() {
        lines.push(format!(
            "{}. {} ({})",
            i,
            c.get("name").and_then(|v| v.as_str()).unwrap_or("?"),
            c.get("kind").and_then(|v| v.as_str()).unwrap_or("?")
        ));
    }
    let prompt = format!(
        "Choose the single best service for the subtask. Respond with JSON only: \
         {{\"index\": 0}} where index is zero-based.\n\nSubtask: {subtask}\n\nCandidates:\n{}",
        lines.join("\n")
    );
    let engine = crate::handler::get_inference_engine(state);
    let _permit = state.admission.acquire().await;
    let raw = engine
        .generate(car_inference::GenerateRequest {
            prompt,
            response_format: Some(car_inference::ResponseFormat::JsonObject),
            ..Default::default()
        })
        .await
        .ok()?;
    drop(_permit);
    let idx = serde_json::from_str::<Value>(&raw)
        .ok()
        .and_then(|v| v.get("index").and_then(|i| i.as_u64()))
        .map(|i| i as usize)?;
    (idx < candidates.len()).then_some(idx)
}

/// Resolve a need into ranked CAR-local services across providers (declarative
/// agents, observe-only registry services, connected MCP connector tools,
/// external CLIs, A2A peers, and an opt-in remote root), each named under the
/// `agentdns://` scheme. Returns `{ services: [{ identifier, name, kind, protocol, score,
/// similarity }], count }`. Pure resolution — it does not invoke anything; the
/// caller selects an identifier and invokes via the matching surface (e.g.
/// `declagents.invoke`, or the connector's canonical tool name). Empty
/// `services` (not an error) when nothing matches or nothing is registered.
pub async fn handle_discovery_resolve(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DiscoveryResolveParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    if params.need.trim().is_empty() {
        return Err("need must be a non-empty capability description".to_string());
    }
    let limit = params
        .limit
        .unwrap_or(DISCOVERY_DEFAULT_LIMIT)
        .clamp(1, DISCOVERY_MAX_LIMIT);

    let services = gather_discovered_services(state, &params.need, limit).await;
    if services.is_empty() {
        return Ok(json!({ "services": [], "count": 0 }));
    }

    let engine = crate::handler::get_inference_engine(state);
    let _permit = state.admission.acquire().await;
    let need_embs = engine
        .embed(car_inference::EmbedRequest {
            texts: vec![params.need.clone()],
            model: None,
            instruction: Some("Match this need to the service best able to perform it".to_string()),
            is_query: true,
        })
        .await
        .map_err(|e| format!("embed failed: {e}"))?;
    drop(_permit);

    let need_emb = need_embs
        .first()
        .ok_or_else(|| "embedder returned no vectors".to_string())?;
    let cap_embs = embed_service_docs(state, &services).await?;
    let routing = state.routing().map(|s| s.snapshot()).unwrap_or_default();

    let ranked = rank_services(need_emb, &cap_embs, &services, &routing);

    let out: Vec<Value> = ranked
        .iter()
        .take(limit)
        .map(|(i, score, similarity)| {
            let s = &services[*i];
            json!({
                "identifier": s.identifier,
                "name": s.name,
                "kind": s.kind,
                "protocol": s.protocol,
                "score": score,
                "similarity": similarity,
            })
        })
        .collect();

    Ok(json!({ "count": out.len(), "services": out }))
}

#[derive(Deserialize)]
struct DiscoveryReportParams {
    /// The `agentdns://…` identifier the outcome is recorded against.
    identifier: String,
    /// `"success"` or `"failure"`.
    outcome: String,
}

/// Parse a `discovery.report` outcome string. Strict — an unknown outcome is
/// an error, not a silent failure-record.
fn parse_report_outcome(outcome: &str) -> Result<bool, String> {
    match outcome {
        "success" => Ok(true),
        "failure" => Ok(false),
        other => Err(format!(
            "outcome must be \"success\" or \"failure\", got \"{other}\""
        )),
    }
}

/// Record a discovery-routed run's outcome into the routing store, keyed by
/// the service's `agentdns://` identifier — for ANY provider kind (connector,
/// registry, external, a2a, declarative). This is the H2 Part 2 feedback
/// surface: it closes the loop `discovery.resolve` learns from, so a failing
/// MCP-connector tool (say) is demoted below a healthy sibling on the next
/// resolve instead of sitting at the neutral prior forever. The identifier is
/// validated against the `agentdns://` scheme — pass it VERBATIM from
/// `discovery.resolve`: the parser validates charset/shape but does not
/// normalize (no lowercasing), so a re-spelled identifier records dead
/// feedback ranking never reads, and any charset-valid identifier is
/// persisted whether or not the service exists (unknown keys never rank,
/// but they do occupy the store). For a
/// declarative agent the identifier-keyed counts are folded together with its
/// agent-id-keyed counts at ranking time ([`posterior_success_prior`]), so
/// both feedback paths teach the same posterior. Returns the updated raw
/// counts: `{ identifier, outcome, successes, failures }`.
pub async fn handle_discovery_report(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DiscoveryReportParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let ok = parse_report_outcome(&params.outcome)?;
    let identifier = car_connectors::discovery::ServiceIdentifier::parse(&params.identifier)
        .map_err(|e| format!("invalid identifier: {e}"))?
        .to_string();
    // In-daemon declarative invocations ALREADY self-record under the
    // agent id (declagents.invoke / route with invoke / route_split), and
    // ranking folds the agent-id and identifier keys together — so a
    // discovery.report against a local declarative agent would teach the
    // same run twice, inflating its evidence weight (review follow-up).
    // Reject with the pointer to the surface that already recorded it.
    if identifier.starts_with("agentdns://local/agent/") {
        return Err(format!(
            "'{identifier}' is an in-daemon declarative agent: its runs are              recorded automatically by declagents.invoke/route — reporting              them again would double-count the outcome. discovery.report is              for the provider kinds that can't self-record (connector,              registry, external, a2a)."
        ));
    }
    let store = state.routing()?;
    store.record_outcome(&identifier, ok)?;
    let (successes, failures) = store.snapshot().outcome_counts(&identifier);
    Ok(json!({
        "identifier": identifier,
        "outcome": params.outcome,
        "successes": successes,
        "failures": failures,
    }))
}

#[derive(Deserialize)]
struct DiscoveryRouteComposeParams {
    need: String,
    #[serde(default)]
    max_subtasks: Option<usize>,
    #[serde(default)]
    decomposition_mode: DecompositionMode,
    #[serde(default)]
    sad_hints: Option<usize>,
    #[serde(default)]
    sad_iterations: Option<usize>,
    #[serde(default)]
    sad_convergence_jaccard: Option<f64>,
    #[serde(default)]
    candidates_per_step: Option<usize>,
    #[serde(default)]
    rerank: bool,
}

/// Compose a cross-service route plan over the same providers as
/// `discovery.resolve`. This plans only; cross-kind invocation remains explicit
/// so connector/A2A/external services stay on their existing governed paths.
pub async fn handle_discovery_route_compose(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: DiscoveryRouteComposeParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    if params.need.trim().is_empty() {
        return Err("need must be a non-empty capability description".to_string());
    }
    let max = params
        .max_subtasks
        .unwrap_or(DEFAULT_MAX_SUBTASKS)
        .clamp(1, MAX_SUBTASKS_CAP);
    let candidates_per_step = params
        .candidates_per_step
        .unwrap_or(DEFAULT_CANDIDATES_PER_STEP)
        .clamp(1, MAX_CANDIDATES_PER_STEP);
    let sad = SadConfig::new(
        params.decomposition_mode,
        params.sad_hints,
        params.sad_iterations,
        params.sad_convergence_jaccard,
    );

    let services = gather_discovered_services(state, &params.need, candidates_per_step).await;
    if services.is_empty() {
        return Ok(json!({
            "plan": { "steps": [], "edges": [] },
            "decomposition": {
                "decomposition_mode": sad.mode,
                "rounds": 0,
                "initial_subtasks": [],
                "final_subtasks": [],
                "hints": [],
                "hint_jaccard": null,
            },
            "candidates": [],
            "metadata": { "service_count": 0, "candidates_per_step": candidates_per_step, "rerank": params.rerank },
        }));
    }
    let cap_embs = embed_service_docs(state, &services).await?;
    let routing = state.routing().map(|s| s.snapshot()).unwrap_or_default();
    let decomposition = decompose_with_service_sad(
        state,
        &params.need,
        max,
        &sad,
        &services,
        &cap_embs,
        &routing,
    )
    .await?;
    let subtasks = decomposition.final_subtasks.clone();
    let sub_embs = embed_query_texts(
        state,
        subtasks.clone(),
        "Match this need to the service best able to perform it",
    )
    .await?;

    let mut steps = Vec::new();
    let mut all_candidates = Vec::new();
    for (i, subtask) in subtasks.iter().enumerate() {
        let Some(emb) = sub_embs.get(i) else {
            continue;
        };
        let ranked = rank_services(emb, &cap_embs, &services, &routing);
        let mut candidates: Vec<Value> = ranked
            .iter()
            .take(candidates_per_step)
            .map(|(idx, score, similarity)| {
                let svc = &services[*idx];
                let (invoke_kind, invoke_target) = invoke_kind_and_target(svc);
                json!({
                    "identifier": svc.identifier,
                    "name": svc.name,
                    "kind": svc.kind,
                    "protocol": svc.protocol,
                    "score": score,
                    "similarity": similarity,
                    "invoke_kind": invoke_kind,
                    "invoke_target": invoke_target,
                })
            })
            .collect();
        if params.rerank {
            if let Some(best) = rerank_service_candidates(state, subtask, &candidates).await {
                candidates.swap(0, best);
            }
        }
        let chosen = candidates.first().cloned().unwrap_or_else(|| json!({}));
        let invoke_kind = chosen.get("invoke_kind").cloned().unwrap_or(Value::Null);
        let invoke_target = chosen.get("invoke_target").cloned().unwrap_or(Value::Null);
        steps.push(json!({
            "id": format!("step_{}", i + 1),
            "subtask": subtask,
            "service": chosen,
            "invoke_kind": invoke_kind,
            "invoke_target": invoke_target,
        }));
        all_candidates.push(json!({
            "step_id": format!("step_{}", i + 1),
            "subtask": subtask,
            "candidates": candidates,
        }));
    }

    Ok(json!({
        "plan": {
            "steps": steps,
            "edges": infer_plan_edges(&subtasks),
        },
        "decomposition": {
            "decomposition_mode": decomposition.mode,
            "rounds": decomposition.rounds,
            "initial_subtasks": decomposition.initial_subtasks,
            "final_subtasks": decomposition.final_subtasks,
            "hints": decomposition.hints,
            "hint_jaccard": decomposition.hint_jaccard,
        },
        "candidates": all_candidates,
        "metadata": {
            "service_count": services.len(),
            "candidates_per_step": candidates_per_step,
            "rerank": params.rerank,
            "auto_invoked": false,
        },
    }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::coder::native_loop::TurnGenerator;
    use async_trait::async_trait;
    use car_inference::{GenerateRequest, InferenceResult};
    use std::sync::atomic::{AtomicUsize, Ordering};

    fn spec(
        id: &str,
        identity: &str,
        tools: &[&str],
    ) -> car_registry::declarative::DeclarativeAgentSpec {
        car_registry::declarative::DeclarativeAgentSpec {
            id: id.to_string(),
            name: id.to_string(),
            identity: identity.to_string(),
            tools: tools.iter().map(|t| t.to_string()).collect(),
            denied_tools: vec![],
            standing_goal: String::new(),
            scenarios: vec![],
            enabled: true,
        }
    }

    #[test]
    fn registry_service_composes_capability_and_endpoint() {
        let entry = car_registry::AgentEntry::new("fms-feasibility", "http://127.0.0.1:8132")
            .with_display_name("FMS Feasibility")
            .with_capability("checks whether a flight trip is feasible for the fleet")
            .with_status(car_registry::AgentStatus::Running);
        let svc = registry_entry_to_service(entry).expect("running entry is routable");
        assert_eq!(svc.identifier, "agentdns://local/service/fms-feasibility");
        assert_eq!(svc.kind, "registry");
        assert_eq!(svc.protocol, "http");
        assert_eq!(svc.name, "FMS Feasibility");
        assert_eq!(svc.endpoint.as_deref(), Some("http://127.0.0.1:8132"));
        // Label + capability fold into the embed doc that drives ranking.
        assert_eq!(
            svc.capability_text,
            "FMS Feasibility. checks whether a flight trip is feasible for the fleet"
        );
        // Plans route to the dashboard URL over plain HTTP.
        assert_eq!(invoke_kind_and_target(&svc), ("http", "http://127.0.0.1:8132".to_string()));
    }

    #[test]
    fn registry_service_without_capability_falls_back_to_label() {
        let entry = car_registry::AgentEntry::new("trader", "http://127.0.0.1:9101")
            .with_status(car_registry::AgentStatus::Idle);
        let svc = registry_entry_to_service(entry).expect("idle entry is routable");
        // No display_name, no capability → bare name carries ranking.
        assert_eq!(svc.name, "trader");
        assert_eq!(svc.capability_text, "trader");
    }

    #[test]
    fn registry_entry_freshness_tracks_heartbeat_age() {
        let mut entry = car_registry::AgentEntry::new("svc", "http://x");
        entry.last_heartbeat_at = 1_000;
        // Within the staleness window → routable.
        assert!(registry_entry_is_fresh(&entry, 1_000 + REGISTRY_STALE_AFTER_SECS));
        // One second past the window → a crashed-but-unreaped entry is hidden.
        assert!(!registry_entry_is_fresh(
            &entry,
            1_000 + REGISTRY_STALE_AFTER_SECS + 1
        ));
        // Clock-read failure (now = 0) fails open rather than blanking discovery.
        assert!(registry_entry_is_fresh(&entry, 0));
    }

    #[test]
    fn registry_service_skips_non_routable_status() {
        for status in [
            car_registry::AgentStatus::Stopping,
            car_registry::AgentStatus::Errored,
        ] {
            let entry = car_registry::AgentEntry::new("gone", "http://x").with_status(status);
            assert!(
                registry_entry_to_service(entry).is_none(),
                "{status:?} must not be surfaced as routable"
            );
        }
    }

    #[test]
    fn cosine_is_one_for_identical_and_zero_for_orthogonal() {
        let a = [1.0, 2.0, 3.0];
        assert!((cosine(&a, &a) - 1.0).abs() < 1e-6);
        assert!((cosine(&[1.0, 0.0], &[0.0, 1.0])).abs() < 1e-6);
    }

    #[test]
    fn cosine_zero_norm_is_zero_not_nan() {
        let z = cosine(&[0.0, 0.0], &[1.0, 2.0]);
        assert_eq!(z, 0.0);
        assert!(!z.is_nan());
    }

    #[test]
    fn capability_text_includes_identity_goal_and_tools() {
        let mut s = spec("billing", "Handles invoices.", &["fetch", "parse"]);
        s.standing_goal = "Keep ledgers reconciled".to_string();
        let text = capability_text(&s);
        assert!(text.contains("Handles invoices."));
        assert!(text.contains("Keep ledgers reconciled"));
        assert!(text.contains("fetch, parse"));
    }

    #[test]
    fn blended_score_keeps_similarity_dominant() {
        // Strong match with no track record still beats a weak match with a
        // perfect record — similarity carries the 0.7 weight.
        let strong_unproven = blended_score(0.9, 0.5);
        let weak_proven = blended_score(0.2, 1.0);
        assert!(strong_unproven > weak_proven);
    }

    #[test]
    fn blended_score_prior_breaks_ties() {
        // Equal similarity: the agent that actually succeeds ranks higher.
        assert!(blended_score(0.8, 1.0) > blended_score(0.8, 0.5));
    }

    #[test]
    fn blended_score_clamps_negative_similarity() {
        // Anti-correlated similarity is clamped to 0; only the prior term remains.
        let s = blended_score(-0.5, 0.5);
        assert!((s - (1.0 - ROUTE_SIMILARITY_WEIGHT) * 0.5).abs() < 1e-6);
    }

    fn run(
        turns: u32,
        output: &str,
        error: Option<&str>,
    ) -> super::super::declarative::AgentRunResult {
        super::super::declarative::AgentRunResult {
            output: output.to_string(),
            turns,
            tool_calls: 0,
            error: error.map(|s| s.to_string()),
        }
    }

    #[test]
    fn infra_noise_runs_are_not_recorded() {
        // Errored before any turn → infra noise, don't teach the prior.
        assert!(!run_is_recordable(&run(0, "", Some("model load failed"))));
        // Errored after real work → a genuine agent failure, do record it.
        assert!(run_is_recordable(&run(3, "", Some("gave up"))));
        // Clean completion → record it.
        assert!(run_is_recordable(&run(2, "done", None)));
    }

    #[test]
    fn run_succeeded_requires_no_error_and_nonempty_output() {
        assert!(run_succeeded(&run(2, "hello", None)));
        assert!(!run_succeeded(&run(2, "   ", None))); // whitespace-only
        assert!(!run_succeeded(&run(2, "hello", Some("boom"))));
    }

    fn svc(kind: &'static str, agent_id: Option<&str>) -> DiscoveredService {
        DiscoveredService {
            identifier: format!("agentdns://x/{kind}/y"),
            name: "y".into(),
            kind: kind.to_string(),
            protocol: "p".to_string(),
            capability_text: "y".into(),
            agent_id: agent_id.map(|s| s.to_string()),
            endpoint: None,
        }
    }

    #[test]
    fn external_capability_text_lists_enabled_features() {
        let spec = car_external_agents::ExternalAgentSpec {
            id: "claude-code".into(),
            display_name: "Claude Code".into(),
            binary_path: "/usr/local/bin/claude".into(),
            version: None,
            auth_kind: Default::default(),
            capabilities: car_external_agents::Capabilities {
                tool_use: true,
                mcp: true,
                hooks: false,
                sessions: true,
                streaming: false,
                images: false,
            },
            detected_at: 0,
            health: None,
        };
        let text = external_capability_text(&spec);
        assert!(text.contains("Claude Code"));
        assert!(text.contains("tool use, MCP, sessions")); // only enabled, in order
        assert!(!text.contains("hooks"));
    }

    #[test]
    fn bearer_only_to_trusted_parslee_https_host() {
        // Default Parslee host (api.parslee.ai) when PARSLEE_API_BASE is unset.
        assert!(root_host_is_trusted(
            "https://api.parslee.ai/agentdns/resolve"
        ));
        // Cleartext to the right host: refused (no token over http).
        assert!(!root_host_is_trusted("http://api.parslee.ai"));
        // HTTPS to a different host: refused (no token to a third party).
        assert!(!root_host_is_trusted(
            "https://attacker.example/agentdns/resolve"
        ));
        // Garbage URL: refused.
        assert!(!root_host_is_trusted("not a url"));
    }

    #[test]
    fn truncate_chars_is_char_boundary_safe() {
        assert_eq!(truncate_chars("hello", 3), "hel");
        assert_eq!(truncate_chars("hello", 10), "hello");
        // Multi-byte chars truncated by count, not bytes (no panic).
        assert_eq!(truncate_chars("héllo", 2), "hé");
    }

    #[test]
    fn score_service_uses_neutral_prior_for_non_declarative() {
        let routing = car_registry::routing::RoutingSnapshot::default();
        let s = svc("connector", None);
        // identical need/cap ⇒ cosine 1.0; score = 0.7*1 + 0.3*0.5 = 0.85.
        let (score, sim) = score_service(&s, &[1.0, 0.0], &[1.0, 0.0], &routing);
        assert!((sim - 1.0).abs() < 1e-6);
        assert!((score - 0.85).abs() < 1e-6);
    }

    #[test]
    fn score_service_blends_learning_for_proven_declarative() {
        let mut routing = car_registry::routing::RoutingSnapshot::default();
        routing.agents.insert(
            "a".into(),
            car_registry::routing::AgentStats {
                successes: 4,
                failures: 0,
                ema_success_rate: 1.0,
                learned_vector: vec![],
            },
        );
        let s = svc("declarative", Some("a"));
        // cosine 1.0; prior is the Beta(4+1, 0+1) posterior mean 5/6 ⇒
        // 0.7*1 + 0.3*(5/6) = 0.95, above the 0.85 a history-less service
        // would score — and NOT the EMA's 1.0 (the EMA no longer ranks).
        let (score, _) = score_service(&s, &[1.0, 0.0], &[1.0, 0.0], &routing);
        assert!((score - (0.7 + 0.3 * (5.0 / 6.0))).abs() < 1e-6);
    }

    #[test]
    fn score_service_learns_for_non_declarative_via_identifier_key() {
        // THE point of H2 Part 2: a non-declarative service's history —
        // recorded by `discovery.report` under its agentdns identifier —
        // moves its prior off neutral.
        let mut routing = car_registry::routing::RoutingSnapshot::default();
        let s = svc("connector", None);
        routing.agents.insert(
            s.identifier.clone(),
            car_registry::routing::AgentStats {
                successes: 1,
                failures: 14,
                ema_success_rate: 0.9, // deliberately wrong-way EMA: must not rank
                learned_vector: vec![],
            },
        );
        let (score, _) = score_service(&s, &[1.0, 0.0], &[1.0, 0.0], &routing);
        // Beta(2, 15) mean = 2/17 ⇒ 0.7 + 0.3*(2/17) ≈ 0.7353 — demoted well
        // below the 0.85 a neutral sibling scores, EMA notwithstanding.
        assert!((score - (0.7 + 0.3 * (2.0 / 17.0))).abs() < 1e-6);
    }

    #[test]
    fn declarative_prior_merges_agent_id_and_identifier_keys() {
        // One agent, one score: outcomes recorded under the agent id
        // (declagents.route) and under the discovery identifier
        // (discovery.report) fold into a single posterior.
        let mut routing = car_registry::routing::RoutingSnapshot::default();
        let stats = |s: u64, f: u64| car_registry::routing::AgentStats {
            successes: s,
            failures: f,
            ema_success_rate: 0.0,
            learned_vector: vec![],
        };
        routing.agents.insert("a".into(), stats(3, 0));
        routing
            .agents
            .insert("agentdns://local/agent/a".into(), stats(2, 1));
        let merged = declarative_success_prior(&routing, "a");
        // Beta(5+1, 1+1) mean = 6/8.
        assert!((merged - 6.0 / 8.0).abs() < 1e-6);
        // And score_service sees the identical prior for the same agent.
        let s = DiscoveredService {
            identifier: "agentdns://local/agent/a".into(),
            name: "a".into(),
            kind: "declarative".into(),
            protocol: "in-daemon".into(),
            capability_text: "a".into(),
            agent_id: Some("a".into()),
            endpoint: None,
        };
        let (score, _) = score_service(&s, &[1.0, 0.0], &[1.0, 0.0], &routing);
        assert!((score - (0.7 + 0.3 * merged)).abs() < 1e-6);
    }

    #[test]
    fn parse_report_outcome_is_strict() {
        assert_eq!(parse_report_outcome("success"), Ok(true));
        assert_eq!(parse_report_outcome("failure"), Ok(false));
        assert!(parse_report_outcome("ok").is_err());
        assert!(parse_report_outcome("").is_err());
    }

    #[test]
    fn parse_subtasks_extracts_clean_list() {
        let raw = r#"{"subtasks": ["book flight", "  reserve hotel  ", "", "rent car"]}"#;
        let subs = parse_subtasks(raw, "trip", 5);
        assert_eq!(subs, vec!["book flight", "reserve hotel", "rent car"]); // trimmed, empties dropped
    }

    #[test]
    fn parse_subtasks_caps_at_max() {
        let raw = r#"{"subtasks": ["a","b","c","d"]}"#;
        assert_eq!(parse_subtasks(raw, "x", 2), vec!["a", "b"]);
    }

    #[test]
    fn parse_subtasks_falls_back_to_need() {
        // Malformed, missing key, and all-empty all degrade to [need].
        assert_eq!(parse_subtasks("not json", "do it", 5), vec!["do it"]);
        assert_eq!(
            parse_subtasks(r#"{"other": []}"#, "do it", 5),
            vec!["do it"]
        );
        assert_eq!(
            parse_subtasks(r#"{"subtasks": ["  "]}"#, "do it", 5),
            vec!["do it"]
        );
    }

    #[test]
    fn sad_prompt_includes_hints_and_json_only_contract() {
        let hints = vec![
            "chart-gen: create charts".to_string(),
            "csv-parser".to_string(),
        ];
        let prompt = decomposition_prompt("download and chart a csv", 4, &hints);
        assert!(prompt.contains("Available skills that may be relevant"));
        assert!(prompt.contains("chart-gen"));
        assert!(prompt.contains("Respond with JSON only"));
        assert!(prompt.contains(r#"{"subtasks""#));
    }

    #[test]
    fn hint_jaccard_detects_convergence() {
        let a = vec!["a".to_string(), "b".to_string(), "c".to_string()];
        let b = vec!["b".to_string(), "c".to_string(), "d".to_string()];
        let j = hint_jaccard(&a, &b);
        assert!((j - 0.5).abs() < 1e-6);
        assert_eq!(hint_jaccard(&[], &[]), 1.0);
    }

    #[test]
    fn service_hints_are_deduped_and_sorted() {
        let hints = build_service_hints(
            &["make chart".into()],
            &[vec![1.0, 0.0]],
            &[vec![0.0, 1.0], vec![1.0, 0.0], vec![1.0, 0.0]],
            &[
                svc("connector", None),
                DiscoveredService {
                    identifier: "agentdns://b/tool/chart".into(),
                    name: "chart".into(),
                    kind: "connector".into(),
                    protocol: "mcp".into(),
                    capability_text: "chart".into(),
                    agent_id: None,
                    endpoint: None,
                },
                DiscoveredService {
                    identifier: "agentdns://a/tool/chart".into(),
                    name: "chart duplicate".into(),
                    kind: "connector".into(),
                    protocol: "mcp".into(),
                    capability_text: "chart duplicate".into(),
                    agent_id: None,
                    endpoint: None,
                },
            ],
            &car_registry::routing::RoutingSnapshot::default(),
            2,
        );
        assert_eq!(hints.len(), 2);
        assert!(hints[0].contains("chart duplicate"));
        assert!(hints[1].contains("chart"));
    }

    #[test]
    fn dag_edges_chain_obvious_workflows() {
        let edges = infer_plan_edges(&[
            "download dataset".into(),
            "transform dataset".into(),
            "create report".into(),
        ]);
        assert_eq!(edges.len(), 2);
        assert_eq!(edges[0]["from"], "step_1");
        assert_eq!(edges[0]["to"], "step_2");
    }

    #[test]
    fn service_invoke_metadata_is_non_invoking_target() {
        let declarative = DiscoveredService {
            identifier: "agentdns://local/agent/a".into(),
            name: "Agent A".into(),
            kind: "declarative".into(),
            protocol: "in-daemon".into(),
            capability_text: "Agent A".into(),
            agent_id: Some("a".into()),
            endpoint: None,
        };
        assert_eq!(
            invoke_kind_and_target(&declarative),
            ("declagents.invoke", "a".into())
        );
        let connector = svc("connector", None);
        assert_eq!(invoke_kind_and_target(&connector).0, "tool");
        let external = DiscoveredService {
            identifier: "agentdns://external/agent/codex".into(),
            name: "Codex".into(),
            kind: "external".into(),
            protocol: "cli".into(),
            capability_text: "Codex".into(),
            agent_id: None,
            endpoint: None,
        };
        assert_eq!(
            invoke_kind_and_target(&external),
            ("agents.invoke_external", "codex".into())
        );
    }

    #[test]
    fn focused_fixture_eval_metrics_are_computable() {
        struct Fixture {
            predicted: usize,
            expected: usize,
            top3_hit: bool,
        }
        let fixtures = [
            Fixture {
                predicted: 3,
                expected: 3,
                top3_hit: true,
            },
            Fixture {
                predicted: 4,
                expected: 3,
                top3_hit: true,
            },
            Fixture {
                predicted: 1,
                expected: 3,
                top3_hit: false,
            },
        ];
        let exact = fixtures
            .iter()
            .filter(|f| f.predicted == f.expected)
            .count();
        let relaxed = fixtures
            .iter()
            .filter(|f| f.predicted.abs_diff(f.expected) <= 1)
            .count();
        let top3 = fixtures.iter().filter(|f| f.top3_hit).count();
        assert_eq!(exact, 1);
        assert_eq!(relaxed, 2);
        assert_eq!(top3, 2);
    }

    #[test]
    fn blended_similarity_falls_back_to_coldstart_without_centroid() {
        // No learned vector → pure cold-start.
        assert_eq!(blended_similarity(0.6, None), 0.6);
        // With a learned vector → 0.6*coldstart + 0.4*learned.
        let b = blended_similarity(0.5, Some(1.0));
        assert!((b - (0.6 * 0.5 + 0.4 * 1.0)).abs() < 1e-6);
    }

    #[test]
    fn route_score_edge_boost_promotes_forward_target() {
        // Two peers tie on similarity + prior; the one the delegator has a
        // learned forward edge to ranks higher.
        let plain = route_score(0.6, 0.5, 0.0);
        let forwarded = route_score(0.6, 0.5, 0.9);
        assert!(forwarded > plain);
    }

    #[test]
    fn excludes_delegator_and_visited_path() {
        let visited = vec!["a".to_string(), "b".to_string()];
        assert!(is_excluded("self", Some("self"), &[])); // can't route to itself
        assert!(is_excluded("a", None, &visited)); // already on the path
        assert!(is_excluded("b", Some("self"), &visited));
        assert!(!is_excluded("c", Some("self"), &visited)); // fresh peer is eligible
    }

    #[test]
    fn ranking_prefers_higher_cosine() {
        // Stand-in embeddings: the need points along the first axis; agent A is
        // aligned with it, agent B is orthogonal. A must rank first.
        let need = [1.0_f32, 0.0];
        let agent_embs = [[0.9_f32, 0.1], [0.0, 1.0]];
        let mut ranked: Vec<(usize, f32)> = agent_embs
            .iter()
            .enumerate()
            .map(|(i, e)| (i, cosine(&need, e)))
            .collect();
        ranked.sort_by(|a, b| b.1.total_cmp(&a.1));
        assert_eq!(ranked[0].0, 0);
    }

    struct Script {
        turns: Vec<InferenceResult>,
        cursor: AtomicUsize,
    }

    fn turn(text: &str, tool_calls: Value) -> InferenceResult {
        serde_json::from_value(json!({
            "text": text,
            "tool_calls": tool_calls,
            "trace_id": "t",
            "model_used": "scripted",
            "latency_ms": 0,
        }))
        .expect("scripted InferenceResult shape")
    }

    #[async_trait]
    impl TurnGenerator for Script {
        async fn generate(&self, _req: GenerateRequest) -> Result<InferenceResult, String> {
            let i = self.cursor.fetch_add(1, Ordering::SeqCst);
            self.turns
                .get(i)
                .cloned()
                .ok_or_else(|| "script exhausted".to_string())
        }
    }

    fn init_repo(dir: &Path) {
        for args in [
            vec!["init", "-q", "-b", "main"],
            vec![
                "-c",
                "user.name=t",
                "-c",
                "user.email=t@t",
                "commit",
                "-q",
                "--allow-empty",
                "-m",
                "init",
            ],
        ] {
            let out = std::process::Command::new("git")
                .arg("-C")
                .arg(dir)
                .args(&args)
                .output()
                .unwrap();
            assert!(
                out.status.success(),
                "{}",
                String::from_utf8_lossy(&out.stderr)
            );
        }
    }

    /// End-to-end smoke (plan §Tests): start → confirm → scripted native loop
    /// writes the file → contract green → DiffReady → approve → branch in the
    /// user's repo, user checkout untouched.
    #[tokio::test]
    async fn e2e_start_confirm_run_approve() {
        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        // Script: (1) contract derivation, (2) write_file, (3) done.
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "x.txt contains hello",
                        "checks": [{"name": "content", "command": "grep -q hello x.txt"}]}"#,
                    json!([]),
                ),
                turn(
                    "",
                    json!([{
                        "id": "c1", "name": "write_file",
                        "arguments": {"path": "x.txt", "content": "hello from the coder"}
                    }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });

        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "create x.txt containing hello".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(4),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();

        let session_id = response["session_id"].as_str().unwrap().to_string();
        assert_eq!(response["state"], "contract_proposed");
        assert_eq!(response["contract"]["checks"][0]["name"], "content");

        confirm_session(&state, &session_id, None).await.unwrap();

        // Wait for the loop task to finish.
        let entry = get_entry(&state, &session_id).await.unwrap();
        let handle = entry.task.lock().unwrap().take().unwrap();
        handle.await.unwrap();

        // State + event stream assertions.
        {
            let session = entry.session.lock().await;
            assert_eq!(
                session.state,
                CoderState::NeedsApproval,
                "error: {:?}",
                session.error
            );
            assert!(session.last_check_results.iter().all(|r| r.passed));
        }
        let events = entry.events.lock().await;
        let has = |pred: &dyn Fn(&CoderEventKind) -> bool| events.iter().any(|e| pred(&e.kind));
        assert!(has(&|k| matches!(k, CoderEventKind::EngineSelected { .. })));
        assert!(has(&|k| matches!(
            k,
            CoderEventKind::ContractProposed { .. }
        )));
        assert!(has(
            &|k| matches!(k, CoderEventKind::ToolCall { tool, .. } if tool == "write_file")
        ));
        assert!(has(
            &|k| matches!(k, CoderEventKind::CheckCompleted { result } if result.passed)
        ));
        assert!(has(
            &|k| matches!(k, CoderEventKind::DiffReady { stat, .. } if stat.contains("x.txt"))
        ));
        drop(events);

        // Approve → branch lands in the user's repo; checkout untouched.
        let merged = approve_merge_session(&state, &session_id, true)
            .await
            .unwrap();
        assert_eq!(merged["state"], "merged");
        let branch = merged["branch"].as_str().unwrap();
        let show = std::process::Command::new("git")
            .arg("-C")
            .arg(repo_dir.path())
            .args(["show", &format!("{branch}:x.txt")])
            .output()
            .unwrap();
        assert!(show.status.success());
        assert_eq!(
            String::from_utf8_lossy(&show.stdout),
            "hello from the coder"
        );
        let status = std::process::Command::new("git")
            .arg("-C")
            .arg(repo_dir.path())
            .args(["status", "--porcelain"])
            .output()
            .unwrap();
        assert!(status.stdout.is_empty(), "user checkout dirtied");
        assert!(!repo_dir.path().join("x.txt").exists());
    }

    #[tokio::test]
    async fn project_session_commits_to_main_no_branch() {
        // A managed-project session delivers to the project's main branch
        // (no car/coder/<id> branch); the file lands in the checkout itself.
        let projects_dir = tempfile::tempdir().unwrap();
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        // Point project creation at the temp root (serialize the env mutation).
        let _guard = crate::coder::project::projects_env_lock()
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let prev = std::env::var_os("CAR_PROJECTS_DIR");
        unsafe {
            std::env::set_var("CAR_PROJECTS_DIR", projects_dir.path());
        }

        let project = crate::coder::project::resolve_or_create_project(
            "My App",
            crate::coder::project::ProjectKind::App,
        )
        .unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "x.txt contains hi", "checks": [{"name": "content", "command": "grep -q hi x.txt"}]}"#,
                    json!([]),
                ),
                turn(
                    "",
                    json!([{"id": "c1", "name": "write_file", "arguments": {"path": "x.txt", "content": "hi project"}}]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });

        let response = start_session(
            &state,
            StartArgs {
                repo: project.repo_path.clone(),
                intent: "create x.txt containing hi".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(4),
                state_dir: state_dir.path().to_path_buf(),
                project: Some((project.slug.clone(), project.kind)),
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        confirm_session(&state, &session_id, None).await.unwrap();
        let entry = get_entry(&state, &session_id).await.unwrap();
        entry.task.lock().unwrap().take().unwrap().await.unwrap();

        let merged = approve_merge_session(&state, &session_id, true)
            .await
            .unwrap();
        assert_eq!(merged["state"], "merged");
        assert_eq!(merged["branch"], "main", "project sessions deliver to main");

        // The change is on main AND in the project's checkout (it's CAR-owned).
        let git = |args: &[&str]| {
            std::process::Command::new("git")
                .arg("-C")
                .arg(&project.repo_path)
                .args(args)
                .output()
                .unwrap()
        };
        let show = git(&["show", "main:x.txt"]);
        assert!(show.status.success());
        assert_eq!(String::from_utf8_lossy(&show.stdout), "hi project");
        assert_eq!(
            project.repo_path.join("x.txt").exists(),
            true,
            "lands in the checkout"
        );
        // No car/coder/* branch was created.
        let branches = git(&["branch", "--list", "car/coder/*"]);
        assert!(
            branches.stdout.is_empty(),
            "no coder branch for a project session"
        );

        unsafe {
            match prev {
                Some(v) => std::env::set_var("CAR_PROJECTS_DIR", v),
                None => std::env::remove_var("CAR_PROJECTS_DIR"),
            }
        }
    }

    #[tokio::test]
    async fn e2e_agent_project_builds_registers_and_invokes() {
        // The full coder→agent loop: create an Agent project → coder builds a
        // declarative agent that passes its scenarios → approve commits to main
        // AND registers the agent → it shows in agents.list and runs in-daemon.
        let projects_dir = tempfile::tempdir().unwrap();
        let declagents = tempfile::tempdir().unwrap();
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();

        let _guard = crate::coder::project::projects_env_lock()
            .lock()
            .unwrap_or_else(|e| e.into_inner());
        let prev_proj = std::env::var_os("CAR_PROJECTS_DIR");
        let prev_decl = std::env::var_os("CAR_DECLAGENTS_PATH");
        unsafe {
            std::env::set_var("CAR_PROJECTS_DIR", projects_dir.path());
            std::env::set_var(
                "CAR_DECLAGENTS_PATH",
                declagents.path().join("declagents.json"),
            );
        }

        let project = crate::coder::project::resolve_or_create_project(
            "Greeter Bot",
            crate::coder::project::ProjectKind::Agent,
        )
        .unwrap();

        // Build a state whose inference engine is our Script (so build_agent and
        // the scenario runs are deterministic). Production uses the real engine;
        // here we register the script as the shared inference via a wrapper.
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        // Script: (1) the agent spec, (2) scenario run → contains "hello".
        // The build loop + scenario eval both pull from this script.
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"name":"Greeter","identity":"You greet warmly.","tools":[],
                        "standing_goal":"greet","scenarios":[{"input":"hi","expect":"hello"}]}"#,
                    json!([]),
                ),
                turn("hello, friend!", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });

        let response = start_session(
            &state,
            StartArgs {
                repo: project.repo_path.clone(),
                intent: "a friendly greeter".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(3),
                state_dir: state_dir.path().to_path_buf(),
                project: Some((project.slug.clone(), project.kind)),
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        // Agent projects get a synthesized scenario contract.
        assert_eq!(
            response["contract"]["checks"][0]["name"],
            "agent_scenarios_pass"
        );

        confirm_session(&state, &session_id, None).await.unwrap();
        let entry = get_entry(&state, &session_id).await.unwrap();
        entry.task.lock().unwrap().take().unwrap().await.unwrap();

        {
            let session = entry.session.lock().await;
            assert_eq!(
                session.state,
                CoderState::NeedsApproval,
                "error: {:?}",
                session.error
            );
            assert!(
                session.built_agent.is_some(),
                "spec stashed for registration"
            );
        }

        // Approve → commit to main + register the agent.
        let merged = approve_merge_session(&state, &session_id, true)
            .await
            .unwrap();
        assert_eq!(merged["state"], "merged");
        assert_eq!(merged["branch"], "main");
        assert_eq!(merged["agent_id"].as_str().unwrap(), project.slug);

        // It's registered and shows in the declarative list.
        let reg = state.declagents().unwrap();
        let registered = reg.get(&project.slug).unwrap();
        assert_eq!(registered.name, "Greeter");
        assert_eq!(registered.scenarios.len(), 1);

        // agent.json was committed to the project's main.
        let show = std::process::Command::new("git")
            .arg("-C")
            .arg(&project.repo_path)
            .args(["show", "main:agent.json"])
            .output()
            .unwrap();
        assert!(show.status.success(), "agent.json on main");

        // It runs in-daemon (no process) — a fresh Script drives the run via a
        // second daemon state pointed at the same registry.
        let invoke_script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![turn("hello again!", json!([]))],
            cursor: AtomicUsize::new(0),
        });
        let exec_dir = tempfile::tempdir().unwrap();
        let exec = WorktreeExecutor::new(exec_dir.path());
        let runner = crate::coder::declarative::DeclarativeAgentRunner::new(
            &registered,
            invoke_script.as_ref(),
            &exec,
        );
        let run = runner.run("hi there").await;
        assert!(
            run.output.contains("hello"),
            "agent runs in-daemon: {run:?}"
        );

        unsafe {
            match prev_proj {
                Some(v) => std::env::set_var("CAR_PROJECTS_DIR", v),
                None => std::env::remove_var("CAR_PROJECTS_DIR"),
            }
            match prev_decl {
                Some(v) => std::env::set_var("CAR_DECLAGENTS_PATH", v),
                None => std::env::remove_var("CAR_DECLAGENTS_PATH"),
            }
        }
    }

    #[tokio::test]
    async fn failing_contract_ends_in_failed_with_results() {
        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        // The model never creates the file; 2 iterations then Failed.
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "impossible", "checks": [{"name": "missing", "command": "test -f never.txt"}]}"#,
                    json!([]),
                ),
                turn("i did nothing", json!([])),
                turn("still nothing", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });

        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "impossible task".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(2),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        confirm_session(&state, &session_id, None).await.unwrap();

        let entry = get_entry(&state, &session_id).await.unwrap();
        let handle = entry.task.lock().unwrap().take().unwrap();
        handle.await.unwrap();

        let session = entry.session.lock().await;
        assert_eq!(session.state, CoderState::Failed);
        assert!(session.error.as_deref().unwrap().contains("not satisfied"));
        assert!(!session.last_check_results[0].passed);

        // Approving a failed session is rejected.
        drop(session);
        let err = approve_merge_session(&state, &session_id, true)
            .await
            .unwrap_err();
        assert!(err.contains("expected needs_approval"), "{err}");
    }

    #[tokio::test]
    async fn confirm_with_edited_contract_replaces_proposal() {
        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "original", "checks": [{"name": "a", "command": "true"}]}"#,
                    json!([]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "x".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(2),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();

        let edited = OutcomeContract {
            description: "edited".into(),
            checks: vec![crate::coder::contract::ContractCheck {
                name: "edited_check".into(),
                command: "true".into(),
                expect_exit_zero: true,
                output_contains: None,
                timeout_secs: 10,
            }],
        };
        confirm_session(&state, &session_id, Some(edited))
            .await
            .unwrap();
        let entry = get_entry(&state, &session_id).await.unwrap();
        let handle = entry.task.lock().unwrap().take().unwrap();
        handle.await.unwrap();
        let session = entry.session.lock().await;
        assert_eq!(session.contract.as_ref().unwrap().description, "edited");
        // `true` always passes but there are no changes → diff fails → the
        // session still reaches NeedsApproval (diff failure is advisory).
        assert_eq!(session.state, CoderState::NeedsApproval);
    }

    #[tokio::test]
    async fn cancel_mid_run_abandons_session() {
        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        // Derivation, then a slow shell so cancel lands mid-run.
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "slow", "checks": [{"name": "n", "command": "test -f done.txt"}]}"#,
                    json!([]),
                ),
                turn(
                    "",
                    json!([{
                        "id": "c1", "name": "shell",
                        "arguments": {"command": "sleep 20", "timeout_secs": 30}
                    }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "slow".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(2),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        confirm_session(&state, &session_id, None).await.unwrap();
        // Give the loop a beat to get into the sleep, then cancel.
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
        let started = std::time::Instant::now();
        let result = cancel_session(&state, &session_id).await.unwrap();
        assert_eq!(result["state"], "abandoned");
        assert!(started.elapsed() < std::time::Duration::from_secs(5));

        // Worktree is cleaned up on the terminal transition.
        let entry = get_entry(&state, &session_id).await.unwrap();
        let session = entry.session.lock().await;
        assert!(session.workspace.is_none());
    }

    /// Full round-trip: the native loop's `ask_user` parks on the gate and
    /// emits `UserInputRequested`; `coder.respond` (driven from another task)
    /// fulfills it; the answer reaches the model, which writes it through to
    /// satisfy the contract → NeedsApproval.
    #[tokio::test]
    async fn respond_fulfills_a_pending_ask_user_request() {
        use car_inference::tasks::generate::Message;

        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        // Derivation turn, then ask_user, then write back the received answer.
        struct AskGen {
            cursor: AtomicUsize,
        }
        #[async_trait]
        impl TurnGenerator for AskGen {
            async fn generate(&self, req: GenerateRequest) -> Result<InferenceResult, String> {
                let i = self.cursor.fetch_add(1, Ordering::SeqCst);
                Ok(match i {
                    // Contract derivation.
                    0 => turn(
                        r#"{"description": "ans.txt records the answer",
                            "checks": [{"name": "c", "command": "grep -q FORTY-TWO ans.txt"}]}"#,
                        json!([]),
                    ),
                    // Loop turn 1: ask the user.
                    1 => turn(
                        "",
                        json!([{"id": "a1", "name": "ask_user",
                                "arguments": {"prompt": "what is the answer?"}}]),
                    ),
                    // Loop turn 2: echo the answer the loop fed back into a file.
                    2 => {
                        let answer = req
                            .messages
                            .as_ref()
                            .and_then(|ms| {
                                ms.iter().rev().find_map(|m| match m {
                                    Message::ToolResult { content, .. } => Some(content.clone()),
                                    _ => None,
                                })
                            })
                            .unwrap_or_default();
                        turn(
                            "",
                            json!([{"id": "w1", "name": "write_file",
                                    "arguments": {"path": "ans.txt", "content": answer}}]),
                        )
                    }
                    _ => turn("done", json!([])),
                })
            }
        }

        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "record the user's answer".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(4),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            Arc::new(AskGen {
                cursor: AtomicUsize::new(0),
            }),
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        confirm_session(&state, &session_id, None).await.unwrap();

        let entry = get_entry(&state, &session_id).await.unwrap();

        // Another task: wait for the question to park, then answer it.
        {
            let state = state.clone();
            let sid = session_id.clone();
            let gate = entry.user_input.clone();
            tokio::spawn(async move {
                for _ in 0..200 {
                    if gate.is_pending() {
                        break;
                    }
                    tokio::time::sleep(std::time::Duration::from_millis(20)).await;
                }
                let req: JsonRpcMessage = serde_json::from_value(json!({
                    "jsonrpc": "2.0", "id": 1, "method": "coder.respond",
                    "params": {"session_id": sid, "text": "FORTY-TWO"},
                }))
                .unwrap();
                handle_coder_respond(&req, &state).await.unwrap();
            });
        }

        let handle = entry.task.lock().unwrap().take().unwrap();
        handle.await.unwrap();

        let session = entry.session.lock().await;
        assert_eq!(
            session.state,
            CoderState::NeedsApproval,
            "error: {:?}",
            session.error
        );
        drop(session);
        assert!(entry.events.lock().await.iter().any(|e| matches!(
            &e.kind,
            CoderEventKind::UserInputRequested { prompt } if prompt == "what is the answer?"
        )));
    }

    /// `coder.respond` errors clearly when nothing is pending.
    #[tokio::test]
    async fn respond_errors_when_no_request_pending() {
        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![turn(
                r#"{"description": "x", "checks": [{"name": "a", "command": "true"}]}"#,
                json!([]),
            )],
            cursor: AtomicUsize::new(0),
        });
        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "x".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(1),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();

        let req: JsonRpcMessage = serde_json::from_value(json!({
            "jsonrpc": "2.0", "id": 1, "method": "coder.respond",
            "params": {"session_id": session_id, "text": "unexpected"},
        }))
        .unwrap();
        let err = handle_coder_respond(&req, &state).await.unwrap_err();
        assert!(err.contains("no pending user-input request"), "{err}");
    }

    /// Cancel unblocks a request parked on the gate: the `GateAsker` returns an
    /// error (not a hang) and the session ends Abandoned.
    #[tokio::test]
    async fn cancel_unblocks_a_waiting_ask_user_request() {
        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        // Derivation, then ask_user (and nothing more — it will block on the
        // gate until cancel unblocks it).
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "blocks", "checks": [{"name": "n", "command": "test -f done.txt"}]}"#,
                    json!([]),
                ),
                turn(
                    "",
                    json!([{"id": "a1", "name": "ask_user",
                            "arguments": {"prompt": "blocking question"}}]),
                ),
            ],
            cursor: AtomicUsize::new(0),
        });
        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "blocks".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(2),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        confirm_session(&state, &session_id, None).await.unwrap();

        let entry = get_entry(&state, &session_id).await.unwrap();
        // Wait for the question to park on the gate.
        for _ in 0..200 {
            if entry.user_input.is_pending() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
        assert!(entry.user_input.is_pending(), "ask_user should have parked");

        let started = std::time::Instant::now();
        let result = cancel_session(&state, &session_id).await.unwrap();
        assert_eq!(result["state"], "abandoned");
        // Cancel must unblock immediately — never wait out the ask timeout.
        assert!(started.elapsed() < std::time::Duration::from_secs(5));

        let session = entry.session.lock().await;
        assert_eq!(session.state, CoderState::Abandoned);
    }

    /// Serialize the env-var-touching tests below: `CAR_CODER_CONFIG` is
    /// process-global, so two tests setting it must not overlap.
    fn config_env_lock() -> &'static std::sync::Mutex<()> {
        static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
        LOCK.get_or_init(|| std::sync::Mutex::new(()))
    }

    /// End-to-end config wiring: a `coder.toml` with `keep_workspace_on_failure`
    /// and `default_max_iterations` takes effect through `handle_coder_start` —
    /// the session honors the iteration default and retains its worktree on a
    /// Failed terminal state.
    #[tokio::test]
    async fn coder_toml_keep_on_failure_and_default_iterations_take_effect() {
        let _guard = config_env_lock().lock().unwrap();

        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let cfg_dir = tempfile::tempdir().unwrap();
        let cfg_path = cfg_dir.path().join("coder.toml");
        std::fs::write(
            &cfg_path,
            "[coder]\nkeep_workspace_on_failure = true\ndefault_max_iterations = 3\n",
        )
        .unwrap();
        // SAFETY: single-threaded test body, guarded by config_env_lock.
        std::env::set_var("CAR_CODER_CONFIG", &cfg_path);

        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));

        // The model never creates the file → contract stays red → Failed.
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "impossible", "checks": [{"name": "missing", "command": "test -f never.txt"}]}"#,
                    json!([]),
                ),
                turn("nothing", json!([])),
                turn("still nothing", json!([])),
                turn("nope", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });

        // No max_iterations in args (None) → start_session falls back to the
        // config's default. Assert the config value first, then drive the
        // actual fallback path below.
        assert_eq!(
            CoderConfig::load().default_max_iterations,
            3,
            "config default_max_iterations should load"
        );

        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "impossible task".into(),
                engine: EngineChoice::Native,
                max_iterations: None,
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        let worktree = PathBuf::from(response["worktree"].as_str().unwrap());

        confirm_session(&state, &session_id, None).await.unwrap();
        let entry = get_entry(&state, &session_id).await.unwrap();
        let handle = entry.task.lock().unwrap().take().unwrap();
        handle.await.unwrap();

        let session = entry.session.lock().await;
        assert_eq!(session.state, CoderState::Failed);
        assert!(session.keep_workspace_on_failure);
        // Iteration cap came from the config default, not the built-in 8.
        assert_eq!(session.max_iterations, 3);
        // Worktree retained for postmortem, path reported in the snapshot.
        assert!(
            worktree.is_dir(),
            "worktree should survive Failed under keep flag"
        );
        assert_eq!(session.workspace_path.as_deref(), Some(worktree.as_path()));
        drop(session);

        // A retained-worktree notice was emitted for the operator.
        assert!(entry.events.lock().await.iter().any(|e| matches!(
            &e.kind,
            CoderEventKind::Error { message } if message.contains("retained for postmortem")
        )));

        std::env::remove_var("CAR_CODER_CONFIG");
        // Reap the leaked worktree registration.
        let _ = std::process::Command::new("git")
            .arg("-C")
            .arg(repo_dir.path())
            .args(["worktree", "remove", "--force"])
            .arg(&worktree)
            .output();
    }

    /// A missing config file yields the documented defaults (worktree reaped on
    /// failure, no retention notice).
    #[tokio::test]
    async fn missing_coder_toml_uses_defaults() {
        let _guard = config_env_lock().lock().unwrap();

        let repo_dir = tempfile::tempdir().unwrap();
        init_repo(repo_dir.path());
        let state_dir = tempfile::tempdir().unwrap();
        let journal = tempfile::tempdir().unwrap();
        let cfg_dir = tempfile::tempdir().unwrap();
        // Point at a path that does not exist → load() returns defaults.
        std::env::set_var("CAR_CODER_CONFIG", cfg_dir.path().join("absent.toml"));

        assert_eq!(CoderConfig::load(), CoderConfig::default());

        let state = Arc::new(ServerState::standalone(journal.path().to_path_buf()));
        let script: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    r#"{"description": "impossible", "checks": [{"name": "missing", "command": "test -f never.txt"}]}"#,
                    json!([]),
                ),
                turn("nothing", json!([])),
                turn("still nothing", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let response = start_session(
            &state,
            StartArgs {
                repo: repo_dir.path().to_path_buf(),
                intent: "impossible".into(),
                engine: EngineChoice::Native,
                max_iterations: Some(2),
                state_dir: state_dir.path().to_path_buf(),
                project: None,
            },
            script,
        )
        .await
        .unwrap();
        let session_id = response["session_id"].as_str().unwrap().to_string();
        let worktree = PathBuf::from(response["worktree"].as_str().unwrap());

        confirm_session(&state, &session_id, None).await.unwrap();
        let entry = get_entry(&state, &session_id).await.unwrap();
        let handle = entry.task.lock().unwrap().take().unwrap();
        handle.await.unwrap();

        let session = entry.session.lock().await;
        assert_eq!(session.state, CoderState::Failed);
        assert!(!session.keep_workspace_on_failure, "default is not to keep");
        // Default behavior: worktree reaped.
        assert!(
            !worktree.exists(),
            "worktree should be reaped under defaults"
        );

        std::env::remove_var("CAR_CODER_CONFIG");
    }

    /// H2 Part 2 ranking harness — RUNS the merged eval fixtures in
    /// `car-registry/eval/{fleet.json,discovery_ranking.jsonl}` against the
    /// REAL ranking implementation (`rank_services`/`score_service`) with the
    /// SHIPPED scoring defaults (config dump printed per run). Deterministic
    /// and inference-free: the only live-model step of `discovery.resolve` is
    /// text→embedding, and `rank_services` takes pre-computed embeddings, so
    /// the harness injects a deterministic lexical embedder (hashed token +
    /// character-4-gram counts) at that seam — the fixtures' needs and
    /// capability texts were written for lexical separability. Routing-store
    /// state is seeded/reset per run through the real [`RoutingStore`], keyed
    /// by agentdns identifier — exactly what `discovery.report` records — so
    /// the post-feedback regime exercises the same persistence path.
    ///
    /// Targets (acceptance spec, `docs/proposals/h2-builder-discovery-acceptance.md`):
    /// top-1 ≥ 85% and MRR ≥ 0.9, cold-start and post-feedback asserted
    /// separately; the non-declarative demotion case; the deterministic
    /// identifier tie-break.
    mod ranking_eval {
        use super::*;

        const FLEET: &str = include_str!("../../../car-registry/eval/fleet.json");
        const CASES: &str = include_str!("../../../car-registry/eval/discovery_ranking.jsonl");

        #[derive(Debug, serde::Deserialize)]
        struct FleetEntry {
            identifier: String,
            name: String,
            kind: String,
            #[serde(default)]
            agent_id: Option<String>,
            capability_text: String,
            #[serde(default)]
            successes: u64,
            #[serde(default)]
            failures: u64,
        }

        #[derive(Debug, serde::Deserialize)]
        struct Fleet {
            agents: Vec<FleetEntry>,
        }

        #[derive(Debug, serde::Deserialize)]
        struct RankingCase {
            id: String,
            mode: String,
            need: String,
            expected_top: String,
            #[serde(default)]
            expected_below: Option<String>,
            #[serde(default)]
            non_declarative: Option<bool>,
            #[serde(default)]
            tie_break: Option<bool>,
        }

        fn load_fleet() -> Fleet {
            serde_json::from_str(FLEET).expect("fleet.json parses")
        }

        fn load_cases(mode: &str) -> Vec<RankingCase> {
            CASES
                .lines()
                .filter(|l| !l.trim().is_empty())
                .map(|l| serde_json::from_str::<RankingCase>(l).expect("ranking case parses"))
                .filter(|c| c.mode == mode)
                .collect()
        }

        // --- deterministic test embedder (the injectable seam) ---

        const EMB_DIM: usize = 2048;

        /// Stopwords stripped before hashing — function words that would add
        /// shared-but-meaningless mass between every need and every doc.
        const STOPWORDS: &[&str] = &[
            "a", "an", "and", "are", "as", "at", "back", "be", "by", "few", "for", "from",
            "give", "has", "have", "in", "into", "is", "it", "me", "my", "of", "on", "or",
            "out", "s", "the", "this", "that", "these", "those", "to", "was", "what", "when",
            "where", "which", "with", "your",
        ];

        fn fnv1a(bytes: &[u8]) -> u64 {
            let mut h: u64 = 0xcbf29ce484222325;
            for b in bytes {
                h ^= *b as u64;
                h = h.wrapping_mul(0x100000001b3);
            }
            h
        }

        /// Deterministic lexical embedding: hashed counts of each token plus
        /// its character 4-grams (so morphological variants — "translate" /
        /// "Translates", "search" / "searches" — still overlap). Pure, no
        /// model, identical across runs/platforms; identical texts embed to
        /// identical vectors, which is what makes the tie-break case an exact
        /// score tie.
        fn test_embed(text: &str) -> Vec<f32> {
            let mut v = vec![0f32; EMB_DIM];
            let lower = text.to_lowercase();
            for tok in lower.split(|c: char| !c.is_ascii_alphanumeric()) {
                if tok.is_empty() || STOPWORDS.contains(&tok) {
                    continue;
                }
                v[(fnv1a(tok.as_bytes()) % EMB_DIM as u64) as usize] += 1.0;
                if tok.len() > 4 {
                    for gram in tok.as_bytes().windows(4) {
                        v[(fnv1a(gram) % EMB_DIM as u64) as usize] += 1.0;
                    }
                }
            }
            v
        }

        fn services_from_fleet(fleet: &Fleet) -> Vec<DiscoveredService> {
            fleet
                .agents
                .iter()
                .map(|e| DiscoveredService {
                    identifier: e.identifier.clone(),
                    name: e.name.clone(),
                    kind: e.kind.clone(),
                    protocol: "test".into(),
                    capability_text: e.capability_text.clone(),
                    agent_id: e.agent_id.clone(),
                    endpoint: None,
                })
                .collect()
        }

        /// Seed the fleet's outcome histories into a REAL routing store, keyed
        /// by agentdns identifier — the exact writes `discovery.report` makes.
        fn seeded_routing(
            fleet: &Fleet,
            dir: &tempfile::TempDir,
        ) -> car_registry::routing::RoutingSnapshot {
            let store = car_registry::routing::RoutingStore::at(dir.path().join("routing.json"));
            for e in &fleet.agents {
                for _ in 0..e.successes {
                    store.record_outcome(&e.identifier, true).unwrap();
                }
                for _ in 0..e.failures {
                    store.record_outcome(&e.identifier, false).unwrap();
                }
            }
            store.snapshot()
        }

        fn dump_config() {
            println!(
                "ranking-eval config (SHIPPED defaults): \
                 ROUTE_SIMILARITY_WEIGHT={ROUTE_SIMILARITY_WEIGHT} \
                 prior_weight={} ROUTE_PRIOR_EXPLORATION={ROUTE_PRIOR_EXPLORATION} \
                 LEARNED_SIM_WEIGHT={LEARNED_SIM_WEIGHT} ROUTE_EDGE_WEIGHT={ROUTE_EDGE_WEIGHT} \
                 prior=Beta(success+1,fail+1) UCB (car-memgine::utility) \
                 embedder=deterministic lexical (token + char-4-gram FNV-1a counts, dim {EMB_DIM})",
                1.0 - ROUTE_SIMILARITY_WEIGHT
            );
        }

        /// Run one regime's cases through the real ranker; assert the spec
        /// targets plus every case-level ordering/tie-break claim.
        fn run_mode(mode: &str, routing: &car_registry::routing::RoutingSnapshot) {
            dump_config();
            let fleet = load_fleet();
            let services = services_from_fleet(&fleet);
            let cap_embs: Vec<Vec<f32>> = services
                .iter()
                .map(|s| test_embed(&s.capability_text))
                .collect();
            let cases = load_cases(mode);
            assert!(!cases.is_empty(), "no cases for mode {mode}");

            let mut top1 = 0usize;
            let mut mrr = 0f64;
            for case in &cases {
                let need_emb = test_embed(&case.need);
                let ranked = rank_services(&need_emb, &cap_embs, &services, routing);
                let rank_of = |ident: &str| -> usize {
                    ranked
                        .iter()
                        .position(|(i, ..)| services[*i].identifier == ident)
                        .unwrap_or_else(|| panic!("{ident} not in ranking"))
                };
                let got_rank = rank_of(&case.expected_top);
                if got_rank == 0 {
                    top1 += 1;
                }
                mrr += 1.0 / (got_rank + 1) as f64;
                println!(
                    "  [{mode}] {}: expected_top={} rank={} (top={})",
                    case.id,
                    case.expected_top,
                    got_rank + 1,
                    services[ranked[0].0].identifier,
                );

                if let Some(below) = &case.expected_below {
                    assert!(
                        rank_of(&case.expected_top) < rank_of(below),
                        "[{}] {} must outrank {}",
                        case.id,
                        case.expected_top,
                        below
                    );
                    if case.non_declarative == Some(true) {
                        // THE demotion proof: the demoted candidate is NOT a
                        // declarative agent — impossible before the unified
                        // substrate (only declarative agents learned).
                        let demoted = services
                            .iter()
                            .find(|s| &s.identifier == below)
                            .expect("demoted candidate in fleet");
                        assert_ne!(demoted.kind, "declarative");
                        assert!(demoted.agent_id.is_none());
                    }
                }

                if case.tie_break == Some(true) {
                    // Twins with identical capability text and identical
                    // (empty) history tie EXACTLY; ascending identifier wins.
                    let alpha = rank_of("agentdns://local/service/alpha-echo");
                    let beta = rank_of("agentdns://local/service/beta-echo");
                    assert_eq!(
                        ranked[alpha].1, ranked[beta].1,
                        "echo twins must tie exactly"
                    );
                    assert!(
                        alpha < beta,
                        "tie must break on ascending identifier (alpha before beta)"
                    );
                    assert_eq!(case.expected_top, "agentdns://local/service/alpha-echo");
                }
            }

            let n = cases.len() as f64;
            let top1_rate = top1 as f64 / n;
            let mrr = mrr / n;
            println!("  [{mode}] top-1 = {top1}/{} ({top1_rate:.2}), MRR = {mrr:.3}", cases.len());
            assert!(
                top1_rate >= 0.85,
                "[{mode}] top-1 {top1_rate:.2} below the 0.85 target"
            );
            assert!(mrr >= 0.9, "[{mode}] MRR {mrr:.3} below the 0.9 target");
        }

        #[test]
        fn cold_start_cases_hit_targets() {
            // Cold start: a fresh (empty) routing store — every prior is the
            // uniform posterior's 0.5; ranking is capability similarity alone.
            let dir = tempfile::tempdir().unwrap();
            let routing = car_registry::routing::RoutingStore::at(dir.path().join("routing.json"))
                .snapshot();
            assert!(routing.agents.is_empty());
            run_mode("cold_start", &routing);
        }

        #[test]
        fn post_feedback_cases_hit_targets() {
            // Post feedback: the fleet's seeded successes/failures recorded
            // through the real store under each agentdns identifier (the
            // discovery.report path), then ranked.
            let dir = tempfile::tempdir().unwrap();
            let routing = seeded_routing(&load_fleet(), &dir);
            run_mode("post_feedback", &routing);
        }

        /// Acceptance #1: ONE scoring substrate — the same fleet ranked
        /// through `declagents.route`'s `rank_agents` and
        /// `discovery.resolve`'s `rank_services` yields the same relative
        /// order for the shared (declarative) candidates, with history seeded
        /// under a MIX of agent-id and identifier keys so the merged-posterior
        /// fold is what's proven, not a single lookup path.
        #[test]
        fn both_ranking_paths_order_shared_candidates_identically() {
            let fleet = load_fleet();
            let decl: Vec<&FleetEntry> = fleet
                .agents
                .iter()
                .filter(|e| e.kind == "declarative")
                .collect();
            assert!(decl.len() >= 4, "fleet must carry declarative agents");

            let specs: Vec<car_registry::declarative::DeclarativeAgentSpec> = decl
                .iter()
                .map(|e| spec(e.agent_id.as_deref().unwrap(), &e.capability_text, &[]))
                .collect();
            let services: Vec<DiscoveredService> = decl
                .iter()
                .map(|e| DiscoveredService {
                    identifier: e.identifier.clone(),
                    name: e.name.clone(),
                    kind: e.kind.clone(),
                    protocol: "test".into(),
                    capability_text: e.capability_text.clone(),
                    agent_id: e.agent_id.clone(),
                    endpoint: None,
                })
                .collect();
            // Both paths score the same capability surface: hand them the
            // SAME per-candidate embeddings.
            let embs: Vec<Vec<f32>> = decl.iter().map(|e| test_embed(&e.capability_text)).collect();

            // Seed history alternating between the two key spaces: agent id
            // (what declagents.route/invoke records) and agentdns identifier
            // (what discovery.report records).
            let dir = tempfile::tempdir().unwrap();
            let store = car_registry::routing::RoutingStore::at(dir.path().join("routing.json"));
            for (i, e) in decl.iter().enumerate() {
                let key = if i % 2 == 0 {
                    e.agent_id.clone().unwrap()
                } else {
                    e.identifier.clone()
                };
                for _ in 0..e.successes {
                    store.record_outcome(&key, true).unwrap();
                }
                for _ in 0..e.failures {
                    store.record_outcome(&key, false).unwrap();
                }
            }
            // One agent also gets a learned capability centroid, so the
            // learned-similarity blend is covered by the parity claim too.
            store
                .record_capability(
                    decl[0].agent_id.as_deref().unwrap(),
                    &test_embed("plan a research report"),
                )
                .unwrap();
            let routing = store.snapshot();

            for case in load_cases("cold_start")
                .into_iter()
                .chain(load_cases("post_feedback"))
            {
                let need_emb = test_embed(&case.need);
                let via_route: Vec<String> =
                    rank_agents(&need_emb, &embs, &specs, &routing, None)
                        .into_iter()
                        .map(|(i, ..)| specs[i].id.clone())
                        .collect();
                let via_discovery: Vec<String> =
                    rank_services(&need_emb, &embs, &services, &routing)
                        .into_iter()
                        .map(|(i, ..)| services[i].agent_id.clone().unwrap())
                        .collect();
                assert_eq!(
                    via_route, via_discovery,
                    "need {:?}: declagents.route and discovery.resolve disagree",
                    case.need
                );
            }
        }
    }

    #[test]
    fn summarize_repo_reports_entries_and_build_system() {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join("Cargo.toml"), "[package]").unwrap();
        std::fs::write(dir.path().join("main.rs"), "").unwrap();
        let s = summarize_repo(dir.path());
        assert!(s.contains("Cargo.toml"));
        assert!(s.contains("Rust (cargo)"));
    }
}
