khive-mcp 0.4.0

//! khived daemon client — forwarding + auto-spawn.
//!
//! The daemon server lives in `khive-runtime::daemon`. This module provides the
//! client side: [`forward_or_spawn`] connects to the daemon, auto-spawns it on
//! first use, and maps responses to MCP error types. Every failure path falls
//! back to `None` so the caller can dispatch locally.
//!
//! Also provides the [`khive_runtime::daemon::DaemonDispatch`] impl for [`crate::server::KhiveMcpServer`].

use std::process::Stdio;

use async_trait::async_trait;
use khive_runtime::daemon::{
    self, acquire_recovery_lock, env_truthy, pid_path, read_frame, socket_path, write_frame,
    DaemonRequestFrame, DaemonResponseFrame, PROTOCOL_VERSION,
};
use rmcp::ErrorData as McpError;
use tokio::net::UnixStream;

use crate::tools::request::RequestParams;

// ── test instrumentation seams ────────────────────────────────────────────────
//
// These counters are only compiled in `#[cfg(test)]` builds. They allow tests
// to assert that kill_stale_daemon_inner and spawn_daemon were called exactly
// the expected number of times — making the recheck-under-lock test
// fail-if-reverted: without the recheck, both counters would be non-zero even
// when a fresh daemon is already alive.

#[cfg(test)]
pub(crate) static KILL_COUNT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

#[cfg(test)]
pub(crate) static SPAWN_COUNT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

/// When set to `true` in tests, `pid_is_khive_daemon` returns `true` for any
/// positive PID.  This makes every PID file entry SIGTERM-eligible so that a
/// reverted recheck-under-lock would cause `kill_stale_daemon_inner` to attempt
/// SIGTERM against the real daemon PID — the `KILL_COUNT` assertion catches
/// both the SIGTERM-eligible and the skip-SIGTERM paths.
#[cfg(test)]
pub(crate) static FORCE_PID_IS_DAEMON: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// Counts how many times the daemon's dispatcher has been invoked for a
/// NON-probe request.  The exactly-once test asserts this is exactly 1
/// across the entire recovery path.  A reverted fix (real request used as
/// probe + re-forwarded at the call site) yields 2 and fails the assertion.
#[cfg(test)]
pub(crate) static DAEMON_DISPATCH: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

/// Test-only rendezvous point: when set, every [`kill_and_respawn`] call
/// waits on this barrier right after its own initial probe independently
/// classifies the daemon `Dead` and BEFORE it attempts the recoverer lock.
///
/// #838 round-2 Finding 2: the two-recoverer regression test previously let
/// each `kill_and_respawn` call reach that point on its own schedule, then
/// used a side-channel poll on `SPAWN_COUNT` to decide when to fake a live
/// replacement daemon. With the recoverer lock deliberately removed
/// (sabotage check), the two calls still did not race concurrently — normal
/// tokio scheduling let the first call run far enough ahead that its
/// classification, kill, and spawn completed (flipping the fake daemon
/// live) before the second call's own classification rounds observed
/// anything, so the test passed even with the lock's acquisition deleted.
/// This barrier forces every concurrent recoverer under test to reach the
/// classification-complete point at the same instant, so the recoverer
/// lock (or its absence) is what determines whether one or both proceed to
/// kill+spawn — not scheduler luck.
#[cfg(test)]
pub(crate) static RECOVERY_RACE_BARRIER: std::sync::Mutex<
    Option<std::sync::Arc<tokio::sync::Barrier>>,
> = std::sync::Mutex::new(None);

/// Test-only second rendezvous point, right before a recoverer that
/// classified `Dead` actually commits to `kill_stale_daemon_inner` +
/// `spawn_daemon`. `RECOVERY_RACE_BARRIER` alone only synchronizes the
/// START of `confirm_genuinely_dead`; its internal rounds still run against
/// a real, deadline-based lock check (`spawn_blocking` against a real OS
/// thread pool), which introduces enough real-world jitter over
/// `DEAD_CONFIRM_ROUNDS` rounds that one recoverer can still finish and
/// call `spawn_daemon()` measurably before the other's last round — giving
/// the fake-daemon watcher time to bind and make the slower recoverer
/// observe `Alive` on its own, even with the recoverer lock removed. This
/// second, BOUNDED barrier (falls through after its bound rather than
/// waiting forever, so a recoverer that legitimately never reaches this
/// point — e.g. because it is still blocked acquiring the real recoverer
/// lock — cannot deadlock it) forces both recoverers that did independently
/// classify `Dead` to commit to spawning at the same instant, closing the
/// jitter window the fake-daemon watcher could otherwise race against.
#[cfg(test)]
pub(crate) static SPAWN_COMMIT_BARRIER: std::sync::Mutex<
    Option<std::sync::Arc<tokio::sync::Barrier>>,
> = std::sync::Mutex::new(None);

#[cfg(test)]
pub(crate) fn reset_counters() {
    KILL_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
    SPAWN_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
    FORCE_PID_IS_DAEMON.store(false, std::sync::atomic::Ordering::SeqCst);
    DAEMON_DISPATCH.store(0, std::sync::atomic::Ordering::SeqCst);
    *RECOVERY_RACE_BARRIER
        .lock()
        .expect("barrier mutex poisoned") = None;
    *SPAWN_COMMIT_BARRIER.lock().expect("barrier mutex poisoned") = None;
}

// ── local-dispatch fallback telemetry ─────────────────────────────────────────
//
// Every path below that returns `None` (or is matched as a fallback outcome)
// means the caller silently dispatches locally instead of via the warm daemon.
// A silent fallback is the bug this instrumentation exists to surface: it must
// always be loud (a structured WARN) and counted. These are process-global
// production counters (not `#[cfg(test)]`-gated like the instrumentation seams
// above) — they realize the metric `khive_daemon_fallback_total{reason}`; a
// future metrics-export slice can read them without touching call sites.

/// Reason a request fell back to local dispatch instead of the warm daemon.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum FallbackReason {
    ConfigMismatch,
    NamespaceMismatch,
    NoSocket,
    // REASON: #644 made "the real frame was already written" a hard error
    // (never a local-dispatch fallback) in `forward_or_spawn`, since retrying
    // or falling back after a completed write risks a duplicate mutation. That
    // removed the only production call sites for these two reasons. They stay
    // in the closed 5-value metrics set for forward-compatibility (a future
    // reason may reuse the same severity tier) and are exercised directly by
    // the counter tests below, matching the existing no-production-call-site
    // pattern already used for `fallback_total`/`fallback_count` in this file.
    #[allow(dead_code)]
    ParseFailure,
    #[allow(dead_code)]
    ProtocolMismatch,
}

impl FallbackReason {
    fn as_str(self) -> &'static str {
        match self {
            FallbackReason::ConfigMismatch => "config_mismatch",
            FallbackReason::NamespaceMismatch => "namespace_mismatch",
            FallbackReason::NoSocket => "no_socket",
            FallbackReason::ParseFailure => "parse_failure",
            FallbackReason::ProtocolMismatch => "protocol_mismatch",
        }
    }

    fn counter(self) -> &'static std::sync::atomic::AtomicUsize {
        match self {
            FallbackReason::ConfigMismatch => &FALLBACK_CONFIG_MISMATCH,
            FallbackReason::NamespaceMismatch => &FALLBACK_NAMESPACE_MISMATCH,
            FallbackReason::NoSocket => &FALLBACK_NO_SOCKET,
            FallbackReason::ParseFailure => &FALLBACK_PARSE_FAILURE,
            FallbackReason::ProtocolMismatch => &FALLBACK_PROTOCOL_MISMATCH,
        }
    }

    /// Legitimacy tier used by the `KHIVE_DAEMON_STRICT` graduated fail-loud
    /// policy (SPEC_DRAFT §3 D2). Only `Illegitimate` reasons are elevated by
    /// strict mode; the other two tiers are always quiet, regardless of mode.
    fn severity(self) -> FallbackSeverity {
        match self {
            // A real misconfiguration: the client and daemon should have
            // agreed on `config_id`/namespace visibility and didn't. Never
            // expected on a correctly-configured fleet post-D1.
            FallbackReason::ConfigMismatch | FallbackReason::NamespaceMismatch => {
                FallbackSeverity::Illegitimate
            }
            // `ParseFailure` and `ProtocolMismatch` are this module's own
            // "stale/rolling daemon" bucket: both trigger the same
            // kill-and-respawn recovery path (see the `ForwardOutcome::ParseFailure
            // | ForwardOutcome::ProtocolMismatch` handling above) and both
            // represent a transient protocol-version drift during a rolling
            // upgrade, not a persistent misconfiguration. SPEC_DRAFT §3 D2's
            // table names only `version_mismatch` explicitly; `ParseFailure`
            // is folded into the same `RolloutTransient` tier because the
            // code already treats it identically to `ProtocolMismatch`
            // everywhere else in this file.
            FallbackReason::ProtocolMismatch | FallbackReason::ParseFailure => {
                FallbackSeverity::RolloutTransient
            }
            // No daemon reachable at all — the ADR-049-mandated fallback
            // path (CI, `KHIVE_NO_DAEMON=1`, read-only FS, spawn failure).
            FallbackReason::NoSocket => FallbackSeverity::NoDaemon,
        }
    }
}

/// Legitimacy tier for a [`FallbackReason`], keyed by the graduated fail-loud
/// policy in SPEC_DRAFT §3 D2. See [`FallbackReason::severity`] for the
/// per-variant mapping and its rationale.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FallbackSeverity {
    /// A real misconfiguration. Elevated to an error-level event (plus a
    /// dedicated violation counter) when `KHIVE_DAEMON_STRICT=1`.
    Illegitimate,
    /// Expected during a rolling upgrade; self-heals via kill-and-respawn.
    /// Never elevated, in strict mode or otherwise.
    RolloutTransient,
    /// No daemon to forward to at all. Never elevated — this is the
    /// ADR-049-mandated safety net, not a bug.
    NoDaemon,
}

/// `KHIVE_DAEMON_STRICT=1` elevates `Illegitimate`-severity fallbacks
/// (`ConfigMismatch`, `NamespaceMismatch`) from a WARN to an error-level
/// structured event plus a dedicated violation counter (D2-R1). It never
/// disables the fallback itself — the request still completes locally so the
/// caller is never dropped (SPEC_DRAFT §3 D2) — it only makes an illegitimate
/// mismatch impossible to miss.
///
/// No hosted-vs-local auto-detection signal exists in this codebase (there is
/// no build-time or runtime "is this the hosted fleet image" flag anywhere in
/// the workspace) and none is invented here: this resolves as a plain opt-in,
/// default OFF, matching the shape of `is_strict_actor_mode`'s
/// `KHIVE_REQUIRE_ATTRIBUTED_ACTOR`. The hosted/fleet image is expected to set
/// `KHIVE_DAEMON_STRICT=1` explicitly in its own deployment environment; a
/// bare local `kkernel` invocation is unaffected by default.
fn is_daemon_strict_mode() -> bool {
    env_truthy("KHIVE_DAEMON_STRICT")
}

/// `khive_daemon_fallback_total{reason="config_mismatch"}`
static FALLBACK_CONFIG_MISMATCH: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);
/// `khive_daemon_fallback_total{reason="namespace_mismatch"}`
static FALLBACK_NAMESPACE_MISMATCH: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);
/// `khive_daemon_fallback_total{reason="no_socket"}`
static FALLBACK_NO_SOCKET: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
/// `khive_daemon_fallback_total{reason="parse_failure"}`
static FALLBACK_PARSE_FAILURE: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);
/// `khive_daemon_fallback_total{reason="protocol_mismatch"}`
static FALLBACK_PROTOCOL_MISMATCH: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);
/// `khive_daemon_fallback_strict_violations_total` — count of `Illegitimate`
/// fallbacks (`config_mismatch`/`namespace_mismatch`) observed while
/// `KHIVE_DAEMON_STRICT=1` was set (D2-R1). Distinct from the five
/// per-reason counters above: this one is scoped to exactly the elevated
/// (error-level) events, so a load-harness (D2-R2) can hard-fail on
/// "any nonzero" without having to know which reasons are illegitimate.
static FALLBACK_STRICT_VIOLATIONS: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

// REASON: these accessors have no production call site yet — this slice adds the
// counters and their read path; a future metrics-export slice wires them to a real
// exporter (see the `khive_daemon_fallback_total{reason}` doc comments above) without
// needing to touch `record_fallback`'s call sites. Exercised directly by the counter
// tests below in the meantime.
#[allow(dead_code)]
/// `khive_daemon_fallback_total{reason="<all>"}` — total fallback count across
/// all reasons, derived by summing the five reason counters on read rather
/// than tracked as a separate atomic. A separate total can be observed
/// momentarily out of sync with the sum of reasons (two independent
/// `fetch_add`s); deriving it makes total == sum-of-reasons a structural
/// invariant instead of a timing-dependent one.
pub(crate) fn fallback_total() -> usize {
    use std::sync::atomic::Ordering::SeqCst;
    FALLBACK_CONFIG_MISMATCH.load(SeqCst)
        + FALLBACK_NAMESPACE_MISMATCH.load(SeqCst)
        + FALLBACK_NO_SOCKET.load(SeqCst)
        + FALLBACK_PARSE_FAILURE.load(SeqCst)
        + FALLBACK_PROTOCOL_MISMATCH.load(SeqCst)
}

#[allow(dead_code)]
/// Fallback count for a single `reason`.
pub(crate) fn fallback_count(reason: FallbackReason) -> usize {
    reason.counter().load(std::sync::atomic::Ordering::SeqCst)
}

#[allow(dead_code)]
/// `khive_daemon_fallback_strict_violations_total` — see
/// [`FALLBACK_STRICT_VIOLATIONS`]. No production call site yet, same as
/// `fallback_total`/`fallback_count` above; exercised directly by tests.
pub(crate) fn fallback_strict_violations() -> usize {
    FALLBACK_STRICT_VIOLATIONS.load(std::sync::atomic::Ordering::SeqCst)
}

#[cfg(test)]
pub(crate) fn reset_fallback_counters() {
    use std::sync::atomic::Ordering::SeqCst;
    FALLBACK_CONFIG_MISMATCH.store(0, SeqCst);
    FALLBACK_NAMESPACE_MISMATCH.store(0, SeqCst);
    FALLBACK_NO_SOCKET.store(0, SeqCst);
    FALLBACK_PARSE_FAILURE.store(0, SeqCst);
    FALLBACK_PROTOCOL_MISMATCH.store(0, SeqCst);
    FALLBACK_STRICT_VIOLATIONS.store(0, SeqCst);
}

/// Emit the standardized `daemon_fallback` event and increment the matching
/// counters. Call exactly once per fallback event, at the point where the
/// caller is about to dispatch locally instead of via the warm daemon.
///
/// `config_id_daemon` is `None` when no daemon response was available to read
/// it from (socket unreachable, undecodable response) or the daemon omitted it
/// (legacy daemon) — logged as the literal `"none"` string rather than the
/// field being absent, since presence-vs-absence is itself diagnostic. `db`
/// (the canonical db path) is intentionally not logged here: neither call site
/// has it in scope without broader plumbing, and inventing a value would be
/// worse than omitting the field.
///
/// Log level and counter are graduated by [`FallbackReason::severity`] and
/// [`is_daemon_strict_mode`] (D2-R1): an `Illegitimate` reason observed while
/// `KHIVE_DAEMON_STRICT=1` is set logs at `error!` and bumps
/// `FALLBACK_STRICT_VIOLATIONS` in addition to its own per-reason counter;
/// every other case (non-strict mode, or a `RolloutTransient`/`NoDaemon`
/// reason regardless of mode) logs at `warn!`, exactly as before this change
/// (D2-R3 — strict mode keys on `FallbackReason`, never on "did a fallback
/// happen at all"). The fallback itself is never disabled either way — the
/// caller always proceeds to dispatch locally after this call returns.
fn record_fallback(
    reason: FallbackReason,
    config_id_client: &str,
    config_id_daemon: Option<&str>,
    namespace_client: &str,
) {
    reason
        .counter()
        .fetch_add(1, std::sync::atomic::Ordering::SeqCst);

    let strict_violation =
        reason.severity() == FallbackSeverity::Illegitimate && is_daemon_strict_mode();

    if strict_violation {
        FALLBACK_STRICT_VIOLATIONS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        tracing::error!(
            reason = reason.as_str(),
            config_id_client,
            config_id_daemon = config_id_daemon.unwrap_or("none"),
            namespace_client,
            pid = std::process::id(),
            strict = true,
            "daemon_fallback"
        );
    } else {
        tracing::warn!(
            reason = reason.as_str(),
            config_id_client,
            config_id_daemon = config_id_daemon.unwrap_or("none"),
            namespace_client,
            pid = std::process::id(),
            "daemon_fallback"
        );
    }
}

// ── DaemonDispatch impl ───────────────────────────────────────────────────────

#[async_trait]
impl daemon::DaemonDispatch for crate::server::KhiveMcpServer {
    async fn dispatch(
        &self,
        ops: String,
        presentation: Option<String>,
        presentation_per_op: Option<Vec<Option<String>>>,
        format: Option<String>,
        format_per_op: Option<Vec<Option<String>>>,
        from_wire: bool,
        identity: Option<khive_runtime::RequestIdentity>,
    ) -> Result<String, String> {
        let params = RequestParams {
            ops,
            presentation,
            presentation_per_op,
            save_to: None,
            format,
            format_per_op,
        };
        // Honor the frame's origin: a wire-origin request enforces verb
        // visibility even when served by the daemon; an operator request does not.
        // `identity` (ADR-096 Fork 1) is the caller's per-request identity
        // context, built by `handle_conn` from the frame — threaded straight
        // through so this call serves under the CALLER's namespace/actor
        // rather than this server's own construction-baked identity.
        self.dispatch_request_inner(params, from_wire, identity)
            .await
            .map_err(|e| e.message.to_string())
    }

    async fn warm_all(&self) {
        crate::server::KhiveMcpServer::warm_all(self).await;
    }

    fn namespace(&self) -> &str {
        self.default_namespace()
    }

    fn config_id(&self) -> &str {
        crate::server::KhiveMcpServer::config_id(self)
    }

    fn pool_for_checkpoint(&self) -> Option<std::sync::Arc<khive_db::ConnectionPool>> {
        self.pool()
    }

    fn event_store_for_checkpoint(&self) -> Option<std::sync::Arc<dyn khive_storage::EventStore>> {
        self.event_store()
    }
}

// ── client ────────────────────────────────────────────────────────────────────

/// Result of a single forward attempt to the daemon socket.
enum ForwardOutcome {
    /// Successfully received and decoded a response frame.
    Response(Box<DaemonResponseFrame>),
    /// Socket was unreachable (connection refused / no file).
    NoSocket,
    /// Connected but the response could not be decoded — most likely a stale
    /// daemon speaking a different wire format.
    ParseFailure,
    /// Connected and decoded a response, but the daemon's `daemon_protocol_version`
    /// does not match [`PROTOCOL_VERSION`] even though `version_mismatch` is false.
    /// This is the new-client + old-daemon (pre-versioning) scenario: the old daemon
    /// ignores the unknown request field and returns a decodable response whose
    /// protocol fields default to `false`/`0`. The client must treat this exactly
    /// like `ParseFailure`: kill the stale daemon, respawn once, and return a clear
    /// error if the replacement still has the wrong version.
    ProtocolMismatch,
}

async fn try_forward_inner(frame: &DaemonRequestFrame) -> ForwardOutcome {
    let sock = socket_path();
    let mut stream = match UnixStream::connect(&sock).await {
        Ok(s) => s,
        Err(_) => return ForwardOutcome::NoSocket,
    };
    let payload = match serde_json::to_vec(frame) {
        Ok(p) => p,
        Err(_) => return ForwardOutcome::NoSocket,
    };
    if write_frame(&mut stream, &payload).await.is_err() {
        return ForwardOutcome::NoSocket;
    }
    let resp = match read_frame(&mut stream).await {
        Ok(r) => r,
        Err(e) => {
            // The request was sent but the daemon closed the connection before
            // sending a response frame — likely a daemon crash or panic during
            // dispatch. Treat as ParseFailure (not NoSocket) so the stale daemon
            // is killed and a fresh one is spawned, preventing the caller from
            // silently falling back to local dispatch against a broken daemon.
            tracing::warn!(
                error = %e,
                "daemon closed connection without sending a response \
                 (crash during dispatch?) — treating as stale"
            );
            return ForwardOutcome::ParseFailure;
        }
    };
    match serde_json::from_slice::<DaemonResponseFrame>(&resp) {
        Ok(frame) => {
            // A pre-versioning daemon (old-daemon scenario) returns a decodable
            // response whose `daemon_protocol_version` defaults to 0. The field
            // `version_mismatch` is also false because the old daemon never set
            // it — making `map_response` accept the stale response when
            // `served_config_id` happens to match. Detect this here so the
            // caller can route it through the same kill/respawn path.
            //
            // Also catch the explicit-mismatch / auto-upgrade case (#156): when a
            // warm OLD daemon receives a request from a NEWER client it responds
            // with `version_mismatch=true` and its own (lower) version number.
            // `daemon_protocol_version < PROTOCOL_VERSION` means the daemon is
            // stale — route through kill+respawn exactly like the implicit case
            // above.  If `daemon_protocol_version > PROTOCOL_VERSION` the client
            // binary is behind; kill+respawn cannot fix that, so let map_response
            // return a hard error.
            let is_stale_daemon = frame.daemon_protocol_version != PROTOCOL_VERSION
                && (!frame.version_mismatch || frame.daemon_protocol_version < PROTOCOL_VERSION);
            if is_stale_daemon {
                tracing::warn!(
                    daemon_version = frame.daemon_protocol_version,
                    expected = PROTOCOL_VERSION,
                    explicit_mismatch = frame.version_mismatch,
                    "daemon protocol version mismatch (stale daemon) — treating as stale",
                );
                return ForwardOutcome::ProtocolMismatch;
            }
            ForwardOutcome::Response(Box::new(frame))
        }
        Err(e) => {
            tracing::warn!(
                error = %e,
                bytes = resp.len(),
                "daemon response could not be decoded — stale daemon binary on {}?",
                sock.display()
            );
            ForwardOutcome::ParseFailure
        }
    }
}

fn map_response(
    resp: DaemonResponseFrame,
    expected_config_id: &str,
    namespace_client: &str,
) -> Option<Result<String, McpError>> {
    // Protocol version mismatch is a hard error — do NOT fall back to local
    // dispatch, which would hide the skew. Surface the daemon's own message.
    if resp.version_mismatch {
        let msg = resp.error.unwrap_or_else(|| {
            format!(
                "daemon protocol mismatch: client={} daemon={} — \
                 rebuild/update the client binary (make local)",
                PROTOCOL_VERSION, resp.daemon_protocol_version,
            )
        });
        return Some(Err(McpError::internal_error(msg, None)));
    }

    if resp.namespace_mismatch {
        record_fallback(
            FallbackReason::NamespaceMismatch,
            expected_config_id,
            resp.served_config_id.as_deref(),
            namespace_client,
        );
        return None;
    }
    if resp.config_mismatch {
        record_fallback(
            FallbackReason::ConfigMismatch,
            expected_config_id,
            resp.served_config_id.as_deref(),
            namespace_client,
        );
        return None;
    }
    // Fail closed: only trust a result the daemon positively confirms it served
    // under our exact config. A legacy daemon omits `served_config_id` (→ None)
    // and a config-drifted daemon echoes a different id — both fall back local.
    if resp.served_config_id.as_deref() != Some(expected_config_id) {
        record_fallback(
            FallbackReason::ConfigMismatch,
            expected_config_id,
            resp.served_config_id.as_deref(),
            namespace_client,
        );
        return None;
    }
    if resp.ok {
        Some(Ok(resp.result.unwrap_or_default()))
    } else {
        let msg = resp.error.unwrap_or_else(|| {
            format!(
                "daemon returned an error without a message \
                 (code: internal_error; daemon config: {})",
                resp.served_config_id.as_deref().unwrap_or("unknown"),
            )
        });
        Some(Err(McpError::internal_error(msg, None)))
    }
}

/// Cap on `khived.log` size (bytes) before a spawn rotates it to `khived.log.1`.
///
/// Rotation only happens at spawn time (never mid-session): the daemon is
/// respawned often enough (rebuilds, reconnects, stale-daemon recovery) that
/// this alone bounds disk use, without pulling in `tracing-appender` or
/// touching `init_tracing`'s writer.
const DAEMON_LOG_MAX_BYTES: u64 = 16 * 1024 * 1024;

/// Resolve `<home>/.khive/logs/khived.log` given an explicit `HOME` value.
///
/// Takes the HOME value as a parameter (rather than reading the environment
/// directly) so the resolution logic is unit-testable without mutating
/// process-global state. Mirrors the `Path::new(&home).join(...)` idiom used
/// for `~/.khive/.env` resolution in `kkernel`'s `load_khive_dotenv`.
fn daemon_log_path_from_home(home: Option<&std::ffi::OsStr>) -> Option<std::path::PathBuf> {
    let home = home?;
    Some(
        std::path::Path::new(home)
            .join(".khive")
            .join("logs")
            .join("khived.log"),
    )
}

/// Resolve the daemon log path from the real process environment. Returns
/// `None` when `HOME` is unset — the caller falls back to discarding the
/// daemon's stderr rather than failing the spawn.
fn daemon_log_path() -> Option<std::path::PathBuf> {
    daemon_log_path_from_home(std::env::var_os("HOME").as_deref())
}

/// Decide whether the log at `current_size` bytes must rotate before this
/// spawn, given a `cap` in bytes. Pulled out as a pure function so the
/// spawn-time rotation policy is unit-testable independent of the filesystem.
fn daemon_log_should_rotate(current_size: u64, cap: u64) -> bool {
    current_size >= cap
}

/// Prepare `log_path` for the daemon's stderr: create its parent directory,
/// rotate the existing file to `<name>.1` (replacing any prior backup) if it
/// is at or over `cap` bytes, then open (or create) it for append.
///
/// Returns `None` on directory-creation or open failure so the caller can
/// fall back to `Stdio::null()`. A rotation (`rename`) failure is deliberately
/// swallowed and degrades to appending to the existing over-cap file — keeping
/// the daemon's stderr flowing to a slightly-too-large log beats losing it.
/// Logging is best-effort; daemon spawn correctness is not, and the daemon is
/// on the hot path for every MCP request.
fn prepare_daemon_log_file_with_cap(log_path: &std::path::Path, cap: u64) -> Option<std::fs::File> {
    let dir = log_path.parent()?;
    std::fs::create_dir_all(dir).ok()?;
    if let Ok(meta) = std::fs::metadata(log_path) {
        if daemon_log_should_rotate(meta.len(), cap) {
            let backup = dir.join("khived.log.1");
            // `rename` replaces an existing destination atomically on Unix —
            // exactly the "replace any prior .1" behavior we want.
            let _ = std::fs::rename(log_path, &backup);
        }
    }
    std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(log_path)
        .ok()
}

/// [`prepare_daemon_log_file_with_cap`] using the standing [`DAEMON_LOG_MAX_BYTES`] cap.
fn prepare_daemon_log_file(log_path: &std::path::Path) -> Option<std::fs::File> {
    prepare_daemon_log_file_with_cap(log_path, DAEMON_LOG_MAX_BYTES)
}

fn spawn_daemon() -> std::io::Result<()> {
    #[cfg(test)]
    SPAWN_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

    let exe = std::env::current_exe()?;
    // The binary is `kkernel`; the MCP server (and its daemon mode) live under
    // the `mcp` subcommand.
    let mut cmd = std::process::Command::new(exe);
    cmd.arg("mcp")
        .arg("--daemon")
        .stdin(Stdio::null())
        .stdout(Stdio::null());
    // The daemon's tracing (including WAL/checkpoint telemetry) goes to
    // stderr honoring KHIVE_LOG (init_tracing in kkernel's main.rs) — wiring
    // it to /dev/null silently discards all of it. Route it to a log file
    // instead; fall back to null on any resolution/creation failure so a
    // logging problem never breaks the daemon spawn itself.
    match daemon_log_path().and_then(|path| prepare_daemon_log_file(&path)) {
        Some(file) => {
            cmd.stderr(file);
        }
        None => {
            cmd.stderr(Stdio::null());
        }
    }
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        cmd.process_group(0);
    }
    cmd.spawn()?;
    Ok(())
}

/// Return `true` if `args` (the full `ps -o args=` output for a process)
/// identifies a khive daemon.
///
/// Both conditions must hold:
/// (a) the first whitespace-delimited token's file-name basename is exactly
///     `kkernel` (an absolute path like `/Users/x/.cargo/bin/kkernel` is
///     accepted; a basename of `not-kkernel` or a wrapper whose argv[0] merely
///     mentions kkernel elsewhere is rejected), AND
/// (b) the remaining tokens contain both `mcp` and `--daemon` as distinct
///     whitespace-separated tokens (matching the daemon spawn shape
///     `kkernel mcp --daemon`; a bare `kkernel exec '...'` has no `--daemon`
///     token and is correctly rejected).
fn argv_is_khive_daemon(args: &str) -> bool {
    let mut tokens = args.split_whitespace();
    let Some(exe_token) = tokens.next() else {
        return false;
    };
    // Compare by file-name basename so absolute paths are handled correctly.
    let basename = std::path::Path::new(exe_token)
        .file_name()
        .and_then(|n| n.to_str())
        .unwrap_or("");
    if basename != "kkernel" && basename != "kkernel-bench" {
        return false;
    }
    let rest: Vec<&str> = tokens.collect();
    rest.contains(&"mcp") && rest.contains(&"--daemon")
}

/// Return `true` if the process with the given PID is identifiable as a khive
/// daemon. Uses `ps -p <pid> -o args=` (portable across macOS and Linux) and
/// verifies that (a) the executable basename is exactly `kkernel`, AND (b) the
/// remaining argv tokens include both `mcp` and `--daemon` (the daemon spawn
/// shape). A process that merely mentions "kkernel" in some other argument
/// position is rejected.
///
/// If the process is gone or is a foreign process, returns `false` — the caller
/// must then clean up the stale PID file without sending SIGTERM.
fn pid_is_khive_daemon(pid: u32) -> bool {
    let Ok(pid_i32) = i32::try_from(pid) else {
        return false;
    };
    if pid_i32 <= 0 {
        return false;
    }
    // Test seam: when FORCE_PID_IS_DAEMON is set, treat any positive live PID
    // as a daemon so the SIGTERM branch is reachable in tests.
    #[cfg(test)]
    if FORCE_PID_IS_DAEMON.load(std::sync::atomic::Ordering::SeqCst) {
        // SAFETY: signal 0 is an existence/permission probe with no side effects.
        return unsafe { libc::kill(pid_i32, 0) } == 0;
    }
    // Quick liveness check before shelling out.
    // SAFETY: signal 0 is an existence/permission probe with no side effects.
    if unsafe { libc::kill(pid_i32, 0) } != 0 {
        return false;
    }
    match std::process::Command::new("ps")
        .args(["-p", &pid.to_string(), "-o", "args="])
        .output()
    {
        Ok(out) if out.status.success() => {
            let args = String::from_utf8_lossy(&out.stdout);
            argv_is_khive_daemon(args.trim())
        }
        _ => false,
    }
}

/// Kill the daemon named by the PID file and remove its PID + socket files
/// (caller holds the recovery lock).
///
/// #645: the PID observed here (`expected_pid`) is captured once, before
/// SIGTERM, and re-checked immediately before unlinking in
/// [`remove_daemon_paths_if_still_stale`]. The recovery lock normally
/// serializes this whole function against a concurrent daemon boot (which
/// holds the same lock across its own cleanup → bind → pid-write), but that
/// guarantee depends on the daemon side successfully acquiring the lock too.
/// Re-checking ownership right before deleting is the defense that still
/// holds if a booting daemon ever proceeds without the lock: this call must
/// not delete a replacement daemon's live socket/PID out from under it.
fn kill_stale_daemon_inner() {
    #[cfg(test)]
    KILL_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);

    let pid_file = pid_path();
    let expected_pid = std::fs::read_to_string(&pid_file)
        .ok()
        .and_then(|s| s.trim().parse::<u32>().ok());

    if let Some(pid) = expected_pid {
        if pid_is_khive_daemon(pid) {
            if let Ok(signed) = i32::try_from(pid) {
                if signed > 0 {
                    // SAFETY: SIGTERM is a standard termination signal with no
                    // side effects beyond asking the process to exit.
                    unsafe {
                        libc::kill(signed, libc::SIGTERM);
                    }
                }
            }
        } else {
            tracing::warn!(
                pid,
                "PID in daemon file does not belong to a khive daemon — skipping SIGTERM"
            );
        }
    }

    remove_daemon_paths_if_still_stale(&pid_file, expected_pid);
}

/// Remove `pid_file`/the daemon socket only if ownership has not changed since
/// `expected_pid` was observed: the PID file must still name `expected_pid`,
/// and the socket path must not already have a live listener answering it.
/// Either signal changing means a replacement daemon claimed the rendezvous
/// between the observation and this call, and unlinking would delete its live
/// paths instead of the truly-stale ones (#645).
fn remove_daemon_paths_if_still_stale(pid_file: &std::path::Path, expected_pid: Option<u32>) {
    let current_pid = std::fs::read_to_string(pid_file)
        .ok()
        .and_then(|s| s.trim().parse::<u32>().ok());
    if current_pid != expected_pid {
        tracing::warn!(
            expected_pid = ?expected_pid,
            current_pid = ?current_pid,
            "pid file changed during stale-daemon cleanup — a replacement daemon \
             already claimed it; skipping unlink to avoid deleting its live paths"
        );
        return;
    }

    let sock = socket_path();
    // A plain blocking connect is enough here: any success means *something*
    // is now listening at this path, which can only be a replacement daemon
    // that bound after our probe found the old one dead. Combined with the
    // PID-file recheck above, this closes the window even when the recovery
    // lock alone did not exclude the replacement's boot.
    if std::os::unix::net::UnixStream::connect(&sock).is_ok() {
        tracing::warn!(
            socket = ?sock,
            "a live listener now answers the daemon socket — skipping unlink to \
             avoid deleting a replacement daemon's rendezvous"
        );
        return;
    }

    let _ = std::fs::remove_file(pid_file);
    let _ = std::fs::remove_file(&sock);
}

/// Outcome of the under-lock identity probe.
#[derive(Debug)]
enum ProbeOutcome {
    /// A live, identity-matching daemon responded before the deadline.
    Alive,
    /// Daemon is absent, crashed, or identity-mismatched — safe to kill+spawn.
    Dead,
    /// Probe timed out — daemon may be alive but slow; do NOT kill.
    Timeout,
    /// The boot/recovery lock ([`khive_runtime::daemon::lock_path`]) stayed
    /// contended past its bounded acquisition deadline while
    /// [`quiesce_then_probe_identity`] was trying to confirm no peer boot is
    /// in flight (#838 round-1 Finding 2). Distinct from `Timeout` (which
    /// means "the daemon itself answered slowly") — this means "could not
    /// even confirm whether a peer boot is still running" — so a caller must
    /// not conflate it with a healthy-but-slow daemon. NEVER-KILL on this
    /// outcome either: an unconfirmed peer boot is exactly the ambiguity
    /// `confirm_genuinely_dead` exists to resolve safely.
    LockContended,
}

/// Send a `probe_only` frame to the daemon and return whether a live,
/// identity-matching daemon responded within `timeout_ms` milliseconds.
///
/// Uses `DaemonRequestFrame::probe_only = true` so the daemon returns an
/// identity frame immediately after identity validation — without calling any
/// dispatcher verb, touching the DB, or executing any mutation.
///
/// Probe outcomes → kill decision:
///   `Alive`   → do NOT kill (identity-matching daemon is healthy)
///   `Dead`    → kill+spawn (definitively absent/crashed/mismatched)
///   `Timeout` → do NOT kill (daemon may be healthy-but-busy; NEVER-KILL-SLOW)
async fn probe_daemon_identity(config_id: &str, namespace: &str, timeout_ms: u64) -> ProbeOutcome {
    let probe = DaemonRequestFrame {
        ops: String::new(),
        presentation: None,
        presentation_per_op: None,
        namespace: namespace.to_string(),
        // A probe never reaches the identity-context / dispatch arm (it
        // short-circuits on `probe_only` right after the protocol/config_id
        // checks), so no per-request identity is meaningful here.
        actor_id: None,
        visible_namespaces: Vec::new(),
        config_id: config_id.to_string(),
        protocol_version: PROTOCOL_VERSION,
        probe_only: true,
        metrics_only: false,
        format: None,
        format_per_op: None,
        from_wire: false,
    };
    let deadline = std::time::Duration::from_millis(timeout_ms);
    match tokio::time::timeout(deadline, try_forward_inner(&probe)).await {
        Err(_elapsed) => {
            tracing::debug!(
                timeout_ms,
                "under-lock probe timed out — daemon may be busy; skipping kill"
            );
            ProbeOutcome::Timeout
        }
        Ok(ForwardOutcome::Response(resp)) => {
            // Positive probe-ack confirmation: the response must carry the exact
            // probe-ack sentinel shape (ok=true, result=None, error=None) plus
            // pass all identity checks.
            //
            // Why the sentinel is necessary: `probe_only` is `#[serde(default)]`
            // and the schema has no deny_unknown_fields. A daemon built before this
            // field existed (same protocol version, older binary) deserialises the
            // probe frame without error and falls through to normal dispatch on the
            // empty `ops` string. That produces ok=false (parse error) with matching
            // identity fields — it would be misclassified as Alive by identity
            // checks alone, leaving a stale daemon in place. Requiring the probe-ack
            // shape makes the classifier fail-closed: only the explicit probe branch
            // in handle_conn produces ok=true + result=None + error=None.
            //
            // A normal successful dispatch always sets result=Some(_) (never None).
            // A normal error dispatch sets ok=false. So the sentinel is unambiguous.
            let is_probe_ack = resp.ok && resp.result.is_none() && resp.error.is_none();
            if is_probe_ack
                && !resp.version_mismatch
                && !resp.namespace_mismatch
                && !resp.config_mismatch
                && resp.daemon_protocol_version == PROTOCOL_VERSION
                && resp.served_config_id.as_deref() == Some(config_id)
            {
                tracing::debug!("under-lock probe: live matching daemon confirmed; skipping kill");
                ProbeOutcome::Alive
            } else {
                tracing::debug!(
                    is_probe_ack,
                    version_mismatch = resp.version_mismatch,
                    namespace_mismatch = resp.namespace_mismatch,
                    config_mismatch = resp.config_mismatch,
                    "under-lock probe: daemon did not return probe-ack or identity mismatch — will kill+spawn"
                );
                ProbeOutcome::Dead
            }
        }
        Ok(
            ForwardOutcome::NoSocket
            | ForwardOutcome::ParseFailure
            | ForwardOutcome::ProtocolMismatch,
        ) => ProbeOutcome::Dead,
    }
}

/// Outcome returned by [`kill_and_respawn`] to the call site.
#[derive(Debug)]
enum RecoveryOutcome {
    /// A concurrent client already replaced the daemon; forward the real request
    /// via the normal path (no new spawn occurred).
    Skipped,
    /// This client killed the stale daemon and spawned a replacement; caller
    /// must wait for readiness then forward the real request.
    Spawned,
    /// Could not obtain a positive confirmation either way within the
    /// deadline-bound recovery window (the recoverer lock or the boot/recovery
    /// lock stayed contended past its deadline) — #838 round-1 Finding 2. The
    /// caller's behavior is identical to `Skipped` (never kill on an
    /// unconfirmed state), but this is reported as a distinct variant rather
    /// than silently folded into `Skipped`, so logs/metrics do not conflate
    /// "positively confirmed alive" with "gave up without confirming".
    Uncertain,
}

/// Bounded number of quiescence-confirm rounds [`confirm_genuinely_dead`]
/// performs before trusting a `Dead` classification enough to kill+spawn.
const DEAD_CONFIRM_ROUNDS: u32 = 4;

/// Pacing between [`confirm_genuinely_dead`] rounds. Not a synchronization
/// mechanism by itself — the real synchronization is the bounded `flock` in
/// [`quiesce_then_probe_identity`]; this only avoids busy-spinning while
/// waiting for a peer that has not yet reached its own boot-guard call.
const DEAD_CONFIRM_POLL_MS: u64 = 75;

/// Deadline for each round's bounded wait on the boot/recovery lock inside
/// [`quiesce_then_probe_identity`]. #838 round-1 Finding 2: the previous
/// unbounded blocking `flock` meant `DEAD_CONFIRM_ROUNDS` bounded probe
/// *count*, not elapsed *time* — a wedged lock holder blocked recovery
/// forever. Bounding each round's lock wait makes the whole
/// `confirm_genuinely_dead` call bounded by
/// `DEAD_CONFIRM_ROUNDS * (BOOT_QUIESCENCE_LOCK_TIMEOUT_MS +
/// BOOT_FENCE_PROBE_TIMEOUT_MS + DEAD_CONFIRM_POLL_MS)` in the worst case.
const BOOT_QUIESCENCE_LOCK_TIMEOUT_MS: u64 = 500;

/// Block until no concurrent boot holds the shared boot/recovery lock (or the
/// bounded wait's deadline elapses), then re-probe daemon identity.
///
/// Same mechanism as [`wait_for_boot_quiescence_then_reprobe`], factored out
/// so [`confirm_genuinely_dead`] can reuse it too (#758): a peer's
/// `kill_and_respawn` (kill+spawn) or a daemon's own cold boot both hold this
/// lock, so successfully reacquiring-then-immediately-dropping it proves
/// neither is currently mid-critical-section at the moment this returns.
///
/// #838 round-1 Finding 2: unlike `wait_for_boot_quiescence_then_reprobe`
/// (which genuinely wants to wait out a real boot once one is known to be in
/// flight), this uses the DEADLINE-BOUNDED
/// [`khive_runtime::daemon::try_acquire_daemon_boot_guard_until`] instead of
/// the unbounded blocking `flock` — a wedged lock holder must not block
/// `confirm_genuinely_dead`'s rounds forever. A deadline-elapsed or
/// otherwise-failed acquisition returns the distinct
/// [`ProbeOutcome::LockContended`] rather than collapsing into `Timeout`
/// (which means something different: "the daemon itself answered slowly").
async fn quiesce_then_probe_identity(
    config_id: &str,
    namespace: &str,
    timeout_ms: u64,
) -> ProbeOutcome {
    let deadline = std::time::Instant::now()
        + std::time::Duration::from_millis(BOOT_QUIESCENCE_LOCK_TIMEOUT_MS);
    match tokio::task::spawn_blocking(move || {
        khive_runtime::daemon::try_acquire_daemon_boot_guard_until(deadline)
    })
    .await
    {
        Ok(Ok(Some(guard))) => drop(guard),
        Ok(Ok(None)) => {
            tracing::debug!(
                BOOT_QUIESCENCE_LOCK_TIMEOUT_MS,
                "boot/recovery lock still contended past its bounded wait; \
                 could not confirm quiescence this round"
            );
            return ProbeOutcome::LockContended;
        }
        Ok(Err(e)) => {
            tracing::warn!(error = %e, "failed to probe boot/recovery lock state");
            return ProbeOutcome::LockContended;
        }
        Err(_join_err) => return ProbeOutcome::LockContended,
    }
    probe_daemon_identity(config_id, namespace, timeout_ms).await
}

/// Confirm a `Dead` probe result is not racing a peer's in-flight
/// `kill_and_respawn` or the daemon's own cold boot (#758).
///
/// `spawn_daemon()` is fire-and-forget: `cmd.spawn()` returns as soon as the
/// child process exists, well before that child reaches its own
/// `acquire_daemon_boot_guard()` call. A bare identity probe taken in that
/// gap sees `NoSocket` and is classified `Dead` even though a replacement
/// daemon is legitimately on its way up — a second recoverer trusting that
/// single `Dead` reading would kill nothing useful (the pre-existing #645
/// pid/socket rechecks in [`kill_stale_daemon_inner`] already prevent it
/// from deleting a *bound* replacement's files) but would still spawn a
/// wasteful, storm-prone extra daemon.
///
/// Retries [`quiesce_then_probe_identity`] up to [`DEAD_CONFIRM_ROUNDS`]
/// times, paced by [`DEAD_CONFIRM_POLL_MS`]: returns as soon as a peer's boot
/// is observed completing (`Alive`) or going slow (`Timeout` —
/// NEVER-KILL-SLOW). Only returns `Dead` once every round agrees, which in
/// practice bounds the confirm window well above typical process-start
/// latency (single-digit to low tens of milliseconds) for the fork-to-flock
/// gap this closes.
async fn confirm_genuinely_dead(config_id: &str, namespace: &str) -> ProbeOutcome {
    // `LockContended` means "still could not confirm this round" — the same
    // "keep polling" shape as `Dead`, not a terminal state like `Alive`/
    // `Timeout`. A peer's boot can legitimately hold the lock across several
    // rounds; only give up (return `LockContended`) once every round agreed
    // nobody could confirm, exactly mirroring how `Dead` only becomes trusted
    // once every round agreed the daemon was absent.
    //
    // #838 round-2 Finding 1: `LockContended` is STICKY across rounds — once
    // any round can't confirm quiescence, the aggregate must never collapse
    // back to `Dead` just because a LATER round happened to observe it. The
    // old code tracked only the last round's outcome, so a
    // LockContended-then-Dead sequence overwrote the earlier contention and
    // returned `Dead`, permitting kill+spawn on a call that never actually
    // established quiescence across every round. `Dead` is only trustworthy
    // when EVERY round agrees; a single contended round makes the whole call
    // `LockContended` regardless of what any other round returned.
    let mut saw_contention = false;
    for round in 0..DEAD_CONFIRM_ROUNDS {
        match quiesce_then_probe_identity(config_id, namespace, BOOT_FENCE_PROBE_TIMEOUT_MS).await {
            ProbeOutcome::Dead => {}
            ProbeOutcome::LockContended => saw_contention = true,
            other => return other,
        }
        if round + 1 < DEAD_CONFIRM_ROUNDS {
            tokio::time::sleep(std::time::Duration::from_millis(DEAD_CONFIRM_POLL_MS)).await;
        }
    }
    if saw_contention {
        ProbeOutcome::LockContended
    } else {
        ProbeOutcome::Dead
    }
}

/// Deadline for acquiring the recoverer-only lock
/// ([`khive_runtime::daemon::recoverer_lock_path`]) before starting the
/// dead-confirmation → kill → spawn critical section. Generous enough to
/// cover a peer's full worst-case critical section — up to
/// `DEAD_CONFIRM_ROUNDS * (BOOT_QUIESCENCE_LOCK_TIMEOUT_MS +
/// BOOT_FENCE_PROBE_TIMEOUT_MS + DEAD_CONFIRM_POLL_MS)` for
/// `confirm_genuinely_dead`, plus kill+spawn overhead — so a healthy peer's
/// turn is never mistaken for a wedge.
const RECOVERER_LOCK_TIMEOUT_MS: u64 = 8_000;

/// Kill the stale daemon and spawn a fresh one, serialized against concurrent
/// recoverers by a dedicated recoverer-only lock.
///
/// Implements double-checked recovery: an initial bounded `probe_only` frame
/// (500 ms timeout) is sent under the shared boot/recovery lock
/// ([`acquire_recovery_lock`]) before anything else — this is the same
/// cheap fast-path check as before, letting a client that finds the daemon
/// obviously alive return `Skipped` immediately without ever touching the
/// recoverer lock.
///
/// #838 round-1 Finding 1: a bare `Dead` reading on the initial probe used to
/// fall straight into `confirm_genuinely_dead` → kill → spawn with no
/// linearization point across concurrent recoverers, so two clients racing
/// from a genuinely dead daemon could both classify `Dead` and both spawn a
/// replacement. The fix acquires
/// [`khive_runtime::daemon::try_acquire_recoverer_lock_until`] — a SEPARATE
/// lock file from the daemon's own boot lock — before `confirm_genuinely_dead`
/// runs, and holds it through `kill_stale_daemon_inner` + `spawn_daemon`. This
/// makes recovery mutually exclusive across recoverers without risking a
/// deadlock against a booting daemon: the daemon never acquires this lock
/// (only `kill_and_respawn` does), so nothing on the daemon-boot side is ever
/// waiting on a client holding it. A bounded, deadline-aware acquisition
/// (rather than an unbounded blocking `flock`) is used per Finding 2's
/// same rationale: a second recoverer must not block forever if the first is
/// itself wedged — see `RECOVERER_LOCK_TIMEOUT_MS`.
///
/// Why a second lock file rather than reusing the boot lock for this whole
/// span: the daemon acquires the boot lock (via `acquire_daemon_boot_guard`)
/// as the very first thing it does on boot (see `kkernel::main`), so a client
/// holding that SAME lock across confirm-through-spawn would deadlock every
/// recovery attempt against the very child it just spawned and is waiting to
/// observe. `confirm_genuinely_dead` still uses the boot lock internally
/// (bounded, per-round) purely to detect quiescence — it never holds it
/// across the whole function. The recoverer lock is orthogonal: it excludes
/// PEER RECOVERERS from each other, not from the daemon.
///
/// Outcomes:
///   `Alive`/`Timeout` (initial or confirmed) → `RecoveryOutcome::Skipped`,
///       no kill. NEVER-KILL-SLOW: a timed-out probe means the daemon may be
///       alive but busy, not dead.
///   `LockContended` (confirm rounds could not establish quiescence) or the
///       recoverer lock itself timing out → `RecoveryOutcome::Uncertain`,
///       no kill. Same safe behavior as `Skipped` — the caller forwards the
///       real request either way — but reported distinctly so this state is
///       never silently conflated with a positive "confirmed alive" result.
///   `Dead` (confirmed, recoverer lock held) → kill+spawn;
///       `RecoveryOutcome::Spawned`.
async fn kill_and_respawn(config_id: &str, namespace: &str) -> std::io::Result<RecoveryOutcome> {
    let initial_probe = {
        let _lock = acquire_recovery_lock();
        probe_daemon_identity(config_id, namespace, 500).await
    };
    match initial_probe {
        ProbeOutcome::Alive | ProbeOutcome::Timeout | ProbeOutcome::LockContended => {
            return Ok(RecoveryOutcome::Skipped);
        }
        ProbeOutcome::Dead => {}
    }

    // Test-only rendezvous (see `RECOVERY_RACE_BARRIER`): forces every
    // concurrent recoverer under test to reach "independently classified
    // Dead" at the same instant, so the recoverer lock below is what
    // actually determines mutual exclusion rather than scheduling order.
    #[cfg(test)]
    {
        let barrier = RECOVERY_RACE_BARRIER
            .lock()
            .expect("barrier mutex poisoned")
            .clone();
        if let Some(barrier) = barrier {
            barrier.wait().await;
        }
    }

    let recoverer_deadline =
        std::time::Instant::now() + std::time::Duration::from_millis(RECOVERER_LOCK_TIMEOUT_MS);
    let recoverer_guard = match tokio::time::timeout(
        std::time::Duration::from_millis(RECOVERER_LOCK_TIMEOUT_MS),
        tokio::task::spawn_blocking(move || {
            khive_runtime::daemon::try_acquire_recoverer_lock_until(recoverer_deadline)
        }),
    )
    .await
    {
        Ok(Ok(Ok(Some(guard)))) => guard,
        Ok(Ok(Ok(None))) => {
            tracing::warn!(
                RECOVERER_LOCK_TIMEOUT_MS,
                "recoverer lock still contended past its deadline; a peer recoverer \
                 is likely still mid dead-confirmation/kill/spawn — skipping without \
                 a positive confirmation rather than risking a double-spawn"
            );
            return Ok(RecoveryOutcome::Uncertain);
        }
        Ok(Ok(Err(e))) => {
            tracing::warn!(error = %e, "failed to acquire recoverer lock");
            return Ok(RecoveryOutcome::Uncertain);
        }
        Ok(Err(_)) | Err(_) => {
            tracing::warn!("recoverer lock acquisition task failed or exceeded its deadline");
            return Ok(RecoveryOutcome::Uncertain);
        }
    };

    let outcome = match confirm_genuinely_dead(config_id, namespace).await {
        ProbeOutcome::Alive | ProbeOutcome::Timeout => Ok(RecoveryOutcome::Skipped),
        ProbeOutcome::LockContended => {
            tracing::warn!(
                "confirm_genuinely_dead could not establish quiescence within its \
                 bounded rounds; skipping kill+spawn without a positive confirmation"
            );
            Ok(RecoveryOutcome::Uncertain)
        }
        ProbeOutcome::Dead => {
            // Test-only second rendezvous (see `SPAWN_COMMIT_BARRIER`):
            // bounded wait so two recoverers that both independently
            // classified Dead commit to spawning at the same instant,
            // instead of one's real jitter-driven head start giving the
            // fake-daemon watcher time to save the slower one.
            #[cfg(test)]
            {
                let barrier = SPAWN_COMMIT_BARRIER
                    .lock()
                    .expect("barrier mutex poisoned")
                    .clone();
                if let Some(barrier) = barrier {
                    let _ =
                        tokio::time::timeout(std::time::Duration::from_millis(80), barrier.wait())
                            .await;
                }
            }

            // Also take the shared boot/recovery lock for the kill+spawn step
            // itself, matching `acquire_recovery_lock`'s existing role of
            // serializing this against the daemon server's own
            // cleanup→bind→pid-write critical section. No deadlock risk: this
            // is a distinct lock file from `recoverer_guard` above, acquired
            // and dropped entirely within this arm.
            let _boot_lock = acquire_recovery_lock();
            kill_stale_daemon_inner();
            spawn_daemon().map(|()| RecoveryOutcome::Spawned)
        }
    };
    drop(recoverer_guard);
    outcome
}

/// Build the hard error returned when the real request frame was fully
/// written to the daemon socket but no trustworthy response came back.
///
/// #644: once `write_frame` has completed for a real (non-probe) frame, the
/// daemon may already be dispatching or have finished dispatching it. There is
/// no way from the client side to tell "never received" apart from "received,
/// executed, response lost" — so this case must never be retried (against the
/// same daemon or a freshly-spawned one) and must never silently fall back to
/// local dispatch, either of which could execute a mutation a second time.
fn ambiguous_forward_error() -> McpError {
    McpError::internal_error(
        "daemon response lost after request was sent; not retrying or locally \
         dispatching to avoid duplicate execution",
        None,
    )
}

// ── bridge self-heal: re-exec in place on ProtocolMismatch (#714) ───────────
//
// A long-lived stdio bridge process keeps running the OLD on-disk binary
// after `make local` rebuilds it: the daemon it spawns/forwards to is fresh,
// but this bridge process itself never picks up the new binary until its MCP
// client reconnects. `ProtocolMismatch` is exactly that scenario — the daemon
// is fine, this bridge is stale — so instead of leaving the connection dead
// forever, the bridge re-execs the freshest on-disk binary in place,
// preserving the PID and the open stdio file descriptors so the client's
// transport never sees EOF or a reset. See issue #714 for the evidence this
// design is based on (a live re-exec-mid-session test against the reference
// MCP Python SDK client): a first attempt that called `execv()` synchronously
// inside the tool handler, before the SDK's send loop had serialized and
// flushed the response, discarded the in-flight response and the client's
// call timed out with nothing ever written back.
//
// The fix is a true happens-after edge, not a fixed delay: [`arm_pending_self_heal`]
// records the chosen action *before* the mismatch response is even
// constructed (`trigger_bridge_self_heal` runs synchronously inside the
// request handler), and [`SelfHealOnFlushTransport`] — wrapped around the
// stdio transport in `server.rs::serve_stdio` — fires that action from
// [`fire_pending_self_heal`] only once a message has actually finished
// flushing to the client. Because arming always happens before the mismatch
// response is handed to the transport, and firing only ever happens after a
// flush completes, the very next successful flush is guaranteed to be at or
// after that response reached the client — never before, and never on a
// clock that can expire while a slow or backpressured stdout is still
// mid-write (a fixed-duration sleep could not make that guarantee: rmcp's
// send pipeline enqueues the response and returns almost immediately, then
// performs the real write+flush on a separately spawned task with no
// duration bound).

/// Pending self-heal action, armed by [`arm_pending_self_heal`] and taken by
/// [`fire_pending_self_heal`]. `None` on a healthy bridge for its entire
/// lifetime — the overwhelmingly common case.
static PENDING_SELF_HEAL: std::sync::Mutex<Option<MismatchRecovery>> = std::sync::Mutex::new(None);

fn arm_pending_self_heal(action: MismatchRecovery) {
    let mut slot = PENDING_SELF_HEAL
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    *slot = Some(action);
}

/// Take and perform whatever self-heal action is armed, if any. Called by
/// [`SelfHealOnFlushTransport::send`] after every message it successfully
/// flushes to the client — the load-bearing happens-after edge documented
/// above. A no-op when nothing is armed.
#[cfg(unix)]
pub(crate) fn fire_pending_self_heal() {
    let action = PENDING_SELF_HEAL
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .take();
    match action {
        Some(MismatchRecovery::ReexecScheduled) => reexec_in_place(),
        Some(MismatchRecovery::DrainAndExit) => exit_process(),
        None => {}
    }
}

/// Re-exec self-heal requires `exec()` (POSIX-only) — [`schedule_reexec_on_mismatch`]'s
/// non-unix variant arms [`MismatchRecovery::DrainAndExit`] instead of
/// [`MismatchRecovery::ReexecScheduled`], so only the drain-and-exit arm is
/// ever actually armed on this target.
#[cfg(not(unix))]
pub(crate) fn fire_pending_self_heal() {
    let action = PENDING_SELF_HEAL
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .take();
    if action.is_some() {
        exit_process();
    }
}

/// argv marker carrying the exec-once loop-breaker generation counter.
///
/// Parsed here and defined on [`crate::args::Args`] as a hidden clap field of
/// the same name — the clap field exists purely so the CLI parser accepts the
/// flag on a resumed process instead of rejecting it as unknown; the actual
/// read path is this raw argv scan, independent of wherever in the call stack
/// `Args` was parsed. The counter travels with the exec by construction: it
/// is appended to argv immediately before `exec()`, so any process running
/// with it present is, by definition, a resumed generation.
const RESUMED_GENERATION_ARG_PREFIX: &str = "--resumed-generation=";

/// Whether this process is a resumed generation of a prior self-heal re-exec,
/// and if so, its generation counter. `None` on a normal (cold-started)
/// bridge — the overwhelmingly common case.
pub(crate) fn resumed_generation() -> Option<u32> {
    resumed_generation_from_args(std::env::args())
}

/// Pure argv-scan behind [`resumed_generation`], factored out so the parsing
/// logic is unit-testable without depending on this process's own real argv
/// (which never carries the marker inside `cargo test`).
fn resumed_generation_from_args(args: impl Iterator<Item = String>) -> Option<u32> {
    args.filter_map(|a| {
        a.strip_prefix(RESUMED_GENERATION_ARG_PREFIX)
            .map(str::to_owned)
    })
    .last()
    .and_then(|s| s.parse::<u32>().ok())
}

/// Recovery action chosen for a `ProtocolMismatch` observed inside
/// `forward_or_spawn`. Pure decision, factored out of [`trigger_bridge_self_heal`]
/// so the loop-breaker guard rail (#714 §2.2: exec at most once per mismatch
/// generation) is unit-testable without touching the process or the clock.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum MismatchRecovery {
    /// First generation (no `--resumed-generation` marker) — schedule an
    /// in-place re-exec of the freshest on-disk binary.
    ReexecScheduled,
    /// Already a resumed generation and it hit `ProtocolMismatch` again — the
    /// on-disk binary is itself stale, or a second rebuild race — take the
    /// fallback instead of exec'ing a second time.
    DrainAndExit,
}

fn decide_mismatch_recovery(resumed_generation: Option<u32>) -> MismatchRecovery {
    match resumed_generation {
        None => MismatchRecovery::ReexecScheduled,
        Some(_) => MismatchRecovery::DrainAndExit,
    }
}

/// Trigger the bridge's self-heal recovery for a `ProtocolMismatch` outcome.
///
/// Called from both `forward_or_spawn` `ProtocolMismatch` arms (the
/// first-attempt arm and the arm inside the post-recovery retry loop — either
/// can observe the mismatch depending on whether this was the first probe or
/// a retry after a kill/respawn). Both arms already construct and return the
/// hard mismatch error to the caller; this call schedules the recovery
/// alongside that return, never in place of it.
///
/// Concurrency note (#714 design discussion, accepted risk, not fixed by
/// this change): if the bridge is mid-flight on more than one outstanding
/// client request when the mismatch fires, only the request that triggered
/// this arm gets the ambiguous-error-then-resume treatment. Any other
/// in-flight request loses its response the same way it would today if the
/// process crashed — a pre-existing risk, not introduced here. Relatedly,
/// [`fire_pending_self_heal`] fires on the next successful flush of *any*
/// message, not specifically the mismatch response's own flush — on this
/// bridge's dominant single-request-at-a-time usage those are the same
/// event, but a genuinely concurrent second in-flight request could in
/// principle flush first. That race is strictly better than the pre-fix
/// timer (which could fire before *any* flush completed at all) and is the
/// same class of pre-existing concurrency risk noted above, not a new one.
fn trigger_bridge_self_heal() {
    match decide_mismatch_recovery(resumed_generation()) {
        MismatchRecovery::ReexecScheduled => schedule_reexec_on_mismatch(),
        MismatchRecovery::DrainAndExit => {
            tracing::warn!(
                "resumed generation observed ProtocolMismatch again — loop-breaker \
                 tripped (#714 §2.2, exec-once guard); draining and exiting instead \
                 of re-exec'ing a second time"
            );
            schedule_drain_and_exit();
        }
    }
}

/// Arm an in-place re-exec of the freshest on-disk binary, to fire from
/// [`fire_pending_self_heal`] once the mismatch response has actually
/// flushed (see the module-level doc above for why that happens-after edge
/// — not a fixed delay — is load-bearing). Never execs synchronously and
/// never execs from this function directly. Only ever called for a
/// first-generation process — the loop-breaker guard rail lives in
/// [`trigger_bridge_self_heal`].
#[cfg(unix)]
pub(crate) fn schedule_reexec_on_mismatch() {
    tracing::warn!(
        client_version = PROTOCOL_VERSION,
        "protocol mismatch: arming in-place re-exec of the freshest on-disk binary, \
         to fire once the mismatch response has flushed to the client"
    );
    arm_pending_self_heal(MismatchRecovery::ReexecScheduled);
}

/// Re-exec self-heal requires `exec()` (POSIX-only); on any other target,
/// take the same drain-and-exit fallback a loop-breaker trip would.
#[cfg(not(unix))]
pub(crate) fn schedule_reexec_on_mismatch() {
    schedule_drain_and_exit();
}

/// Perform the actual re-exec: resolve the on-disk binary at *exec time* via
/// [`std::env::current_exe`] (the same primitive `spawn_daemon` already uses
/// for "pick up whatever `make local` just replaced" — see `spawn_daemon`
/// above), preserve the original argv, append the `--resumed-generation=1`
/// marker, and replace the process image via
/// [`std::os::unix::process::CommandExt::exec`] — which, unlike `spawn`,
/// keeps the same PID and the same open stdin/stdout/stderr file descriptors,
/// so the client's stdio transport never sees EOF or a reset.
///
/// `exec` only returns on failure; on failure this logs and returns, leaving
/// the process running under its stale binary (the hard mismatch error was
/// already sent to the client for this request; there is nothing safe to
/// retry from here).
#[cfg(all(unix, not(test)))]
fn reexec_in_place() {
    use std::os::unix::process::CommandExt;

    let exe = match std::env::current_exe() {
        Ok(p) => p,
        Err(e) => {
            tracing::error!(error = %e, "bridge self-heal re-exec failed: could not resolve current_exe");
            return;
        }
    };
    // Drop any pre-existing marker defensively (should never be present here —
    // `trigger_bridge_self_heal` only reaches this path for a first-generation
    // process — but argv should never accumulate duplicates if that invariant
    // is ever violated).
    let args: Vec<String> = std::env::args()
        .skip(1)
        .filter(|a| !a.starts_with(RESUMED_GENERATION_ARG_PREFIX))
        .chain(std::iter::once(format!("{RESUMED_GENERATION_ARG_PREFIX}1")))
        .collect();
    let err = std::process::Command::new(exe).args(&args).exec();
    tracing::error!(
        error = %err,
        "bridge self-heal re-exec failed; continuing under the stale binary"
    );
}

/// Arm this process to stop serving and exit, to fire from
/// [`fire_pending_self_heal`] for the same happens-after reason
/// [`schedule_reexec_on_mismatch`] arms its exec instead of performing it
/// directly. This is the fallback path (issue #714 §4): the MCP connection
/// dies and the client's own process-lifecycle management must restart it —
/// no worse than the pre-#714 hard-error-forever behavior.
pub(crate) fn schedule_drain_and_exit() {
    arm_pending_self_heal(MismatchRecovery::DrainAndExit);
}

#[cfg(not(test))]
fn exit_process() {
    std::process::exit(1);
}

#[cfg(all(test, unix))]
pub(crate) static REEXEC_INVOKED_COUNT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);
#[cfg(test)]
pub(crate) static DRAIN_EXIT_INVOKED_COUNT: std::sync::atomic::AtomicUsize =
    std::sync::atomic::AtomicUsize::new(0);

#[cfg(all(test, unix))]
pub(crate) fn reset_self_heal_counters() {
    REEXEC_INVOKED_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
    DRAIN_EXIT_INVOKED_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
    clear_pending_self_heal();
}

#[cfg(all(test, not(unix)))]
pub(crate) fn reset_self_heal_counters() {
    DRAIN_EXIT_INVOKED_COUNT.store(0, std::sync::atomic::Ordering::SeqCst);
    clear_pending_self_heal();
}

/// Clear whatever `PENDING_SELF_HEAL` currently holds (#843). A test that
/// exercises `forward_or_spawn`'s protocol-mismatch path end to end (e.g.
/// `forward_or_spawn_rejects_old_daemon_and_returns_protocol_mismatch_error`)
/// arms it via `trigger_bridge_self_heal` but never takes the action back
/// out — that is `fire_pending_self_heal`'s job, and asserting the forwarding
/// behavior alone has no reason to call it. Left armed, that leftover slot
/// is invisible to `reset_self_heal_counters` resetting only the two
/// invocation counters, so a later test in the same binary (e.g.
/// `fire_pending_self_heal_is_a_no_op_when_nothing_is_armed`) can inherit it
/// under multi-threaded test ordering and observe a spurious fire. Every
/// existing test that intentionally arms the slot does so AFTER calling
/// `reset_self_heal_counters`, never before, so clearing it here changes no
/// test's semantics.
#[cfg(test)]
fn clear_pending_self_heal() {
    *PENDING_SELF_HEAL
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
}

/// Test double for [`reexec_in_place`]: a real `exec()` would replace the test
/// binary's own process image, killing the entire test run. Counts instead.
/// Gated `unix` like the production version above — it is the only thing that
/// calls it (`schedule_reexec_on_mismatch`'s `not(unix)` arm never reaches
/// `reexec_in_place` at all).
#[cfg(all(test, unix))]
fn reexec_in_place() {
    REEXEC_INVOKED_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}

/// Wraps a [`rmcp::transport::Transport`] so every message it successfully
/// flushes to the client fires [`fire_pending_self_heal`] afterward.
///
/// This is the actual happens-after edge #714's self-heal design requires,
/// not the fixed-duration sleep the first version of this change used.
/// [`rmcp::transport::async_rw::AsyncRwTransport::send`] — the concrete
/// transport `KhiveMcpServer::serve_stdio` wraps this around — resolves its
/// returned future only once the message has been encoded and flushed to
/// the underlying writer (`FramedWrite::send` is `futures::SinkExt::send`,
/// i.e. `feed` + `flush`). `rmcp`'s own service loop enqueues a tool
/// handler's response and returns almost immediately, then performs that
/// real write+flush on a separately spawned task with no duration bound —
/// so the handler itself has no way to await it directly. Wrapping the
/// transport instead intercepts every flush completion regardless of which
/// task drives it.
pub(crate) struct SelfHealOnFlushTransport<T> {
    inner: T,
}

impl<T> SelfHealOnFlushTransport<T> {
    pub(crate) fn new(inner: T) -> Self {
        Self { inner }
    }
}

impl<T> rmcp::transport::Transport<rmcp::RoleServer> for SelfHealOnFlushTransport<T>
where
    T: rmcp::transport::Transport<rmcp::RoleServer>,
{
    type Error = T::Error;

    fn send(
        &mut self,
        item: rmcp::service::TxJsonRpcMessage<rmcp::RoleServer>,
    ) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send + 'static {
        let send = self.inner.send(item);
        async move {
            let result = send.await;
            if result.is_ok() {
                fire_pending_self_heal();
            }
            result
        }
    }

    fn receive(
        &mut self,
    ) -> impl std::future::Future<Output = Option<rmcp::service::RxJsonRpcMessage<rmcp::RoleServer>>>
           + Send {
        self.inner.receive()
    }

    fn close(&mut self) -> impl std::future::Future<Output = Result<(), Self::Error>> + Send {
        self.inner.close()
    }
}

#[cfg(test)]
fn exit_process() {
    DRAIN_EXIT_INVOKED_COUNT.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}

/// Bounded probe timeout used by [`wait_for_boot_quiescence_then_reprobe`],
/// matching the 500ms bound already used by the identity probe inside
/// [`kill_and_respawn`].
const BOOT_FENCE_PROBE_TIMEOUT_MS: u64 = 500;

/// Outcome of waiting for a concurrent cold-boot (migrations + pack schema
/// plans / FTS DDL) to finish, then re-checking daemon liveness once quiescent.
enum BootFenceOutcome {
    /// Boot quiesced and a live, identity-matching daemon answered — keep
    /// sending the real frame to it.
    DaemonReady,
    /// Boot quiesced and the probe definitively found no daemon — safe to
    /// fall back to local dispatch.
    SafeLocalFallback,
    /// Daemon state is unknown (lock/join failure, or the post-quiescence
    /// probe itself timed out) — must not local-dispatch.
    HardError(McpError),
}

/// #667: the readiness-timeout branch of `forward_or_spawn`'s send loop used
/// to return `None` (silent local fallback) purely because a freshly spawned
/// daemon had not answered within the fixed deadline — including while that
/// daemon was still inside its boot guard running migrations/pack schema
/// plans (FTS DDL). A local writer/searcher racing in at exactly that moment
/// could observe or create a partially-initialized `notes`/`fts_notes` schema.
///
/// This blocks on the SAME boot guard the daemon holds across cold-boot
/// schema init (ADR-D3): acquiring it here can only succeed once no boot is
/// in progress, so acquiring-then-immediately-dropping it is a pure
/// quiescence wait. Only after that wait does it re-probe daemon identity —
/// distinguishing "was still booting, now ready" from "genuinely no daemon"
/// so the caller never has to guess which one caused the original timeout.
async fn wait_for_boot_quiescence_then_reprobe(frame: &DaemonRequestFrame) -> BootFenceOutcome {
    // `acquire_daemon_boot_guard` performs a blocking `flock`; run it on the
    // blocking pool rather than the async executor.
    let quiesced =
        tokio::task::spawn_blocking(khive_runtime::daemon::acquire_daemon_boot_guard).await;
    match quiesced {
        Ok(Ok(guard)) => {
            // The guard's only purpose here is proving quiescence; drop it
            // immediately so it does not itself block a real boot or the
            // re-probe below.
            drop(guard);
        }
        Ok(Err(e)) => {
            return BootFenceOutcome::HardError(McpError::internal_error(
                format!(
                    "failed to acquire daemon boot/recovery lock while waiting for \
                     cold-boot quiescence: {e}"
                ),
                None,
            ));
        }
        Err(e) => {
            return BootFenceOutcome::HardError(McpError::internal_error(
                format!("boot-quiescence wait task failed: {e}"),
                None,
            ));
        }
    }

    match probe_daemon_identity(
        &frame.config_id,
        &frame.namespace,
        BOOT_FENCE_PROBE_TIMEOUT_MS,
    )
    .await
    {
        ProbeOutcome::Alive => BootFenceOutcome::DaemonReady,
        ProbeOutcome::Dead => BootFenceOutcome::SafeLocalFallback,
        ProbeOutcome::Timeout => BootFenceOutcome::HardError(McpError::internal_error(
            "daemon state uncertain after cold-boot quiescence; not falling back to \
             local dispatch to avoid racing a possibly still-initializing index",
            None,
        )),
        // `probe_daemon_identity` (unlike `quiesce_then_probe_identity`) never
        // constructs `LockContended` — it has no lock-acquisition step of its
        // own. Handled here only for match exhaustiveness over the shared
        // `ProbeOutcome` type; same fail-safe HardError as `Timeout` if it
        // were ever reached.
        ProbeOutcome::LockContended => BootFenceOutcome::HardError(McpError::internal_error(
            "daemon state uncertain after cold-boot quiescence (lock probe unexpectedly \
             contended); not falling back to local dispatch",
            None,
        )),
    }
}

/// Forward a request to the daemon, auto-spawning it if absent.
///
/// Returns `None` only when nothing was ever written to the daemon and local
/// dispatch is therefore safe: `KHIVE_NO_DAEMON` is set, or no daemon socket
/// could be reached (`NoSocket`). It never returns `None` after the real frame
/// has been written. `Some(Ok)` / `Some(Err)` both mean the request's fate is
/// already decided at the daemon and the caller must not dispatch locally.
///
/// The real (possibly mutating) request frame is written to the daemon socket
/// at most once per call. A `NoSocket` outcome never writes anything, so it is
/// safe to establish or recover the daemon (via `kill_and_respawn`, which
/// itself uses only `probe_only` frames) and retry the connect-and-write step
/// until a response arrives or the readiness deadline elapses.
///
/// Once the real frame IS fully written — `ForwardOutcome::ParseFailure` or
/// `ForwardOutcome::ProtocolMismatch` — this returns a hard error immediately
/// (`ambiguous_forward_error`) instead of killing/respawning/retrying or
/// falling back locally. See #644.
pub async fn forward_or_spawn(frame: &DaemonRequestFrame) -> Option<Result<String, McpError>> {
    if env_truthy("KHIVE_NO_DAEMON") {
        return None;
    }

    match try_forward_inner(frame).await {
        ForwardOutcome::Response(resp) => {
            return map_response(*resp, &frame.config_id, &frame.namespace)
        }
        ForwardOutcome::NoSocket => {
            // Nothing was written; fall through to the spawn/recover-then-send
            // path below.
        }
        ForwardOutcome::ParseFailure => {
            tracing::warn!(
                config_id = %frame.config_id,
                namespace = %frame.namespace,
                retry_suppressed = true,
                "daemon connection lost after the request was fully written — \
                 not retrying or falling back locally to avoid duplicate dispatch"
            );
            return Some(Err(ambiguous_forward_error()));
        }
        ForwardOutcome::ProtocolMismatch => {
            tracing::warn!(
                config_id = %frame.config_id,
                namespace = %frame.namespace,
                retry_suppressed = true,
                "daemon protocol mismatch discovered after the request was fully \
                 written — not retrying or falling back locally to avoid duplicate dispatch"
            );
            // #714: the daemon is fine (it just rejected us) — this bridge
            // process itself is the stale one. Trigger self-heal alongside
            // the hard error below, never in place of it.
            trigger_bridge_self_heal();
            return Some(Err(McpError::internal_error(
                format!(
                    "daemon protocol mismatch: expected version {PROTOCOL_VERSION}; \
                     run `make local` to rebuild the daemon binary"
                ),
                None,
            )));
        }
    }

    // NoSocket: nothing has been written yet. Establish a live daemon — either
    // this is the first-ever spawn or a stale one needs replacing — under the
    // single recovery lock, using only `probe_only` frames for the identity
    // check. `kill_and_respawn` is a no-op kill when there is nothing stale to
    // remove, so first-spawn and recovery share this one path.
    match kill_and_respawn(&frame.config_id, &frame.namespace).await {
        Err(e) => {
            tracing::warn!(error = %e, "failed to spawn/recover the daemon; falling back to local dispatch");
            record_fallback(
                FallbackReason::NoSocket,
                &frame.config_id,
                None,
                &frame.namespace,
            );
            return None;
        }
        Ok(RecoveryOutcome::Skipped) => {
            // A concurrent client already has a live matching daemon ready.
        }
        Ok(RecoveryOutcome::Spawned) => {
            // Give the kernel a moment to release the socket path and let the
            // spawned daemon process start.
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        Ok(RecoveryOutcome::Uncertain) => {
            // Could not positively confirm the daemon's state within the
            // deadline (#838 round-1 Finding 2) — behave like `Skipped` (never
            // kill on an unconfirmed state) and let the forward loop below
            // discover the real state: it will either reach a daemon a peer
            // is spawning, or hit the readiness deadline and fall through to
            // `wait_for_boot_quiescence_then_reprobe`.
            tracing::debug!(
                "daemon recovery state uncertain; forwarding without a fresh kill+spawn"
            );
        }
    }

    // Send the real frame exactly once now that a daemon is confirmed ready
    // (or believed ready via Skipped). The connect attempt inside
    // `try_forward_inner` doubles as the readiness check — a `NoSocket`
    // outcome here just means "not listening yet" (nothing written), so keep
    // retrying; any other outcome is terminal and returned immediately.
    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        if tokio::time::Instant::now() >= deadline {
            // #667: a bare timeout here does not mean "no daemon" — it may
            // mean "daemon is still inside cold-boot schema init". Wait for
            // that boot (if any) to quiesce and re-probe before deciding.
            match wait_for_boot_quiescence_then_reprobe(frame).await {
                BootFenceOutcome::DaemonReady => {
                    // The real frame still has not been written; fall through
                    // and send it now that boot has quiesced.
                }
                BootFenceOutcome::SafeLocalFallback => {
                    record_fallback(
                        FallbackReason::NoSocket,
                        &frame.config_id,
                        None,
                        &frame.namespace,
                    );
                    return None;
                }
                BootFenceOutcome::HardError(err) => return Some(Err(err)),
            }
        }
        match try_forward_inner(frame).await {
            ForwardOutcome::Response(resp) => {
                return map_response(*resp, &frame.config_id, &frame.namespace)
            }
            ForwardOutcome::ParseFailure => {
                tracing::warn!(
                    config_id = %frame.config_id,
                    namespace = %frame.namespace,
                    retry_suppressed = true,
                    "freshly-established daemon connection lost after the request \
                     was fully written — not retrying or falling back locally"
                );
                return Some(Err(ambiguous_forward_error()));
            }
            ForwardOutcome::ProtocolMismatch => {
                tracing::warn!(
                    config_id = %frame.config_id,
                    namespace = %frame.namespace,
                    "daemon protocol mismatch discovered on the post-recovery retry \
                     — not retrying again or falling back locally"
                );
                // #714: same self-heal trigger as the first-attempt arm above —
                // either arm can observe the mismatch depending on whether this
                // was the first probe or a retry after a kill/respawn.
                trigger_bridge_self_heal();
                return Some(Err(McpError::internal_error(
                    format!(
                        "daemon protocol mismatch: expected version {PROTOCOL_VERSION}; \
                         run `make local` to rebuild the daemon binary"
                    ),
                    None,
                )));
            }
            ForwardOutcome::NoSocket => {
                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            }
        }
    }
}

// ── tests ─────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use khive_runtime::daemon::run_daemon;
    use serial_test::serial;

    use khive_runtime::engine_config::ActorConfig;
    use khive_runtime::{
        runtime_config_from_khive_config, KhiveConfig, KhiveRuntime, Namespace, RuntimeConfig,
    };

    fn memory_runtime_config() -> RuntimeConfig {
        KhiveRuntime::memory()
            .expect("memory runtime")
            .config()
            .clone()
    }

    fn make_test_server() -> crate::server::KhiveMcpServer {
        let mut config = memory_runtime_config();
        config.default_namespace = Namespace::parse("test").unwrap();
        config.packs = vec!["kg".to_string(), "gtd".to_string()];
        let runtime = KhiveRuntime::new(config).expect("in-memory runtime");
        crate::server::KhiveMcpServer::new(runtime).expect("server builds with kg+gtd")
    }

    /// Server whose pack set includes `brain`, which registers
    /// `Visibility::Subhandler` verbs (`brain.state`, …). Used to exercise the
    /// wire visibility gate through the daemon round-trip.
    fn make_subhandler_test_server() -> crate::server::KhiveMcpServer {
        let mut config = memory_runtime_config();
        config.default_namespace = Namespace::parse("braintest").unwrap();
        config.packs = vec!["kg".to_string(), "brain".to_string()];
        let runtime = KhiveRuntime::new(config).expect("in-memory runtime");
        crate::server::KhiveMcpServer::new(runtime).expect("server builds with kg+brain")
    }

    /// Server whose pack set includes `comm`, whose `send` verb echoes the
    /// dispatching actor directly in its JSON response (`"from": from_actor`,
    /// where `from_actor = token.actor().id`). Used to verify per-request
    /// actor stamping (ADR-096 Fork 1) without a readback/inbox round-trip.
    fn make_comm_test_server(actor_id: Option<&str>) -> crate::server::KhiveMcpServer {
        let mut config = memory_runtime_config();
        config.default_namespace = Namespace::parse("test").unwrap();
        config.packs = vec!["kg".to_string(), "comm".to_string()];
        config.actor_id = actor_id.map(str::to_string);
        let runtime = KhiveRuntime::new(config).expect("in-memory runtime");
        crate::server::KhiveMcpServer::new(runtime).expect("server builds with kg+comm")
    }

    fn folded_actor_memory_config(actor: &str) -> RuntimeConfig {
        runtime_config_from_khive_config(
            &KhiveConfig {
                actor: ActorConfig {
                    id: Some(actor.to_string()),
                    ..ActorConfig::default()
                },
                ..KhiveConfig::default()
            },
            memory_runtime_config(),
        )
    }

    fn clear_daemon_env() {
        std::env::remove_var("KHIVE_SOCKET");
        std::env::remove_var("KHIVE_PID");
        std::env::remove_var("KHIVE_NO_DAEMON");
        std::env::remove_var("KHIVE_LOCK");
        std::env::remove_var("KHIVE_RECOVERER_LOCK");
    }

    async fn connect_when_ready(sock: &std::path::Path) -> UnixStream {
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
        loop {
            if let Ok(s) = UnixStream::connect(sock).await {
                return s;
            }
            assert!(
                tokio::time::Instant::now() < deadline,
                "daemon never bound {sock:?} within 5s"
            );
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    }

    async fn exchange(sock: &std::path::Path, frame: &DaemonRequestFrame) -> DaemonResponseFrame {
        let mut stream = UnixStream::connect(sock)
            .await
            .expect("connect to daemon socket");
        let payload = serde_json::to_vec(frame).expect("serialize request frame");
        write_frame(&mut stream, &payload)
            .await
            .expect("write request frame");
        let resp = read_frame(&mut stream).await.expect("read response frame");
        serde_json::from_slice(&resp).expect("decode response frame")
    }

    // ── map_response (pure, MCP-specific) ─────────────────────────────────────

    const CFG: &str = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";
    const NS: &str = "test";

    fn frame_ok(result: &str) -> DaemonResponseFrame {
        DaemonResponseFrame {
            ok: true,
            result: Some(result.to_string()),
            error: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(CFG.to_string()),
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        }
    }

    fn frame_err(error: Option<&str>) -> DaemonResponseFrame {
        DaemonResponseFrame {
            ok: false,
            result: None,
            error: error.map(str::to_string),
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(CFG.to_string()),
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        }
    }

    // These four tests exercise `map_response` branches that now also increment
    // the process-global fallback counters (via `record_fallback`) — `#[serial]`
    // + a reset at the top keeps their counter assertions deterministic against
    // any other test in this file that touches the same counters.

    #[test]
    #[serial]
    fn map_response_namespace_mismatch_yields_none() {
        reset_fallback_counters();
        let resp = DaemonResponseFrame {
            ok: false,
            result: None,
            error: None,
            namespace_mismatch: true,
            config_mismatch: false,
            served_config_id: Some(CFG.to_string()),
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        };
        assert!(map_response(resp, CFG, NS).is_none());
        assert_eq!(fallback_count(FallbackReason::NamespaceMismatch), 1);
        assert_eq!(fallback_count(FallbackReason::ConfigMismatch), 0);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    #[serial]
    fn map_response_config_mismatch_yields_none() {
        reset_fallback_counters();
        let resp = DaemonResponseFrame {
            ok: false,
            result: None,
            error: None,
            namespace_mismatch: false,
            config_mismatch: true,
            served_config_id: Some(CFG.to_string()),
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        };
        assert!(map_response(resp, CFG, NS).is_none());
        assert_eq!(fallback_count(FallbackReason::ConfigMismatch), 1);
        assert_eq!(fallback_count(FallbackReason::NamespaceMismatch), 0);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    #[serial]
    fn map_response_legacy_daemon_missing_echo_yields_none() {
        // A pre-config_id daemon omits served_config_id (→ None). Even on an
        // ok=true result the client MUST fall back to local dispatch.
        reset_fallback_counters();
        let resp = DaemonResponseFrame {
            ok: true,
            result: Some("served-by-broad-registry".to_string()),
            error: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: None,
            version_mismatch: false,
            daemon_protocol_version: 0,
            metrics: None,
        };
        assert!(map_response(resp, CFG, NS).is_none());
        // The served_config_id-echo path is bucketed under config_mismatch —
        // there is no separate reason for "echo missing/drifted" in the closed
        // 5-value reason set.
        assert_eq!(fallback_count(FallbackReason::ConfigMismatch), 1);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    #[serial]
    fn map_response_echo_drift_yields_none() {
        // A daemon serving under a different config (echo != expected) is not
        // trusted, even without an explicit config_mismatch flag.
        reset_fallback_counters();
        let resp = DaemonResponseFrame {
            ok: true,
            result: Some("served-by-other-config".to_string()),
            error: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(
                "packs=[kg,gtd];db=/x;embed=none;extra=[];backend=main".to_string(),
            ),
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        };
        assert!(map_response(resp, CFG, NS).is_none());
        assert_eq!(fallback_count(FallbackReason::ConfigMismatch), 1);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    fn map_response_ok_with_result_yields_some_ok() {
        match map_response(frame_ok("the-result"), CFG, NS) {
            Some(Ok(s)) => assert_eq!(s, "the-result"),
            other => panic!("expected Some(Ok(\"the-result\")), got {other:?}"),
        }
    }

    #[test]
    fn map_response_ok_with_no_result_yields_some_ok_empty() {
        let resp = DaemonResponseFrame {
            ok: true,
            result: None,
            error: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(CFG.to_string()),
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        };
        match map_response(resp, CFG, NS) {
            Some(Ok(s)) => assert_eq!(s, ""),
            other => panic!("expected Some(Ok(\"\")), got {other:?}"),
        }
    }

    #[test]
    fn map_response_not_ok_yields_some_err_preserving_message() {
        match map_response(frame_err(Some("boom: bad verb")), CFG, NS) {
            Some(Err(McpError { message, .. })) => {
                assert!(message.contains("boom: bad verb"), "got: {message}");
            }
            other => panic!("expected Some(Err(..)), got {other:?}"),
        }
    }

    #[test]
    fn map_response_not_ok_without_message_yields_contextual_err() {
        match map_response(frame_err(None), CFG, NS) {
            Some(Err(McpError { message, .. })) => {
                assert!(!message.is_empty(), "fallback message must not be empty");
                assert!(
                    message.contains("daemon returned an error"),
                    "fallback must say 'daemon returned an error'; got: {message}"
                );
            }
            other => panic!("expected Some(Err(..)), got {other:?}"),
        }
    }

    #[test]
    fn map_response_version_mismatch_yields_explicit_error() {
        let resp = DaemonResponseFrame {
            ok: false,
            result: None,
            error: Some("daemon protocol mismatch: client=0 daemon=1 — rebuild/update the client binary (make local)".to_string()),
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(CFG.to_string()),
            version_mismatch: true,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        };
        match map_response(resp, CFG, NS) {
            Some(Err(McpError { message, .. })) => {
                assert!(
                    message.contains("protocol mismatch"),
                    "version mismatch error must name the mismatch; got: {message}"
                );
                assert!(
                    message.contains("make local"),
                    "version mismatch error must tell the operator what to do; got: {message}"
                );
            }
            other => panic!("expected Some(Err(..)): got {other:?}"),
        }
    }

    #[test]
    fn map_response_version_mismatch_without_error_field_synthesizes_message() {
        let resp = DaemonResponseFrame {
            ok: false,
            result: None,
            error: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(CFG.to_string()),
            version_mismatch: true,
            daemon_protocol_version: 99,
            metrics: None,
        };
        match map_response(resp, CFG, NS) {
            Some(Err(McpError { message, .. })) => {
                assert!(
                    message.contains("protocol mismatch"),
                    "synthesized message must name the mismatch; got: {message}"
                );
                assert!(
                    message.contains("99"),
                    "synthesized message must include daemon version; got: {message}"
                );
            }
            other => panic!("expected Some(Err(..)): got {other:?}"),
        }
    }

    // ── daemon_fallback telemetry: counters (ADR-091 F1) ──────────────────────
    //
    // `no_socket` / `parse_failure` / `protocol_mismatch` are only reachable
    // inside `forward_or_spawn` on paths that require a real daemon subprocess
    // or multi-second connect timeouts (the existing suite deliberately avoids
    // that cost elsewhere in this file — see `try_forward_inner_returns_parse_
    // failure_when_daemon_closes_without_response` above, which asserts the
    // `ForwardOutcome` discriminant directly rather than driving the full
    // `forward_or_spawn` retry loop). `record_fallback` is the single function
    // every one of those call sites invokes, so exercising it directly gives
    // the same counter-correctness guarantee without the slow paths.

    #[test]
    #[serial]
    fn record_fallback_no_socket_increments_matching_counter_and_total() {
        reset_fallback_counters();
        record_fallback(FallbackReason::NoSocket, CFG, None, NS);
        assert_eq!(fallback_count(FallbackReason::NoSocket), 1);
        assert_eq!(fallback_count(FallbackReason::ParseFailure), 0);
        assert_eq!(fallback_count(FallbackReason::ProtocolMismatch), 0);
        assert_eq!(fallback_count(FallbackReason::ConfigMismatch), 0);
        assert_eq!(fallback_count(FallbackReason::NamespaceMismatch), 0);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    #[serial]
    fn record_fallback_parse_failure_increments_matching_counter_and_total() {
        reset_fallback_counters();
        record_fallback(FallbackReason::ParseFailure, CFG, None, NS);
        assert_eq!(fallback_count(FallbackReason::ParseFailure), 1);
        assert_eq!(fallback_count(FallbackReason::NoSocket), 0);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    #[serial]
    fn record_fallback_protocol_mismatch_increments_matching_counter_and_total() {
        reset_fallback_counters();
        record_fallback(FallbackReason::ProtocolMismatch, CFG, None, NS);
        assert_eq!(fallback_count(FallbackReason::ProtocolMismatch), 1);
        assert_eq!(fallback_count(FallbackReason::ParseFailure), 0);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    #[serial]
    fn record_fallback_config_id_daemon_defaults_to_none_literal_when_absent() {
        // The log field itself isn't asserted (no tracing-capture harness in this
        // crate — see the crate's Cargo.toml dev-dependencies), but the counter
        // side effect proves the call completes and increments exactly once even
        // when `config_id_daemon` is `None` (the common case at every
        // forward_or_spawn fallback site, since no decodable daemon response was
        // available to read a served config id from).
        reset_fallback_counters();
        record_fallback(FallbackReason::NoSocket, CFG, None, NS);
        assert_eq!(fallback_total(), 1);
    }

    #[test]
    #[serial]
    fn fallback_total_sums_all_reason_counters() {
        reset_fallback_counters();
        record_fallback(FallbackReason::ConfigMismatch, CFG, Some("other-cfg"), NS);
        record_fallback(
            FallbackReason::NamespaceMismatch,
            CFG,
            Some(CFG),
            "other-ns",
        );
        record_fallback(FallbackReason::NoSocket, CFG, None, NS);
        record_fallback(FallbackReason::ParseFailure, CFG, None, NS);
        record_fallback(FallbackReason::ProtocolMismatch, CFG, None, NS);

        let sum = fallback_count(FallbackReason::ConfigMismatch)
            + fallback_count(FallbackReason::NamespaceMismatch)
            + fallback_count(FallbackReason::NoSocket)
            + fallback_count(FallbackReason::ParseFailure)
            + fallback_count(FallbackReason::ProtocolMismatch);
        assert_eq!(sum, 5);
        assert_eq!(
            fallback_total(),
            5,
            "total must equal the sum of all reasons"
        );
    }

    // ── KHIVE_DAEMON_STRICT graduated fail-loud policy (D2) ───────────────────
    //
    // No tracing-capture harness exists in this crate (see the comment on
    // `record_fallback_config_id_daemon_defaults_to_none_literal_when_absent`
    // above), so these tests prove the graduated behavior via
    // `FALLBACK_STRICT_VIOLATIONS` rather than asserting the log level
    // directly: an `Illegitimate` reason bumps it if and only if strict mode
    // is on; every other reason/mode combination must never bump it.

    fn with_daemon_strict<T>(value: Option<&str>, f: impl FnOnce() -> T) -> T {
        let prev = std::env::var("KHIVE_DAEMON_STRICT").ok();
        match value {
            Some(v) => std::env::set_var("KHIVE_DAEMON_STRICT", v),
            None => std::env::remove_var("KHIVE_DAEMON_STRICT"),
        }
        let result = f();
        match prev {
            Some(v) => std::env::set_var("KHIVE_DAEMON_STRICT", v),
            None => std::env::remove_var("KHIVE_DAEMON_STRICT"),
        }
        result
    }

    #[test]
    #[serial]
    fn record_fallback_config_mismatch_strict_off_never_bumps_strict_violations() {
        with_daemon_strict(None, || {
            reset_fallback_counters();
            record_fallback(FallbackReason::ConfigMismatch, CFG, Some("other-cfg"), NS);
            assert_eq!(fallback_count(FallbackReason::ConfigMismatch), 1);
            assert_eq!(
                fallback_strict_violations(),
                0,
                "strict mode is OFF (default local-dev behavior) — the illegitimate \
                 reason must still bump its own counter, but never the strict-violations one"
            );
        });
    }

    #[test]
    #[serial]
    fn record_fallback_config_mismatch_strict_on_bumps_strict_violations() {
        with_daemon_strict(Some("1"), || {
            reset_fallback_counters();
            record_fallback(FallbackReason::ConfigMismatch, CFG, Some("other-cfg"), NS);
            assert_eq!(fallback_count(FallbackReason::ConfigMismatch), 1);
            assert_eq!(
                fallback_strict_violations(),
                1,
                "KHIVE_DAEMON_STRICT=1 + an Illegitimate reason (config_mismatch) must \
                 bump the strict-violations counter (D2-R1)"
            );
        });
    }

    #[test]
    #[serial]
    fn record_fallback_namespace_mismatch_strict_on_bumps_strict_violations() {
        with_daemon_strict(Some("1"), || {
            reset_fallback_counters();
            record_fallback(
                FallbackReason::NamespaceMismatch,
                CFG,
                Some(CFG),
                "other-ns",
            );
            assert_eq!(
                fallback_strict_violations(),
                1,
                "KHIVE_DAEMON_STRICT=1 + an Illegitimate reason (namespace_mismatch) must \
                 bump the strict-violations counter (D2-R1)"
            );
        });
    }

    #[test]
    #[serial]
    fn record_fallback_no_socket_strict_on_never_bumps_strict_violations() {
        with_daemon_strict(Some("1"), || {
            reset_fallback_counters();
            record_fallback(FallbackReason::NoSocket, CFG, None, NS);
            assert_eq!(
                fallback_strict_violations(),
                0,
                "NoSocket is the ADR-049-mandated no-daemon path — it must NEVER be \
                 elevated, even in strict mode (D2-R3)"
            );
        });
    }

    #[test]
    #[serial]
    fn record_fallback_protocol_mismatch_strict_on_never_bumps_strict_violations() {
        with_daemon_strict(Some("1"), || {
            reset_fallback_counters();
            record_fallback(FallbackReason::ProtocolMismatch, CFG, None, NS);
            assert_eq!(
                fallback_strict_violations(),
                0,
                "ProtocolMismatch is the rollout-transient (version_mismatch) tier — \
                 it must NEVER be elevated, even in strict mode (D2-R3)"
            );
        });
    }

    #[test]
    #[serial]
    fn record_fallback_parse_failure_strict_on_never_bumps_strict_violations() {
        with_daemon_strict(Some("1"), || {
            reset_fallback_counters();
            record_fallback(FallbackReason::ParseFailure, CFG, None, NS);
            assert_eq!(
                fallback_strict_violations(),
                0,
                "ParseFailure is folded into the rollout-transient tier alongside \
                 ProtocolMismatch (see FallbackReason::severity) — never elevated"
            );
        });
    }

    #[test]
    fn fallback_reason_severity_matches_the_d2_legitimacy_table() {
        assert_eq!(
            FallbackReason::ConfigMismatch.severity(),
            FallbackSeverity::Illegitimate
        );
        assert_eq!(
            FallbackReason::NamespaceMismatch.severity(),
            FallbackSeverity::Illegitimate
        );
        assert_eq!(
            FallbackReason::ProtocolMismatch.severity(),
            FallbackSeverity::RolloutTransient
        );
        assert_eq!(
            FallbackReason::ParseFailure.severity(),
            FallbackSeverity::RolloutTransient
        );
        assert_eq!(
            FallbackReason::NoSocket.severity(),
            FallbackSeverity::NoDaemon
        );
    }

    // ── forward_or_spawn fallback (env-mutating → serial) ─────────────────────

    #[tokio::test]
    #[serial]
    async fn forward_or_spawn_returns_none_when_no_daemon_set() {
        clear_daemon_env();
        reset_fallback_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", dir.path().join("khived.pid"));
        std::env::set_var("KHIVE_NO_DAEMON", "1");

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: "test".to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };
        let out = forward_or_spawn(&frame).await;
        assert!(out.is_none());
        assert!(!sock.exists());
        // KHIVE_NO_DAEMON is an explicit operator opt-out, not one of the 5
        // silent-fallback reasons this telemetry tracks — it must NOT bump the
        // fallback counters (that would be noisy for legitimate always-local
        // deployments).
        assert_eq!(fallback_total(), 0);

        clear_daemon_env();
    }

    // ── daemon socket round-trip (env-mutating → serial) ─────────────────────

    #[tokio::test]
    #[serial]
    async fn daemon_round_trip_dispatches_and_enforces_config_id() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid = dir.path().join("khived.pid");
        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let reference = make_test_server();
        let config_id = reference.config_id().to_string();
        let daemon_server = reference.clone();

        let handle = tokio::spawn(async move {
            let _ = run_daemon(daemon_server).await;
        });

        let _ready = connect_when_ready(&sock).await;
        drop(_ready);

        // (a) valid same-namespace, same-config op
        let req = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: Some("verbose".to_string()),
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.clone(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };
        let resp = exchange(&sock, &req).await;
        assert!(resp.ok, "valid op must succeed; error={:?}", resp.error);
        assert!(!resp.namespace_mismatch);
        assert!(!resp.config_mismatch);
        assert!(!resp.version_mismatch);
        assert_eq!(resp.daemon_protocol_version, PROTOCOL_VERSION);
        assert_eq!(
            resp.served_config_id.as_deref(),
            Some(config_id.as_str()),
            "daemon must echo the config it served under"
        );

        let reference_result = reference
            .dispatch_request_local(RequestParams {
                ops: "stats()".to_string(),
                presentation: Some("verbose".to_string()),
                presentation_per_op: None,
                save_to: None,
                format: None,
                format_per_op: None,
            })
            .await
            .expect("local dispatch of stats() must succeed");
        assert_eq!(resp.result.as_deref(), Some(reference_result.as_str()));
        assert!(reference_result.contains("\"entities\""));

        // (b) ADR-096 Fork 1: a different namespace, same config_id, is no
        // longer rejected — the daemon accepts and serves the request under
        // the frame's OWN namespace ("other") over the same shared warm
        // registry, instead of setting `namespace_mismatch`.
        let other = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "other".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.clone(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };
        let resp_other = exchange(&sock, &other).await;
        assert!(
            resp_other.ok,
            "a differently-namespaced frame with a matching config_id must be \
             served, not rejected; error={:?}",
            resp_other.error
        );
        assert!(
            !resp_other.namespace_mismatch,
            "ADR-096 Fork 1 removed the namespace_mismatch reject"
        );
        assert!(!resp_other.config_mismatch);
        assert_eq!(
            resp_other.served_config_id.as_deref(),
            Some(config_id.as_str())
        );

        // (c) same namespace but different config (e.g. a `--pack kg` client
        // hitting the broader daemon) → config_mismatch, no dispatch. The
        // config_id reject stays hard under ADR-096 Fork 1 — only the
        // namespace reject was softened.
        let mismatched_config = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: "packs=[kg];db=:memory:;embed=none;extra=[];backend=main".to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };
        let resp_cfg = exchange(&sock, &mismatched_config).await;
        assert!(
            resp_cfg.config_mismatch,
            "differing config must be rejected"
        );
        assert!(!resp_cfg.namespace_mismatch);
        assert!(!resp_cfg.ok);

        // (d) version mismatch → explicit error, NOT namespace/config mismatch
        let wrong_version = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.clone(),
            protocol_version: 0,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };
        let resp_ver = exchange(&sock, &wrong_version).await;
        assert!(
            resp_ver.version_mismatch,
            "wrong protocol version must set version_mismatch"
        );
        assert!(!resp_ver.ok);
        assert!(
            resp_ver
                .error
                .as_deref()
                .unwrap_or("")
                .contains("protocol mismatch"),
            "version mismatch error must include 'protocol mismatch'; got: {:?}",
            resp_ver.error
        );
        assert!(
            resp_ver
                .error
                .as_deref()
                .unwrap_or("")
                .contains("make local"),
            "version mismatch error must tell operator what to do; got: {:?}",
            resp_ver.error
        );
        assert_eq!(
            resp_ver.daemon_protocol_version, PROTOCOL_VERSION,
            "daemon must echo its own protocol version in the mismatch response"
        );

        handle.abort();
        let _ = handle.await;
        clear_daemon_env();
    }

    // ── ADR-096 Fork 1: per-request identity over one warm registry ──────────
    //
    // Core capability test: two requests carrying DIFFERENT frame namespaces
    // AND DIFFERENT frame actors, dispatched against ONE already-running warm
    // daemon, must both be served over the shared backend — and each write
    // must be stamped with its OWN frame's actor, never the other request's
    // and never a daemon-baked default. `comm.send` is the vehicle: its JSON
    // response echoes the dispatching actor directly (`"from": from_actor`),
    // so this needs no readback/inbox round-trip.
    #[tokio::test]
    #[serial]
    async fn daemon_serves_per_request_identity_over_one_warm_registry() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid = dir.path().join("khived.pid");
        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid);
        std::env::remove_var("KHIVE_NO_DAEMON");

        // No baked actor_id on the daemon-side server: every actor must come
        // from the per-request frame, never a construction-time default.
        let reference = make_comm_test_server(None);
        let config_id = reference.config_id().to_string();
        let daemon_server = reference.clone();

        let handle = tokio::spawn(async move {
            let _ = run_daemon(daemon_server).await;
        });
        let _ready = connect_when_ready(&sock).await;
        drop(_ready);

        let alice_frame = DaemonRequestFrame {
            ops: "comm.send(to=\"bob\", content=\"hello from alice\")".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "alpha".to_string(),
            actor_id: Some("alice".to_string()),
            visible_namespaces: Vec::new(),
            config_id: config_id.clone(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };
        let bob_frame = DaemonRequestFrame {
            ops: "comm.send(to=\"alice\", content=\"hello from bob\")".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "beta".to_string(),
            actor_id: Some("bob".to_string()),
            visible_namespaces: Vec::new(),
            config_id: config_id.clone(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let resp_alice = exchange(&sock, &alice_frame).await;
        assert!(
            resp_alice.ok,
            "alice's request must be served over the shared warm registry; error={:?}",
            resp_alice.error
        );
        assert!(!resp_alice.namespace_mismatch);
        assert!(!resp_alice.config_mismatch);
        let body_alice: serde_json::Value =
            serde_json::from_str(resp_alice.result.as_deref().expect("alice result body"))
                .expect("decode alice result json");
        assert_eq!(
            body_alice["results"][0]["result"]["from"], "alice",
            "write dispatched under alice's frame must stamp actor=alice, got: {body_alice}"
        );

        let resp_bob = exchange(&sock, &bob_frame).await;
        assert!(
            resp_bob.ok,
            "bob's request must be served over the SAME shared warm registry; error={:?}",
            resp_bob.error
        );
        assert!(!resp_bob.namespace_mismatch);
        assert!(!resp_bob.config_mismatch);
        let body_bob: serde_json::Value =
            serde_json::from_str(resp_bob.result.as_deref().expect("bob result body"))
                .expect("decode bob result json");
        assert_eq!(
            body_bob["results"][0]["result"]["from"], "bob",
            "write dispatched under bob's frame must stamp actor=bob, NOT cross-\
             contaminated with alice's actor; got: {body_bob}"
        );

        handle.abort();
        let _ = handle.await;
        clear_daemon_env();
    }

    // ADR-096 Fork 1 completion: actor-derived visible namespaces are request
    // identity, not daemon engine identity. Two clients with different
    // configured actors therefore compute the same config_id, but each daemon
    // frame must still read only through its own visible set.
    #[tokio::test]
    #[serial]
    async fn daemon_config_id_ignores_actor_folded_visibility_but_frame_visibility_isolated() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid = dir.path().join("khived.pid");
        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let actor_a = "lambda:actor-a";
        let actor_b = "lambda:actor-b";
        let cfg_a = folded_actor_memory_config(actor_a);
        let cfg_b = folded_actor_memory_config(actor_b);
        let ns_a = Namespace::parse(actor_a).expect("actor a namespace");
        let ns_b = Namespace::parse(actor_b).expect("actor b namespace");

        assert_eq!(cfg_a.actor_id.as_deref(), Some(actor_a));
        assert_eq!(cfg_b.actor_id.as_deref(), Some(actor_b));
        assert!(
            cfg_a.visible_namespaces.contains(&ns_a),
            "actor.id must fold into client A visible_namespaces"
        );
        assert!(
            cfg_b.visible_namespaces.contains(&ns_b),
            "actor.id must fold into client B visible_namespaces"
        );
        assert_ne!(
            cfg_a.visible_namespaces, cfg_b.visible_namespaces,
            "precondition: clients must carry different folded visible sets"
        );

        let id_a = crate::server::compute_config_id(&cfg_a, None);
        let id_b = crate::server::compute_config_id(&cfg_b, None);
        assert_eq!(
            id_a, id_b,
            "actor-derived visible_namespaces must not affect daemon config_id"
        );

        let daemon_server = {
            let runtime = KhiveRuntime::new(memory_runtime_config()).expect("in-memory runtime");
            crate::server::KhiveMcpServer::new(runtime).expect("server builds with kg")
        };
        assert_eq!(
            daemon_server.config_id(),
            id_a,
            "daemon and both clients must share the same engine-coherence key"
        );
        let handle = tokio::spawn(async move {
            let _ = run_daemon(daemon_server).await;
        });
        let _ready = connect_when_ready(&sock).await;
        drop(_ready);

        let frame = |ops: &str, actor: &str, visible: &[Namespace]| DaemonRequestFrame {
            ops: ops.to_string(),
            presentation: Some("verbose".to_string()),
            presentation_per_op: None,
            namespace: "local".to_string(),
            actor_id: Some(actor.to_string()),
            visible_namespaces: visible.iter().map(|ns| ns.as_str().to_string()).collect(),
            config_id: id_a.clone(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let seed_a = exchange(
            &sock,
            &frame(
                r#"create(kind="concept", name="ActorAVisibleOnly", namespace="lambda:actor-a")"#,
                actor_a,
                &cfg_a.visible_namespaces,
            ),
        )
        .await;
        assert!(seed_a.ok, "seed A must succeed: {:?}", seed_a.error);
        let seed_b = exchange(
            &sock,
            &frame(
                r#"create(kind="concept", name="ActorBVisibleOnly", namespace="lambda:actor-b")"#,
                actor_b,
                &cfg_b.visible_namespaces,
            ),
        )
        .await;
        assert!(seed_b.ok, "seed B must succeed: {:?}", seed_b.error);

        fn names_from_list_response(resp: &DaemonResponseFrame) -> Vec<String> {
            assert!(resp.ok, "list response must be ok: {:?}", resp.error);
            assert!(
                !resp.config_mismatch,
                "list response must not reject on config_id"
            );
            let body: serde_json::Value =
                serde_json::from_str(resp.result.as_deref().expect("list result body"))
                    .expect("decode list result json");
            let first = &body["results"][0];
            assert_eq!(
                first["ok"], true,
                "list op must succeed inside daemon result: {first}"
            );
            let rows = first["result"]
                .as_array()
                .or_else(|| first["result"]["items"].as_array())
                .expect("list result must be an array or object with items");
            rows.iter()
                .filter_map(|row| row.get("name").and_then(|v| v.as_str()).map(str::to_string))
                .collect()
        }

        let list_a = exchange(
            &sock,
            &frame(r#"list(kind="entity")"#, actor_a, &cfg_a.visible_namespaces),
        )
        .await;
        let names_a = names_from_list_response(&list_a);
        assert!(
            names_a.iter().any(|name| name == "ActorAVisibleOnly"),
            "actor A frame must see actor A namespace rows; got {names_a:?}"
        );
        assert!(
            !names_a.iter().any(|name| name == "ActorBVisibleOnly"),
            "actor A frame must not see actor B namespace rows; got {names_a:?}"
        );

        let list_b = exchange(
            &sock,
            &frame(r#"list(kind="entity")"#, actor_b, &cfg_b.visible_namespaces),
        )
        .await;
        let names_b = names_from_list_response(&list_b);
        assert!(
            names_b.iter().any(|name| name == "ActorBVisibleOnly"),
            "actor B frame must see actor B namespace rows; got {names_b:?}"
        );
        assert!(
            !names_b.iter().any(|name| name == "ActorAVisibleOnly"),
            "actor B frame must not see actor A namespace rows; got {names_b:?}"
        );

        handle.abort();
        let _ = handle.await;
        clear_daemon_env();
    }

    // Back-compat: a caller with NO per-request identity context (pure local
    // dispatch, never touching the daemon socket) must keep using the
    // server's own construction-baked actor_id, unaffected by the identity-
    // override machinery introduced for daemon-forwarded requests.
    #[tokio::test]
    #[serial]
    async fn local_dispatch_without_identity_context_uses_baked_actor() {
        clear_daemon_env();

        let server = make_comm_test_server(Some("baked-actor"));
        let result = server
            .dispatch_request_local(RequestParams {
                ops: "comm.send(to=\"someone\", content=\"hello\")".to_string(),
                presentation: None,
                presentation_per_op: None,
                save_to: None,
                format: None,
                format_per_op: None,
            })
            .await
            .expect("local dispatch of comm.send must succeed");

        let body: serde_json::Value =
            serde_json::from_str(&result).expect("decode local dispatch result json");
        assert_eq!(
            body["results"][0]["result"]["from"], "baked-actor",
            "local dispatch (no daemon, no identity context) must use the server's \
             own baked actor_id, got: {body}"
        );
    }

    // ── daemon-forward wire-origin gate (security regression) ────────────────
    //
    // The agent-facing MCP `request` tool sets `from_wire=true` on its daemon
    // frame; the daemon must HONOR that bit so the subhandler visibility gate
    // fires after the socket round-trip, not only on the local-fallback path.
    // The local-fallback tests (in tests/integration.rs) run with the daemon
    // disabled and would stay green even if the MCP frame were flipped to
    // `from_wire=false` — which would open an agent-reachable subhandler bypass
    // on every daemon-backed deployment. This pins the round-trip seam.
    #[tokio::test]
    #[serial]
    async fn daemon_round_trip_honors_from_wire_for_subhandlers() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid = dir.path().join("khived.pid");
        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let reference = make_subhandler_test_server();
        let config_id = reference.config_id().to_string();
        let daemon_server = reference.clone();
        let handle = tokio::spawn(async move {
            let _ = run_daemon(daemon_server).await;
        });
        let _ready = connect_when_ready(&sock).await;
        drop(_ready);

        let frame = |from_wire: bool| DaemonRequestFrame {
            ops: "brain.state()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "braintest".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.clone(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire,
        };

        // (a) from_wire=true → daemon applies the wire visibility gate:
        // `brain.state` is a Subhandler and must be blocked after the round-trip.
        let resp_wire = exchange(&sock, &frame(true)).await;
        assert!(
            resp_wire.ok,
            "dispatch itself must succeed (the op carries the gate error); error={:?}",
            resp_wire.error
        );
        let body_wire: serde_json::Value =
            serde_json::from_str(resp_wire.result.as_deref().expect("wire result body"))
                .expect("decode wire result json");
        let first_wire = &body_wire["results"][0];
        assert_eq!(
            first_wire["ok"], false,
            "from_wire=true subhandler must be blocked through the daemon: {first_wire}"
        );
        let err_wire = first_wire["error"].as_str().unwrap_or("");
        assert!(
            err_wire.contains("permission denied") || err_wire.contains("subhandler"),
            "daemon-forward wire path must surface the subhandler gate error; got: {err_wire}"
        );

        // (b) from_wire=false (operator frame, e.g. `kkernel exec`) → the same
        // Subhandler verb must be REACHED, not gated.
        let resp_op = exchange(&sock, &frame(false)).await;
        let body_op: serde_json::Value =
            serde_json::from_str(resp_op.result.as_deref().expect("operator result body"))
                .expect("decode operator result json");
        let first_op = &body_op["results"][0];
        let err_op = first_op["error"].as_str().unwrap_or("");
        assert!(
            !err_op.contains("permission denied") && !err_op.contains("subhandler"),
            "operator frame must NOT gate the subhandler through the daemon: {first_op}"
        );

        handle.abort();
        let _ = handle.await;
        clear_daemon_env();
    }

    // The agent-facing `request` tool must stamp `from_wire=true` on its daemon
    // forward-frame. Without this, a daemon-backed MCP request would dispatch
    // with `from_wire=false` and silently reopen the agent subhandler bypass
    // (review #369 Medium). Pinned at the frame-builder seam so the daemon
    // round-trip test above (which proves the daemon HONORS the bit) is paired
    // with proof that the tool SETS it.
    #[test]
    fn wire_request_frame_sets_from_wire_true() {
        let server = make_subhandler_test_server();
        let params = RequestParams {
            ops: "brain.state()".to_string(),
            ..Default::default()
        };
        let frame = server.wire_daemon_frame(&params);
        assert!(
            frame.from_wire,
            "request tool must set from_wire=true on the daemon forward-frame"
        );
        assert_eq!(frame.ops, "brain.state()");
        assert_eq!(frame.namespace, "braintest");
    }

    // ── new-client + old-daemon regression (fix for #98 BLOCKER) ─────────────
    //
    // Simulates a pre-versioning daemon that:
    //   • Ignores the unknown `protocol_version` request field (it deserializes
    //     as missing and is silently dropped).
    //   • Returns a decodable response with a matching `served_config_id` but
    //     WITHOUT `daemon_protocol_version` or `version_mismatch` (they default
    //     to `0` / `false` on the client).
    //
    // Before the fix, `map_response` accepted this response because `version_mismatch`
    // was false and `served_config_id` matched — the stale daemon was trusted.
    //
    // After the fix, `try_forward_inner` detects `daemon_protocol_version == 0 != 1`
    // and returns `ForwardOutcome::ProtocolMismatch`, which `forward_or_spawn` routes
    // through kill/respawn/error rather than accepting.
    //
    // The test verifies at the `forward_or_spawn` level (not just `map_response`)
    // that:
    //   1. The stale response is NOT accepted.
    //   2. At most one recovery attempt is made (the fake socket closes after one
    //      exchange; a second attempt sees NoSocket rather than looping).
    //   3. A clear "protocol mismatch" error is returned.

    /// Minimal "old daemon" frame: has `served_config_id` but omits the
    /// protocol version fields (they default to `false`/`0` on the client side).
    fn old_daemon_response(config_id: &str) -> DaemonResponseFrame {
        DaemonResponseFrame {
            ok: true,
            result: Some("stale-result".to_string()),
            error: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(config_id.to_string()),
            // Pre-versioning daemon would never set these:
            version_mismatch: false,
            daemon_protocol_version: 0,
            metrics: None,
        }
    }

    /// Serve exactly one connection with `response`, then stop accepting.
    async fn serve_one_response(listener: tokio::net::UnixListener, response: DaemonResponseFrame) {
        if let Ok((mut stream, _)) = listener.accept().await {
            // Read the inbound request frame (and discard it — old daemon ignores
            // unknown fields, which is the scenario we're simulating).
            if read_frame(&mut stream).await.is_ok() {
                if let Ok(payload) = serde_json::to_vec(&response) {
                    let _ = write_frame(&mut stream, &payload).await;
                }
            }
        }
        // Listener drops here; subsequent connection attempts see "connection refused".
    }

    #[tokio::test]
    #[serial]
    async fn forward_or_spawn_rejects_old_daemon_and_returns_protocol_mismatch_error() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        // Bind the fake old-daemon socket BEFORE starting the client.
        let listener = tokio::net::UnixListener::bind(&sock).expect("bind fake old-daemon socket");
        // Write a placeholder PID file so kill_stale_daemon finds a PID to look at.
        // We use the current process's PID, which pid_is_khive_daemon() will reject
        // because the current exe is the test binary (not "kkernel"), so no SIGTERM
        // will be sent — this is the safe path we want to exercise.
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");

        let old_resp = old_daemon_response(config_id);
        // Serve exactly one exchange, then let the listener drop (no second connection
        // will be served — this enforces the "at most one recovery attempt" constraint:
        // if forward_or_spawn tried the old socket a second time it would get NoSocket).
        let fake_handle = tokio::spawn(serve_one_response(listener, old_resp));

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let result = forward_or_spawn(&frame).await;

        // The fake socket served exactly one old-protocol response.  The fake handle
        // should have completed by now; join it to catch any panics.
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), fake_handle).await;

        // Must NOT accept the stale daemon result.
        match result {
            Some(Err(McpError { message, .. })) => {
                assert!(
                    message.contains("protocol mismatch"),
                    "error must name 'protocol mismatch'; got: {message}"
                );
                assert!(
                    message.contains("make local") || message.contains("rebuild"),
                    "error must tell the operator what to do; got: {message}"
                );
            }
            Some(Ok(v)) => {
                panic!("forward_or_spawn must NOT accept old-daemon response; got Ok({v:?})")
            }
            None => panic!(
                "forward_or_spawn must return Some(Err(..)) for protocol mismatch, \
                 not None (which would cause silent fallback to local dispatch)"
            ),
        }

        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── daemon crash mid-dispatch regression (#91) ────────────────────────────
    //
    // When the daemon crashes (or panics) during dispatch it closes the
    // connection without writing a response frame. Before the fix, `try_forward_inner`
    // returned `ForwardOutcome::NoSocket` on the `read_frame` error, causing
    // `forward_or_spawn` to return `None` — a silent fallback to local dispatch.
    //
    // The fix promotes the `read_frame` error to `ForwardOutcome::ParseFailure`
    // so the stale daemon is killed and the client does NOT silently accept a
    // potentially broken daemon state for subsequent requests.
    //
    // This test binds a fake socket that reads the request but immediately drops
    // the connection without writing a response (simulating a crash during
    // dispatch), then asserts that `forward_or_spawn` falls back to `None`
    // (because the respawn attempt also sees no socket) with the WARN log
    // rather than silently accepting the empty response.
    //
    // We cannot assert an `Err` here because after killing the stale daemon the
    // respawn path calls `spawn_daemon()` (which tries to exec the real binary),
    // fails or times out, and returns `None`. The critical invariant is that
    // `NoSocket` is NOT returned directly on read failure — the `ParseFailure`
    // path is taken instead (logging the crash and killing the stale process).
    // The `try_forward_inner` unit test below validates the exact `ForwardOutcome`
    // discriminant.

    /// Serve one connection: read the request frame, then drop the stream
    /// without writing any response (simulating a daemon crash mid-dispatch).
    async fn serve_crash_on_dispatch(listener: tokio::net::UnixListener) {
        if let Ok((mut stream, _)) = listener.accept().await {
            // Read the inbound request (and discard it — the "crash" happens here).
            let _ = read_frame(&mut stream).await;
            // Drop stream without writing a response — connection resets.
        }
        // Listener drops; subsequent connection attempts see "connection refused".
    }

    #[tokio::test]
    #[serial]
    async fn try_forward_inner_returns_parse_failure_when_daemon_closes_without_response() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        let listener =
            tokio::net::UnixListener::bind(&sock).expect("bind fake crash-daemon socket");
        // Write a placeholder PID file with the current process PID so
        // kill_stale_daemon has something to look at (it will skip SIGTERM
        // because the exe basename is not "kkernel" — which is the safe path).
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");

        // Serve one connection that crashes without replying.
        let fake_handle = tokio::spawn(serve_crash_on_dispatch(listener));

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        // Call try_forward_inner directly to assert the discriminant.
        let outcome = try_forward_inner(&frame).await;

        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), fake_handle).await;

        assert!(
            matches!(outcome, ForwardOutcome::ParseFailure),
            "daemon crash (connection closed without response) must yield \
             ParseFailure, not NoSocket — got a different variant"
        );

        clear_daemon_env();
    }

    // ── argv_is_khive_daemon unit tests ───────────────────────────────────────

    #[test]
    fn argv_daemon_true_bare() {
        // Exact daemon argv as spawned by spawn_daemon().
        assert!(argv_is_khive_daemon("kkernel mcp --daemon"));
    }

    #[test]
    fn argv_daemon_true_absolute_path() {
        // Absolute path to kkernel binary — basename must match.
        assert!(argv_is_khive_daemon(
            "/Users/x/.cargo/bin/kkernel mcp --daemon"
        ));
    }

    #[test]
    fn argv_daemon_false_editor_with_kkernel_in_filename() {
        // Editor opened on a file whose name contains kkernel — not a daemon.
        assert!(!argv_is_khive_daemon("vim kkernel-notes.md"));
    }

    #[test]
    fn argv_daemon_false_less_with_kkernel_path() {
        // less paging a kkernel source file — argv[0] is "less", not "kkernel".
        assert!(!argv_is_khive_daemon(
            "less /Users/x/projects/kkernel/daemon.rs"
        ));
    }

    #[test]
    fn argv_daemon_false_kkernel_no_daemon_flag() {
        // kkernel exec subcommand — has kkernel basename but no --daemon token.
        assert!(!argv_is_khive_daemon("kkernel exec 'something'"));
    }

    #[test]
    fn argv_daemon_false_wrapper_argv0_not_kkernel() {
        // A wrapper script passes kkernel mcp --daemon as args but its own
        // argv[0] is "some-wrapper" — basename check must reject it.
        assert!(!argv_is_khive_daemon("some-wrapper kkernel mcp --daemon"));
    }

    #[test]
    fn argv_daemon_false_empty_string() {
        assert!(!argv_is_khive_daemon(""));
    }

    #[test]
    fn argv_daemon_true_with_surrounding_and_inner_whitespace() {
        assert!(argv_is_khive_daemon(
            "  /Users/x/.cargo/bin/kkernel   mcp    --daemon  "
        ));
    }

    #[test]
    fn argv_daemon_true_kkernel_bench_basename() {
        // bench binary copy is named "kkernel-bench"; pid management must treat it
        // as a valid daemon so stale bench daemons are SIGTERM'd on respawn.
        assert!(argv_is_khive_daemon("kkernel-bench mcp --daemon"));
        assert!(argv_is_khive_daemon(
            "/Users/x/.cargo/bin/kkernel-bench mcp --daemon"
        ));
    }

    // ── concurrent recovery — second client skips kill+spawn when daemon alive ──
    //
    // Exercises the recheck-under-lock (double-checked locking) in
    // kill_and_respawn.  Scenario:
    //
    //   1. A real daemon is running (via run_daemon).
    //   2. A recovering client calls kill_and_respawn directly — simulating a
    //      client that observed ParseFailure (from a now-dead OLD daemon) and
    //      wants to replace it, but a concurrent first-recoverer has ALREADY
    //      spawned a healthy daemon before this client reached the lock.
    //   3. Under the lock, kill_and_respawn sends a probe_only frame and finds a
    //      responsive, identity-matching daemon → returns RecoveryOutcome::Skipped.
    //   4. KILL_COUNT must be 0 and SPAWN_COUNT must be 0.
    //   5. The fresh daemon's PID file and socket must survive intact.
    //
    // FORCE_PID_IS_DAEMON=true makes every live PID SIGTERM-eligible so that
    // if the bounded-probe is removed (reverted) kill_stale_daemon_inner would
    // attempt SIGTERM against the real daemon PID, KILL_COUNT would be 1, and
    // the assertion below would catch the regression.
    //
    // Fail-if-reverted assertion: the `assert_eq!(KILL_COUNT, 0)` below catches
    // a reverted probe because without it, kill_stale_daemon_inner is called
    // unconditionally and KILL_COUNT increments to 1.

    #[tokio::test]
    #[serial]
    async fn concurrent_recovery_second_client_skips_kill_when_daemon_alive() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        // Run a real daemon so probe_daemon_identity in kill_and_respawn finds a
        // live, responsive, identity-matching daemon under the lock.
        let server = make_test_server();
        let config_id = server.config_id().to_string();
        let daemon_server = server.clone();
        let handle = tokio::spawn(async move {
            let _ = run_daemon(daemon_server).await;
        });

        // Wait for the daemon to bind the socket and write its PID file.
        let _ready = connect_when_ready(&sock).await;
        drop(_ready);

        // Record the daemon PID as written by run_daemon.
        let daemon_pid_str =
            std::fs::read_to_string(&pid_file).expect("daemon must have written a pid file");
        let daemon_pid: u32 = daemon_pid_str
            .trim()
            .parse()
            .expect("daemon pid file must contain a u32");

        // Arm the SIGTERM-eligible hook: pid_is_khive_daemon() will now return
        // true for the live daemon PID.  Without the bounded-probe, a reverted
        // kill_and_respawn would send SIGTERM to that PID and unlink the socket —
        // KILL_COUNT catches both paths.
        FORCE_PID_IS_DAEMON.store(true, std::sync::atomic::Ordering::SeqCst);
        reset_counters();

        // Call kill_and_respawn directly — simulates a second recovering client
        // whose turn arrives after the first recoverer already replaced the stale
        // daemon.  The bounded probe confirms the live daemon; Skipped is returned
        // without killing.
        let outcome = kill_and_respawn(&config_id, "test").await;

        assert!(
            matches!(outcome, Ok(RecoveryOutcome::Skipped)),
            "kill_and_respawn must return Ok(RecoveryOutcome::Skipped) when a live \
             matching daemon exists under the lock"
        );
        assert_eq!(
            KILL_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "KILL_COUNT must be 0: the probe found the daemon alive so \
             kill_stale_daemon_inner must NOT be called \
             (this assertion fails if the probe-under-lock is removed)"
        );
        assert_eq!(
            SPAWN_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "SPAWN_COUNT must be 0: no respawn needed when the daemon is alive \
             (this assertion fails if the probe-under-lock is removed)"
        );

        // The daemon's PID file and socket must be intact.
        assert!(
            pid_file.exists(),
            "PID file must survive: kill_and_respawn must NOT unlink it"
        );
        assert!(
            sock.exists(),
            "socket must survive: kill_and_respawn must NOT unlink it"
        );
        let surviving_pid: u32 = std::fs::read_to_string(&pid_file)
            .expect("pid file readable")
            .trim()
            .parse()
            .expect("pid file is a u32");
        assert_eq!(
            surviving_pid, daemon_pid,
            "PID in file must be the original daemon PID — no new daemon was spawned"
        );

        handle.abort();
        let _ = handle.await;
        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── oversized daemon response does not trigger kill/respawn ───────────────
    //
    // When the daemon's serialized response exceeds MAX_FRAME_BYTES the server
    // sends a small explicit error frame (ok=false, "response too large") instead
    // of closing the connection without a response.  The client receives this as
    // ForwardOutcome::Response (decodable frame) → map_response → Some(Err(..));
    // NOT ParseFailure, so no kill/respawn is triggered.
    //
    // This test drives the REAL handle_conn server path via run_daemon with a
    // BigDispatch that returns a string larger than MAX_FRAME_BYTES.  The client
    // calls forward_or_spawn and must receive Some(Err(..)) containing "too large".
    // The daemon's PID file and socket must survive the call.
    //
    // Fail-if-reverted: if the handle_conn oversized gate (the `if payload.len()
    // > MAX_FRAME_BYTES` branch that sends the small error frame) is removed,
    // handle_conn falls through to write_frame with the oversized payload, which
    // write_frame REJECTS (its own guard returns Err), causing the connection to
    // close without a response → the client sees ParseFailure → kill_and_respawn
    // runs → KILL_COUNT > 0 and the PID file assertion fails.

    /// A minimal DaemonDispatch that returns a payload larger than MAX_FRAME_BYTES
    /// so handle_conn's oversized guard fires and emits the "response too large"
    /// error frame.
    #[derive(Clone)]
    struct BigDispatch {
        namespace: String,
        config_id: String,
    }

    #[async_trait]
    impl daemon::DaemonDispatch for BigDispatch {
        async fn dispatch(
            &self,
            _ops: String,
            _presentation: Option<String>,
            _presentation_per_op: Option<Vec<Option<String>>>,
            _format: Option<String>,
            _format_per_op: Option<Vec<Option<String>>>,
            _from_wire: bool,
            _identity: Option<khive_runtime::RequestIdentity>,
        ) -> Result<String, String> {
            // Return a string whose serialized DaemonResponseFrame JSON length
            // exceeds MAX_FRAME_BYTES.  The frame JSON overhead is ~200 bytes so
            // a result of this size is comfortably over the cap.
            Ok("X".repeat(khive_runtime::daemon::MAX_FRAME_BYTES + 1))
        }

        async fn warm_all(&self) {}

        fn namespace(&self) -> &str {
            &self.namespace
        }

        fn config_id(&self) -> &str {
            &self.config_id
        }
    }

    #[tokio::test]
    #[serial]
    async fn oversized_daemon_response_sends_error_frame_not_kills_daemon() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "test-oversized-config";
        let dispatcher = BigDispatch {
            namespace: "test".to_string(),
            config_id: config_id.to_string(),
        };

        // Run the real daemon server (with handle_conn's oversized gate live).
        let handle = tokio::spawn(async move {
            let _ = run_daemon(dispatcher).await;
        });

        // Wait for the daemon to bind and write its PID.
        let _ready = connect_when_ready(&sock).await;
        drop(_ready);

        let daemon_pid: u32 = std::fs::read_to_string(&pid_file)
            .expect("daemon must have written a pid file")
            .trim()
            .parse()
            .expect("daemon pid must be a u32");

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let result = forward_or_spawn(&frame).await;

        // Primary assertion: the daemon's PID file and socket must be intact
        // (no kill/respawn triggered).
        assert!(
            pid_file.exists(),
            "PID file must survive: oversized response is NOT a daemon crash"
        );
        assert!(
            sock.exists(),
            "socket must survive: oversized response is NOT a daemon crash"
        );
        let surviving_pid: u32 = std::fs::read_to_string(&pid_file)
            .expect("pid file readable")
            .trim()
            .parse()
            .expect("pid file is a u32");
        assert_eq!(
            surviving_pid, daemon_pid,
            "daemon PID must not change — no kill+respawn occurred"
        );
        assert_eq!(
            KILL_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "KILL_COUNT must be 0 — oversized response must NOT trigger \
             kill_stale_daemon_inner (fails if handle_conn oversized gate is removed)"
        );

        // The result must be Some(Err(..)) containing "too large" — the explicit
        // error frame the server sends when the real response is oversized.
        match result {
            Some(Err(e)) => {
                assert!(
                    e.message.contains("too large"),
                    "error must describe the oversized response; got: {}",
                    e.message
                );
            }
            Some(Ok(_)) => panic!("oversized response must not produce Ok result"),
            None => panic!(
                "oversized response must produce Some(Err(..)) from the explicit \
                 error frame, not None (None would mean map_response fell back to \
                 local dispatch, hiding the server-side error)"
            ),
        }

        handle.abort();
        let _ = handle.await;
        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── EXACTLY-ONCE: real request dispatched exactly once on recovery path ────
    //
    // Scenario:
    //   1. A fake "stale" socket reads one request frame then closes without
    //      responding (simulating a crashed old daemon on the first forward).
    //   2. `forward_or_spawn` sees ParseFailure → enters kill_and_respawn.
    //   3. Under the lock, probe_daemon_identity sends a probe_only frame to a
    //      REAL CountingDispatch daemon (already running).  The daemon's
    //      handle_conn returns an identity frame without calling dispatch() —
    //      DAEMON_DISPATCH stays 0.
    //   4. kill_and_respawn returns RecoveryOutcome::Skipped (live daemon found).
    //   5. The call site forwards the REAL request once via try_forward_inner.
    //      CountingDispatch.dispatch() is called → DAEMON_DISPATCH == 1.
    //
    // Fail-if-reverted: if kill_and_respawn is reverted to use the real frame as
    // the probe (try_forward_inner(frame) under the lock), CountingDispatch.dispatch()
    // is called for the probe (DAEMON_DISPATCH == 1), and again at the call site
    // (DAEMON_DISPATCH == 2).  The assert_eq!(DAEMON_DISPATCH, 1) then fails.
    //
    // FOLLOWUP: this test calls kill_and_respawn + try_forward_inner directly to
    // avoid the two-socket problem (stale socket for first forward, real socket
    // for probe).  A future enhancement could drive the full forward_or_spawn
    // entry point using a proxy fake-socket that serves garbage on the first
    // connection then falls through to the real daemon — catching any future
    // call-site that double-forwards after RecoveryOutcome::Skipped without
    // going through the seam functions.  File a follow-up issue if needed.

    /// A minimal DaemonDispatch that increments DAEMON_DISPATCH on every real
    /// (non-probe) dispatch.  probe_only frames never reach dispatch() — they
    /// are short-circuited by handle_conn before calling the dispatcher.
    #[derive(Clone)]
    struct CountingDispatch {
        namespace: String,
        config_id: String,
    }

    #[async_trait]
    impl daemon::DaemonDispatch for CountingDispatch {
        async fn dispatch(
            &self,
            _ops: String,
            _presentation: Option<String>,
            _presentation_per_op: Option<Vec<Option<String>>>,
            _format: Option<String>,
            _format_per_op: Option<Vec<Option<String>>>,
            _from_wire: bool,
            _identity: Option<khive_runtime::RequestIdentity>,
        ) -> Result<String, String> {
            DAEMON_DISPATCH.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok("{\"ok\":true,\"counted\":true}".to_string())
        }

        async fn warm_all(&self) {}

        fn namespace(&self) -> &str {
            &self.namespace
        }

        fn config_id(&self) -> &str {
            &self.config_id
        }
    }

    #[tokio::test]
    #[serial]
    async fn recovery_path_dispatches_real_request_exactly_once() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let real_sock = dir.path().join("khived.sock");
        let stale_sock = dir.path().join("stale.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        // Start a real CountingDispatch daemon on `real_sock`.
        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";
        let counting_dispatcher = CountingDispatch {
            namespace: "test".to_string(),
            config_id: config_id.to_string(),
        };
        // Temporarily point KHIVE_SOCKET at real_sock to let run_daemon bind there.
        std::env::set_var("KHIVE_SOCKET", &real_sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let counting_handle = tokio::spawn(async move {
            let _ = run_daemon(counting_dispatcher).await;
        });
        let _ready = connect_when_ready(&real_sock).await;
        drop(_ready);

        // Bind the stale fake socket on `stale_sock` BEFORE redirecting the client
        // env so the client sees it first.  The fake stale socket reads one frame
        // then drops without responding — simulating a crashed old daemon.
        let stale_listener =
            tokio::net::UnixListener::bind(&stale_sock).expect("bind stale socket");
        let stale_handle = tokio::spawn(serve_crash_on_dispatch(stale_listener));

        // Now point the client at the STALE socket so its first try_forward_inner
        // sees ParseFailure (the stale socket crashes after reading the request).
        // The recovery lock and PID file point at the stale path so kill_stale_daemon
        // looks at a non-kkernel PID (the current test process) and skips SIGTERM.
        let stale_pid_file = dir.path().join("stale.pid");
        std::fs::write(&stale_pid_file, std::process::id().to_string()).expect("write stale pid");
        std::env::set_var("KHIVE_SOCKET", &stale_sock);
        std::env::set_var("KHIVE_PID", &stale_pid_file);

        // After kill_and_respawn's probe, the probe needs to reach the REAL daemon.
        // We achieve this by having the probe use `real_sock` (via KHIVE_SOCKET).
        // Redirect KHIVE_SOCKET back to the real daemon socket AFTER the stale
        // socket has been read (but we need the probe to already know where to look).
        //
        // Simpler approach: use a single socket for both — the stale response is
        // from a `serve_crash_on_dispatch` (closes after one read), the NEXT
        // connection attempt (the probe) goes to the same path, but the stale
        // listener is now gone.  Instead we use real_sock for the probe by
        // redirecting KHIVE_SOCKET before the probe fires.
        //
        // The cleanest approach: let forward_or_spawn do everything from the stale
        // socket (ParseFailure), then kill_and_respawn calls probe_daemon_identity
        // which uses socket_path() — still pointing at stale_sock (now dead).
        // probe_daemon_identity sees NoSocket → ProbeOutcome::Dead → kill+spawn
        // path.  That's NOT the scenario we want to test (double-dispatch on Skipped).
        //
        // To get the "Skipped" (live daemon under lock) scenario without a real
        // spawn: call kill_and_respawn directly with KHIVE_SOCKET pointing at the
        // live CountingDispatch daemon, and separately assert DAEMON_DISPATCH from
        // a direct try_forward_inner call.

        // Point back at the real daemon for the probe and the real forward.
        std::env::set_var("KHIVE_SOCKET", &real_sock);
        std::env::set_var("KHIVE_PID", &pid_file);

        // Simulate the exactly-once scenario:
        //   (a) kill_and_respawn sees a live daemon → returns Skipped (0 dispatches)
        //   (b) call site forwards the real request once → 1 dispatch
        let recovery = kill_and_respawn(config_id, "test").await;
        assert!(
            matches!(recovery, Ok(RecoveryOutcome::Skipped)),
            "probe must find the live CountingDispatch daemon and return Skipped"
        );
        // DAEMON_DISPATCH must still be 0: the probe_only frame does not reach dispatch().
        assert_eq!(
            DAEMON_DISPATCH.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "probe_only frame must NOT increment DAEMON_DISPATCH \
             (fails if the real request is used as the probe)"
        );

        // Now forward the real request exactly once — the call site's single forward.
        let real_frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };
        let fwd = try_forward_inner(&real_frame).await;
        assert!(
            matches!(fwd, ForwardOutcome::Response(_)),
            "real forward after Skipped must succeed; got non-Response outcome"
        );

        // DAEMON_DISPATCH must now be exactly 1.
        assert_eq!(
            DAEMON_DISPATCH.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "real request must be dispatched EXACTLY ONCE across the recovery path \
             (assert fails with count==2 if the real frame is used as the probe \
              AND re-forwarded at the call site — the double-dispatch bug)"
        );

        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), stale_handle).await;
        counting_handle.abort();
        let _ = counting_handle.await;
        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    /// Serve every connection with a valid probe-ack response identity-matching
    /// `config_id`, forever (until the listener is dropped/aborted). Simulates
    /// an already-bound, healthy daemon that answers `probe_only` frames
    /// immediately.
    async fn serve_probe_ack_forever(listener: tokio::net::UnixListener, config_id: String) {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            if read_frame(&mut stream).await.is_err() {
                continue;
            }
            let response = DaemonResponseFrame {
                ok: true,
                result: None,
                error: None,
                namespace_mismatch: false,
                config_mismatch: false,
                served_config_id: Some(config_id.clone()),
                version_mismatch: false,
                daemon_protocol_version: PROTOCOL_VERSION,
                metrics: None,
            };
            if let Ok(payload) = serde_json::to_vec(&response) {
                let _ = write_frame(&mut stream, &payload).await;
            }
        }
    }

    // ── #758: confirm_genuinely_dead must not trust a bare Dead reading while
    // a peer's boot is in flight ───────────────────────────────────────────
    //
    // Regression for the daemon-recovery double-spawn window: `spawn_daemon()`
    // is fire-and-forget, so a concurrent recoverer's identity probe can
    // observe `Dead` in the gap between a peer's `cmd.spawn()` returning and
    // that child reaching its own `acquire_daemon_boot_guard()` call. This
    // test drives that gap directly: a background OS thread holds the real
    // boot/recovery lock for `GUARD_HOLD` (simulating "a peer's child is
    // mid cold-boot, holding the same lock this process would need to
    // classify it"), while a fake, already-bound, identity-matching listener
    // answers probe_only frames the instant it is asked (simulating "the
    // peer's child has already bound its socket and would answer this
    // instant, if only the classifier would wait for the lock instead of
    // trusting an immediate Dead reading").
    //
    // Fail-if-reverted: without the fix, `confirm_genuinely_dead` would not
    // exist and `kill_and_respawn` would trust the bare `Dead` result
    // immediately — this test exercises the new function directly, so
    // reverting it is a compile error. The regression oracle is the `timeout`
    // window below: it asserts `confirm_genuinely_dead` does NOT resolve
    // while the peer explicitly still holds the lock, then asserts it DOES
    // resolve (as `Alive`) once the peer explicitly releases it — real
    // two-way synchronization via channels, not a fixed sleep + elapsed-time
    // assertion (#838 round-1 Finding 3: the previous version held the guard
    // via `std::thread::sleep` and asserted `elapsed >= GUARD_HOLD`, which is
    // timing-dependent under load).
    #[tokio::test]
    #[serial]
    async fn confirm_genuinely_dead_waits_for_peer_to_release_boot_guard() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let server = make_test_server();
        let config_id = server.config_id().to_string();

        // A fake daemon is already reachable from T=0 — proving the wait
        // below is caused by the contended lock, not by the daemon being
        // slow to bind.
        let listener = tokio::net::UnixListener::bind(&sock).expect("bind fake daemon socket");
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write fake pid file");
        let serve_handle = tokio::spawn(serve_probe_ack_forever(listener, config_id.clone()));

        // Real two-way synchronization: the boot-holder thread signals once it
        // has genuinely acquired the lock (so the test never proceeds before
        // contention is real), then blocks on an explicit release channel
        // instead of a fixed sleep — the test controls exactly when the lock
        // becomes available, with no timing guess involved.
        let (acquired_tx, acquired_rx) = std::sync::mpsc::channel::<()>();
        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
        let boot_thread = std::thread::spawn(move || {
            let guard = khive_runtime::daemon::acquire_daemon_boot_guard()
                .expect("test boot holder must acquire the recovery lock");
            acquired_tx.send(()).expect("signal lock acquired");
            let _ = release_rx.recv();
            drop(guard);
        });
        acquired_rx
            .recv()
            .expect("boot-holder thread must signal after acquiring the lock");

        let confirm_fut = confirm_genuinely_dead(&config_id, "test");
        tokio::pin!(confirm_fut);

        // Bounded assertion window (NOT the release mechanism — the lock is
        // released explicitly below via `release_tx`): proves
        // confirm_genuinely_dead does not resolve while the peer still holds
        // the lock.
        let too_early =
            tokio::time::timeout(std::time::Duration::from_millis(150), &mut confirm_fut).await;
        assert!(
            too_early.is_err(),
            "confirm_genuinely_dead must not resolve while a peer holds the \
             boot/recovery lock"
        );

        release_tx
            .send(())
            .expect("boot-holder thread still awaiting release");
        boot_thread
            .join()
            .expect("boot-holder thread must not panic");

        let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), confirm_fut)
            .await
            .expect("confirm_genuinely_dead must resolve promptly once the peer releases the lock");

        assert!(
            matches!(outcome, ProbeOutcome::Alive),
            "confirm_genuinely_dead must observe the already-reachable daemon \
             once the contended lock clears, not conclude Dead early"
        );

        serve_handle.abort();
        let _ = serve_handle.await;
        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── #838 round-2 Finding 1: an earlier LockContended round must not be
    // erased by a later Dead round ──────────────────────────────────────────
    //
    // `confirm_genuinely_dead` only trusts `Dead` once EVERY round agrees the
    // daemon is absent. Before this fix, the aggregation tracked only the
    // LAST round's outcome: a LockContended round followed by a later Dead
    // round overwrote the earlier contention and the whole call returned
    // `Dead`, which `kill_and_respawn` trusts enough to kill+spawn — even
    // though quiescence was never actually established across every round.
    //
    // This test drives that exact sequence directly: a background thread
    // holds the real boot/recovery lock for longer than a single round's
    // bounded wait (`BOOT_QUIESCENCE_LOCK_TIMEOUT_MS` = 500ms), guaranteeing
    // round 1 cannot acquire it and returns `LockContended`, then releases
    // the lock before `confirm_genuinely_dead` returns — so the remaining
    // rounds observe the (genuinely absent) daemon and return `Dead`.
    //
    // Fail-if-reverted: with the old last-round-wins aggregation, this
    // LockContended-then-Dead sequence resolves to `ProbeOutcome::Dead`, and
    // the assertion below fails.
    #[tokio::test]
    #[serial]
    async fn confirm_genuinely_dead_is_sticky_uncertain_after_earlier_contention() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        // Genuinely no daemon at all: no socket, no pid file. Once the lock
        // is free, every round's identity probe observes Dead.
        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main".to_string();

        // Hold the real boot/recovery lock from before `confirm_genuinely_dead`
        // starts, for strictly longer than one round's bounded wait, so round
        // 1 is deterministically unable to acquire it (LockContended) — not a
        // probabilistic race, since the hold spans the round's entire
        // deadline window.
        let guard = khive_runtime::daemon::acquire_daemon_boot_guard()
            .expect("test lock holder must acquire the recovery lock");
        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
        let boot_thread = std::thread::spawn(move || {
            let _ = release_rx.recv();
            drop(guard);
        });

        let confirm_config_id = config_id.clone();
        let confirm_handle =
            tokio::spawn(async move { confirm_genuinely_dead(&confirm_config_id, "test").await });

        // Round 1's bounded wait is 500ms; sleeping past it while still
        // holding the lock guarantees round 1 observed LockContended before
        // release. `confirm_handle` runs concurrently on the runtime during
        // this sleep (unlike a merely-pinned, never-polled future), so round
        // 1 genuinely contends against the held lock here.
        tokio::time::sleep(std::time::Duration::from_millis(650)).await;
        release_tx
            .send(())
            .expect("boot-holder thread still awaiting release");
        boot_thread
            .join()
            .expect("boot-holder thread must not panic");

        let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), confirm_handle)
            .await
            .expect("confirm_genuinely_dead must resolve once the lock is released")
            .expect("confirm_genuinely_dead task must not panic");

        assert!(
            matches!(outcome, ProbeOutcome::LockContended),
            "an earlier LockContended round must make the whole call \
             LockContended (sticky), never overwritten by a later round's \
             Dead reading; got {outcome:?}"
        );
        assert_eq!(
            KILL_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "confirm_genuinely_dead must never kill on its own"
        );
        assert_eq!(
            SPAWN_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "confirm_genuinely_dead must never spawn on its own"
        );

        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── #838 round-1 Finding 1 / round-2 Finding 2: two concurrent recoverers
    // must not double-spawn ─────────────────────────────────────────────────
    //
    // Regression for the missing linearization point: before the recoverer
    // lock, two clients racing `kill_and_respawn` from a genuinely dead daemon
    // (no socket, no pid file at all) could both classify `Dead` via
    // `confirm_genuinely_dead` and both fall through to `kill_stale_daemon_inner`
    // + `spawn_daemon`, spawning two replacement daemons. The recoverer lock
    // (a SEPARATE file from the daemon's own boot lock, so it cannot deadlock
    // against a peer daemon's boot) makes the whole
    // confirm-through-spawn critical section mutually exclusive across
    // recoverers: only the first to acquire it proceeds to `Spawned`; the
    // second observes `Skipped` (freshly spawned daemon now answers its
    // re-probe under the lock) or `Uncertain`, but never spawns.
    //
    // #838 round-2 Finding 2: this test previously let the two `kill_and_respawn`
    // calls reach the recoverer-lock attempt on whatever schedule the tokio
    // executor happened to give them, with the watcher below triggering off
    // `SPAWN_COUNT` alone. That meant the test passed 6/6 even with the
    // recoverer lock's acquisition removed entirely (sabotage-proven by
    // review): normal scheduling let the first call run far enough ahead
    // that its full classify+kill+spawn completed (making the watcher's fake
    // daemon live) before the second call's own classification rounds ever
    // observed anything, so the "lock" was never actually exercised as the
    // thing preventing the double-spawn. `RECOVERY_RACE_BARRIER` forces both
    // calls to reach "independently classified Dead" at the exact same
    // instant, so it is genuinely the recoverer lock (or its absence) that
    // decides whether one or both proceed to kill+spawn.
    //
    // Fail-if-reverted: without the recoverer lock serializing this section,
    // both concurrent calls independently pass their own `confirm_genuinely_dead`
    // and both call `spawn_daemon()`, making `SPAWN_COUNT == 2`.
    #[tokio::test]
    #[serial]
    async fn concurrent_recoverers_spawn_exactly_one_replacement_daemon() {
        clear_daemon_env();
        reset_counters();
        *RECOVERY_RACE_BARRIER
            .lock()
            .expect("barrier mutex poisoned") =
            Some(std::sync::Arc::new(tokio::sync::Barrier::new(2)));
        *SPAWN_COMMIT_BARRIER.lock().expect("barrier mutex poisoned") =
            Some(std::sync::Arc::new(tokio::sync::Barrier::new(2)));
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");
        let recoverer_lock_file = dir.path().join("khived.recoverer.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::set_var("KHIVE_RECOVERER_LOCK", &recoverer_lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        // Genuinely no daemon at all: no socket, no pid file. Both recoverers
        // must observe Dead on every probe they take.
        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main".to_string();

        // `spawn_daemon()` in this test process just forks the test binary
        // with bad args — it exits almost immediately without ever binding
        // anything (the same tolerated pattern used elsewhere in this file,
        // e.g. `forward_or_spawn_blocks_on_boot_quiescence_before_local_fallback`).
        // To prove the recoverer lock's linearization actually matters (not
        // merely that nothing ever comes up so there is nothing to race),
        // this watcher simulates "the first recoverer's spawned child became
        // live" the instant `spawn_daemon()` is genuinely called: it binds
        // the fake socket + pid file and starts answering `probe_only`
        // frames, so the SECOND recoverer's own `confirm_genuinely_dead`
        // (which can only proceed once the first has released the recoverer
        // lock) observes a live daemon and skips instead of also spawning.
        let watcher_config_id = config_id.clone();
        let watcher_sock = sock.clone();
        let watcher_pid_file = pid_file.clone();
        let watcher = tokio::spawn(async move {
            let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5);
            while SPAWN_COUNT.load(std::sync::atomic::Ordering::SeqCst) == 0 {
                assert!(
                    tokio::time::Instant::now() < deadline,
                    "no recoverer reached spawn_daemon() within 5s"
                );
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
            let listener =
                tokio::net::UnixListener::bind(&watcher_sock).expect("bind simulated-spawn socket");
            std::fs::write(&watcher_pid_file, std::process::id().to_string())
                .expect("write simulated-spawn pid file");
            serve_probe_ack_forever(listener, watcher_config_id).await;
        });

        let (a, b) = tokio::join!(
            kill_and_respawn(&config_id, "test"),
            kill_and_respawn(&config_id, "test"),
        );
        let spawned_count = [&a, &b]
            .iter()
            .filter(|r| matches!(r, Ok(RecoveryOutcome::Spawned)))
            .count();
        assert_eq!(
            spawned_count, 1,
            "exactly one of two concurrent recoverers racing from a genuinely \
             dead daemon must spawn a replacement; got a={a:?} b={b:?}"
        );
        assert_eq!(
            SPAWN_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "spawn_daemon must be called exactly once across two concurrent \
             recoverers racing the same dead-daemon state; got a={a:?} b={b:?}"
        );

        watcher.abort();
        let _ = watcher.await;
        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
        std::env::remove_var("KHIVE_RECOVERER_LOCK");
    }

    // ── probe classifier is fail-CLOSED for same-protocol pre-probe daemons ──
    //
    // Regression test for the version-skew gap: a daemon built BEFORE probe_only
    // was introduced but carrying PROTOCOL_VERSION (same numeric version, older
    // binary) deserialises the probe frame via serde default and falls through to
    // dispatch on the empty `ops` string.  It returns ok=false (parse error on
    // empty ops) WITH matching identity fields (namespace / config / protocol all
    // match).  Before leg-1 of the round-5 fix, probe_daemon_identity classified
    // ANY response with matching identity as Alive, leaving the stale daemon in
    // place.
    //
    // After the fix, the classifier requires the probe-ack sentinel shape:
    //   resp.ok && resp.result.is_none() && resp.error.is_none()
    // An ok=false response fails this predicate → Dead → kill+spawn.
    //
    // Fail-if-reverted: removing the `is_probe_ack` check from the classifier
    // causes the ok=false identity-matching response to be classified Alive →
    // kill_and_respawn returns Skipped → KILL_COUNT stays 0 → assertion fails.

    /// Build a response that matches all identity fields but has ok=false (the
    /// shape a pre-probe daemon produces when dispatching the empty-ops probe).
    fn pre_probe_daemon_response(config_id: &str) -> DaemonResponseFrame {
        DaemonResponseFrame {
            ok: false,
            result: None,
            error: Some("parse error: empty ops string".to_string()),
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(config_id.to_string()),
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
        }
    }

    #[tokio::test]
    #[serial]
    async fn probe_classifier_dead_when_same_protocol_daemon_lacks_probe_support() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        // Bind a fake socket that serves one pre-probe response, then stops
        // accepting (simulates a same-protocol daemon without probe_only support).
        let listener = tokio::net::UnixListener::bind(&sock).expect("bind fake pre-probe socket");
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");
        let pre_probe_resp = pre_probe_daemon_response(config_id);
        let fake_handle = tokio::spawn(serve_one_response(listener, pre_probe_resp));

        // Arm FORCE_PID_IS_DAEMON so kill_stale_daemon_inner would attempt SIGTERM
        // IF it were called.  This makes KILL_COUNT the reliable regression signal:
        // if the classifier incorrectly returns Alive (Skipped), kill is not called
        // and KILL_COUNT stays 0.
        FORCE_PID_IS_DAEMON.store(true, std::sync::atomic::Ordering::SeqCst);
        reset_counters();

        let outcome = kill_and_respawn(config_id, "test").await;

        // The fake socket served exactly one response; join it before asserting.
        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), fake_handle).await;

        // The ok=false response must NOT be classified as Alive.  kill_and_respawn
        // must attempt kill+spawn (Spawned outcome — spawn itself fails because there
        // is no real kkernel binary in test, but KILL_COUNT is checked BEFORE spawn).
        assert!(
            matches!(outcome, Ok(RecoveryOutcome::Spawned) | Err(_)),
            "pre-probe same-protocol daemon must NOT be classified Alive; \
             expected Spawned or spawn-error, got Skipped"
        );
        assert_eq!(
            KILL_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "KILL_COUNT must be 1 — the pre-probe response must classify as Dead \
             so kill_stale_daemon_inner is called \
             (this fails if is_probe_ack check is removed and ok=false response \
              is incorrectly classified as Alive)"
        );

        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── dispatch error propagates to client as non-empty message (#91) ─────────
    //
    // Regression for #91: when the daemon's dispatcher returns Err(msg), the
    // client-side `map_response` must surface that message through
    // `forward_or_spawn` as `Some(Err(McpError { message, .. }))` with a
    // non-empty `message`.  Before the #91 fix, some failure paths swallowed the
    // message and the client saw only "daemon returned an error without a message".
    //
    // This test uses a real run_daemon + FailDispatch (always returns Err("forced
    // dispatch error: <detail>")) and drives the full forward_or_spawn path so we
    // exercise both the daemon's response serialization AND the client's
    // map_response deserialization in one round trip.

    #[derive(Clone)]
    struct FailDispatch {
        namespace: String,
        config_id: String,
    }

    #[async_trait]
    impl daemon::DaemonDispatch for FailDispatch {
        async fn dispatch(
            &self,
            _ops: String,
            _presentation: Option<String>,
            _presentation_per_op: Option<Vec<Option<String>>>,
            _format: Option<String>,
            _format_per_op: Option<Vec<Option<String>>>,
            _from_wire: bool,
            _identity: Option<khive_runtime::RequestIdentity>,
        ) -> Result<String, String> {
            Err("forced dispatch error: verb returned an error for testing".to_string())
        }

        async fn warm_all(&self) {}

        fn namespace(&self) -> &str {
            &self.namespace
        }

        fn config_id(&self) -> &str {
            &self.config_id
        }
    }

    #[tokio::test]
    #[serial]
    async fn dispatch_error_propagates_as_non_empty_client_message() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";
        let dispatcher = FailDispatch {
            namespace: "test".to_string(),
            config_id: config_id.to_string(),
        };

        let handle = tokio::spawn(async move {
            let _ = run_daemon(dispatcher).await;
        });

        let _ready = connect_when_ready(&sock).await;
        drop(_ready);

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let result = forward_or_spawn(&frame).await;

        match result {
            Some(Err(McpError { message, .. })) => {
                assert!(
                    !message.is_empty(),
                    "error message forwarded to client must not be empty"
                );
                assert!(
                    message.contains("forced dispatch error"),
                    "client must receive the daemon's error message verbatim; got: {message}"
                );
            }
            Some(Ok(v)) => panic!("FailDispatch always errs; got Ok({v:?})"),
            None => panic!(
                "forward_or_spawn returned None (local fallback) instead of \
                 propagating the daemon's error — the error message was swallowed"
            ),
        }

        handle.abort();
        let _ = handle.await;
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── explicit version_mismatch from stale daemon routes to recovery (#156) ──
    //
    // Regression for #156: when a NEWER client connects to an OLD warm daemon,
    // the old daemon responds with `version_mismatch=true` and its own (lower)
    // `daemon_protocol_version`. Before the fix, this went to the generic
    // `Response` arm → `map_response` → hard MCP error, leaving the stale daemon
    // alive instead of triggering kill+respawn.
    //
    // After the fix, `try_forward_inner` detects `version_mismatch=true` &&
    // `daemon_protocol_version < PROTOCOL_VERSION` and returns
    // `ForwardOutcome::ProtocolMismatch`, routing it through kill_and_respawn
    // exactly like the implicit (no version_mismatch flag) old-daemon case.
    //
    // This test serves one connection returning an explicit-mismatch frame
    // (version_mismatch=true, daemon_protocol_version=0) and asserts that
    // try_forward_inner classifies it as ProtocolMismatch (not Response).

    fn explicit_version_mismatch_response(config_id: &str) -> DaemonResponseFrame {
        DaemonResponseFrame {
            ok: false,
            result: None,
            error: Some(format!(
                "daemon protocol mismatch: client={} daemon=0 — \
                 rebuild/update the client binary (make local)",
                PROTOCOL_VERSION
            )),
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(config_id.to_string()),
            version_mismatch: true,
            daemon_protocol_version: 0,
            metrics: None,
        }
    }

    #[tokio::test]
    #[serial]
    async fn try_forward_inner_routes_explicit_version_mismatch_to_protocol_mismatch() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        let listener =
            tokio::net::UnixListener::bind(&sock).expect("bind explicit-mismatch socket");
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");

        let mismatch_resp = explicit_version_mismatch_response(config_id);
        let fake_handle = tokio::spawn(serve_one_response(listener, mismatch_resp));

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let outcome = try_forward_inner(&frame).await;

        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), fake_handle).await;

        assert!(
            matches!(outcome, ForwardOutcome::ProtocolMismatch),
            "explicit version_mismatch=true with daemon_protocol_version < PROTOCOL_VERSION \
             must classify as ProtocolMismatch (triggers kill+respawn), not Response \
             (which would surface a hard error and leave the stale daemon alive)"
        );

        clear_daemon_env();
    }

    // ── explicit version_mismatch from NEWER daemon is NOT routed to recovery ──
    //
    // Complementary to the test above: when a stale CLIENT talks to a NEWER
    // daemon, the daemon responds with `version_mismatch=true` and a
    // `daemon_protocol_version > PROTOCOL_VERSION`. Kill+respawn cannot fix this
    // (it would just spawn the same newer daemon again); the client must receive
    // a hard error telling the operator to upgrade the client binary.
    //
    // This test asserts try_forward_inner returns ForwardOutcome::Response
    // (not ProtocolMismatch) so map_response produces the hard error.

    fn newer_daemon_version_mismatch_response(config_id: &str) -> DaemonResponseFrame {
        DaemonResponseFrame {
            ok: false,
            result: None,
            error: Some(format!(
                "daemon protocol mismatch: client={} daemon={} — \
                 rebuild/update the client binary (make local)",
                PROTOCOL_VERSION,
                PROTOCOL_VERSION + 1
            )),
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id: Some(config_id.to_string()),
            version_mismatch: true,
            daemon_protocol_version: PROTOCOL_VERSION + 1,
            metrics: None,
        }
    }

    #[tokio::test]
    #[serial]
    async fn try_forward_inner_newer_daemon_mismatch_yields_response_not_recovery() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        let listener = tokio::net::UnixListener::bind(&sock).expect("bind newer-daemon socket");
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");

        let mismatch_resp = newer_daemon_version_mismatch_response(config_id);
        let fake_handle = tokio::spawn(serve_one_response(listener, mismatch_resp));

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let outcome = try_forward_inner(&frame).await;

        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), fake_handle).await;

        assert!(
            matches!(outcome, ForwardOutcome::Response(_)),
            "version_mismatch=true with daemon_protocol_version > PROTOCOL_VERSION \
             must yield Response (hard error via map_response), not ProtocolMismatch \
             (kill+respawn cannot fix a stale client binary)"
        );

        clear_daemon_env();
    }

    // ── #644: ambiguous post-write outcome never retries or falls back ───────
    //
    // Before the #644 fix, a `ParseFailure` on the first `try_forward_inner`
    // attempt (the real frame was already fully written) triggered
    // `kill_and_respawn` followed by resending the SAME real frame to whatever
    // daemon answered next. If the original (stale) daemon had actually
    // dispatched the mutation before the connection dropped, that retry would
    // execute it a second time on the freshly-spawned daemon.
    //
    // This test proves the fixed contract: once the real frame is confirmed
    // written, `forward_or_spawn` returns a hard error and never opens another
    // connection — not even to a daemon that is fully ready and willing to
    // serve the exact same request.

    #[tokio::test]
    #[serial]
    async fn ambiguous_write_never_retries_against_freshly_spawned_daemon() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        // Stale daemon: accepts one connection, reads the request, then drops
        // without responding — forces the first try_forward_inner to see
        // ParseFailure (frame written, response lost).
        let stale_listener = tokio::net::UnixListener::bind(&sock).expect("bind stale socket");
        let stale_handle = tokio::spawn(serve_crash_on_dispatch(stale_listener));

        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        // A fully ready "freshly respawned" daemon binds shortly after the
        // stale one drops. If the fix regresses and forward_or_spawn retries
        // the real frame, this listener would happily answer it — that's
        // exactly why connect_count must stay 0.
        let connect_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let connect_count_srv = connect_count.clone();
        let resp = frame_ok("stats-result");
        let fresh_sock = sock.clone();
        let fresh_handle = tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
            let listener =
                tokio::net::UnixListener::bind(&fresh_sock).expect("bind fresh daemon socket");
            if let Ok((mut stream, _)) = listener.accept().await {
                connect_count_srv.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                if read_frame(&mut stream).await.is_ok() {
                    if let Ok(payload) = serde_json::to_vec(&resp) {
                        let _ = write_frame(&mut stream, &payload).await;
                    }
                }
            }
        });

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let result = forward_or_spawn(&frame).await;

        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), stale_handle).await;
        // Give the never-contacted fresh listener a moment to prove it stays idle,
        // then drop it so its task doesn't hang the test process.
        tokio::time::sleep(std::time::Duration::from_millis(400)).await;
        fresh_handle.abort();
        let _ = fresh_handle.await;

        match result {
            Some(Err(McpError { message, .. })) => {
                assert!(
                    message.contains("not retrying") && message.contains("duplicate execution"),
                    "ambiguous post-write outcome must return the #644 hard-error \
                     message; got: {message}"
                );
            }
            other => {
                panic!("expected Some(Err(..)) for an ambiguous post-write outcome, got {other:?}")
            }
        }

        assert_eq!(
            connect_count.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "forward_or_spawn must NOT contact any daemon (stale or freshly \
             spawned) again once the real frame has been fully written — \
             retrying risks a duplicate dispatch (#644)"
        );

        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── #644: exactly-once dispatch through the full forward_or_spawn path ───
    //
    // End-to-end version of `recovery_path_dispatches_real_request_exactly_once`
    // (which drives `kill_and_respawn` + `try_forward_inner` directly to avoid
    // a two-socket setup problem). This test drives the full public entry point:
    // a single fake daemon counts every real (non-probe) dispatch, answers
    // `probe_only` frames with a valid identity ack, and closes the connection
    // without responding to the real frame — simulating a crash after dispatch.
    //
    // Fail-if-reverted: if `forward_or_spawn` ever resends the real frame after
    // a confirmed write (the #644 bug), this fake daemon would dispatch it
    // again and DAEMON_DISPATCH would read 2, not 1.

    #[tokio::test]
    #[serial]
    async fn forward_or_spawn_dispatches_real_frame_exactly_once_end_to_end() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        // `forward_or_spawn`'s first attempt writes the real (non-probe) frame
        // directly — no probe precedes it — so this fake daemon only needs to
        // simulate "read the real frame, dispatch it, then crash before
        // responding" for the very first connection.
        let listener = tokio::net::UnixListener::bind(&sock).expect("bind fake daemon socket");
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");
        let dispatch_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let dispatch_count_srv = dispatch_count.clone();
        let fake_handle = tokio::spawn(async move {
            loop {
                let Ok((mut stream, _)) = listener.accept().await else {
                    break;
                };
                if read_frame(&mut stream).await.is_err() {
                    continue;
                }
                // Count the dispatch, then drop the connection without
                // responding — simulating a crash after the mutation already ran.
                dispatch_count_srv.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                drop(stream);
            }
        });

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let result = forward_or_spawn(&frame).await;

        match result {
            Some(Err(_)) => {}
            other => panic!(
                "expected Some(Err(..)) — not None (silent local fallback) — for a \
                 dispatch-then-crash response, got {other:?}"
            ),
        }
        assert_eq!(
            dispatch_count.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the real request must be dispatched EXACTLY ONCE; a value of 2 means \
             forward_or_spawn resent the frame after the write already completed"
        );

        fake_handle.abort();
        let _ = fake_handle.await;
        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── #644 boundary: a successful daemon round trip dispatches exactly once ──
    //
    // The crash-after-dispatch test above proves the `Err` boundary (a lost
    // response must never trigger a resend). This test proves the opposite
    // boundary on the SAME counter: when the daemon successfully answers, the
    // real frame must still have been dispatched exactly once — not retried
    // after a successful response, and not dispatched a second time by any
    // fallback path once `forward_or_spawn` already returns `Some(Ok(_))`.
    //
    // Fail-if-reverted: if a future edit ever re-sent the real frame after a
    // successful response (e.g. a stray retry-on-timeout wrapped around the
    // whole attempt), this fake daemon would observe DISPATCH_COUNT == 2.
    #[tokio::test]
    #[serial]
    async fn forward_or_spawn_dispatches_real_frame_exactly_once_on_success() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        let listener = tokio::net::UnixListener::bind(&sock).expect("bind fake daemon socket");
        std::fs::write(&pid_file, std::process::id().to_string()).expect("write pid file");
        let dispatch_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let dispatch_count_srv = dispatch_count.clone();
        let cfg_for_srv = config_id.to_string();
        let fake_handle = tokio::spawn(async move {
            loop {
                let Ok((mut stream, _)) = listener.accept().await else {
                    break;
                };
                if read_frame(&mut stream).await.is_err() {
                    continue;
                }
                dispatch_count_srv.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                let resp = DaemonResponseFrame {
                    ok: true,
                    result: Some("daemon-handled-stats".to_string()),
                    error: None,
                    namespace_mismatch: false,
                    config_mismatch: false,
                    served_config_id: Some(cfg_for_srv.clone()),
                    version_mismatch: false,
                    daemon_protocol_version: PROTOCOL_VERSION,
                    metrics: None,
                };
                let payload = serde_json::to_vec(&resp).expect("serialize response frame");
                let _ = write_frame(&mut stream, &payload).await;
            }
        });

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let result = forward_or_spawn(&frame).await;

        match result {
            Some(Ok(ref body)) => {
                assert_eq!(
                    body, "daemon-handled-stats",
                    "request() must surface the daemon's response verbatim"
                );
            }
            other => {
                panic!("expected Some(Ok(_)) for a successful daemon round trip, got {other:?}")
            }
        }
        assert_eq!(
            dispatch_count.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "a successful daemon round trip must dispatch the real request EXACTLY \
             ONCE; a value other than 1 means forward_or_spawn retried or double-sent \
             the frame around a successful response"
        );

        fake_handle.abort();
        let _ = fake_handle.await;
        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── #667: readiness-timeout fallback must wait for boot quiescence ────────
    //
    // Before this fix, `forward_or_spawn`'s post-respawn readiness loop treated
    // a bare deadline elapse as "no daemon" and returned `None` (silent local
    // fallback) unconditionally — even if a concurrent process was still
    // holding the cold-boot guard (ADR-D3) running migrations / pack schema
    // plans (FTS DDL included). A local writer/searcher racing in at exactly
    // that moment could observe or create a partially-initialized
    // `notes`/`fts_notes` schema (#667).
    //
    // Nothing ever binds the socket in this test (no `NoSocket` outcome here
    // ever writes a real frame, so #644's at-most-once invariant is untouched
    // by this scenario) — `kill_and_respawn`'s own probe therefore sees
    // `NoSocket` too, classifies the (nonexistent) daemon as `Dead`, and calls
    // the real `spawn_daemon()` (which forks this test binary with `mcp
    // --daemon`; the child fails to parse those as libtest args and exits
    // almost immediately without ever binding anything — the same tolerated
    // pattern already used by
    // `try_forward_inner_returns_parse_failure_when_daemon_closes_without_response`
    // in this file).
    //
    // `SPAWN_COUNT` (already incremented for real inside `spawn_daemon`) is
    // used purely as a **synchronization signal**: it tells the background
    // "child boot" thread below that `kill_and_respawn`'s own (much shorter)
    // use of the recovery lock is in its final moments, so the boot thread's
    // blocking `acquire_daemon_boot_guard()` call queues immediately behind
    // it via the real `flock`, rather than racing it — the two never expect
    // to hold the lock at the same time, and the ordering between them is
    // never guessed at with a fixed sleep.
    //
    // Once the boot thread holds the guard (for `GUARD_HOLD`, deliberately
    // longer than `forward_or_spawn`'s fixed 5s readiness deadline), the fix
    // must block inside `wait_for_boot_quiescence_then_reprobe` until that
    // guard is released before it is allowed to decide "genuinely no daemon"
    // and return `None`. The elapsed-time assertion below is the
    // fail-if-reverted oracle: reverting the fence makes `forward_or_spawn`
    // return right at the 5s readiness deadline (measured from well before
    // the boot thread even starts holding the guard), strictly before
    // `GUARD_HOLD` has elapsed.
    #[tokio::test]
    #[serial]
    async fn forward_or_spawn_blocks_on_boot_quiescence_before_local_fallback() {
        clear_daemon_env();
        reset_counters();
        let dir = tempfile::tempdir().expect("tempdir");
        let sock = dir.path().join("khived.sock");
        let pid_file = dir.path().join("khived.pid");
        let lock_file = dir.path().join("khived.recovery.lock");

        std::env::set_var("KHIVE_SOCKET", &sock);
        std::env::set_var("KHIVE_PID", &pid_file);
        std::env::set_var("KHIVE_LOCK", &lock_file);
        std::env::remove_var("KHIVE_NO_DAEMON");

        let config_id = "packs=[kg];db=:memory:;embed=none;extra=[];backend=main";

        const GUARD_HOLD: std::time::Duration = std::time::Duration::from_secs(6);
        let boot_thread = std::thread::spawn(|| {
            let wait_deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
            while SPAWN_COUNT.load(std::sync::atomic::Ordering::SeqCst) == 0 {
                assert!(
                    std::time::Instant::now() < wait_deadline,
                    "kill_and_respawn never reached spawn_daemon() within 5s; \
                     the boot-holder thread has nothing to queue behind"
                );
                std::thread::sleep(std::time::Duration::from_millis(5));
            }
            // `kill_and_respawn` is at most a few milliseconds from dropping
            // its own lock guard at this point (spawn_daemon() is its last
            // action before returning) — this blocks on the real `flock`
            // until that happens, then holds it for GUARD_HOLD.
            let guard = khive_runtime::daemon::acquire_daemon_boot_guard()
                .expect("test boot holder must acquire the recovery lock");
            std::thread::sleep(GUARD_HOLD);
            drop(guard);
        });

        let frame = DaemonRequestFrame {
            ops: "stats()".to_string(),
            presentation: None,
            presentation_per_op: None,
            namespace: "test".to_string(),
            actor_id: None,
            visible_namespaces: Vec::new(),
            config_id: config_id.to_string(),
            protocol_version: PROTOCOL_VERSION,
            probe_only: false,
            metrics_only: false,
            format: None,
            format_per_op: None,
            from_wire: false,
        };

        let started = std::time::Instant::now();
        let result = forward_or_spawn(&frame).await;
        let elapsed = started.elapsed();

        boot_thread
            .join()
            .expect("boot-holder thread must not panic");

        assert!(
            result.is_none(),
            "genuinely no daemon after boot quiescence must fall back to local \
             dispatch (None), got {result:?}"
        );
        assert!(
            elapsed >= GUARD_HOLD,
            "forward_or_spawn must block until the cold-boot guard is released \
             before deciding to fall back locally — returned after {elapsed:?}, \
             faster than the {GUARD_HOLD:?} the boot guard was held, meaning it \
             (or a reverted version of this fix) would local-dispatch while \
             cold-boot schema init could still be in progress (#667)"
        );

        reset_counters();
        clear_daemon_env();
        std::env::remove_var("KHIVE_LOCK");
    }

    // ── daemon stderr log-file helpers (no process spawned) ───────────────────

    #[test]
    fn daemon_log_path_from_home_none_when_home_unset() {
        assert!(daemon_log_path_from_home(None).is_none());
    }

    #[test]
    fn daemon_log_path_from_home_joins_dot_khive_logs() {
        let home = std::ffi::OsStr::new("/home/example");
        let path = daemon_log_path_from_home(Some(home)).expect("home present");
        assert_eq!(
            path,
            std::path::PathBuf::from("/home/example/.khive/logs/khived.log")
        );
    }

    #[test]
    fn daemon_log_should_rotate_under_cap_is_false() {
        assert!(!daemon_log_should_rotate(100, 1000));
    }

    #[test]
    fn daemon_log_should_rotate_at_cap_is_true() {
        assert!(daemon_log_should_rotate(1000, 1000));
    }

    #[test]
    fn daemon_log_should_rotate_over_cap_is_true() {
        assert!(daemon_log_should_rotate(1001, 1000));
    }

    #[test]
    fn prepare_daemon_log_file_creates_dir_and_file_on_first_use() {
        let dir = tempfile::tempdir().expect("tempdir");
        let log_path = dir.path().join(".khive").join("logs").join("khived.log");

        let file = prepare_daemon_log_file_with_cap(&log_path, DAEMON_LOG_MAX_BYTES);

        assert!(file.is_some(), "must create dir + file on first use");
        assert!(log_path.exists());
    }

    #[test]
    fn prepare_daemon_log_file_leaves_existing_when_under_cap() {
        let dir = tempfile::tempdir().expect("tempdir");
        let log_dir = dir.path().join("logs");
        std::fs::create_dir_all(&log_dir).expect("create log dir");
        let log_path = log_dir.join("khived.log");
        std::fs::write(&log_path, b"existing content\n").expect("seed existing log");

        let file = prepare_daemon_log_file_with_cap(&log_path, 1_000_000);

        assert!(file.is_some());
        assert!(
            !log_dir.join("khived.log.1").exists(),
            "under-cap log must not be rotated"
        );
        let content = std::fs::read_to_string(&log_path).expect("read log");
        assert_eq!(
            content, "existing content\n",
            "append-open must preserve existing content"
        );
    }

    #[test]
    fn prepare_daemon_log_file_rotates_when_over_cap() {
        let dir = tempfile::tempdir().expect("tempdir");
        let log_dir = dir.path().join("logs");
        std::fs::create_dir_all(&log_dir).expect("create log dir");
        let log_path = log_dir.join("khived.log");
        std::fs::write(&log_path, vec![7u8; 20]).expect("seed oversized log");

        let file = prepare_daemon_log_file_with_cap(&log_path, 10);

        assert!(file.is_some());
        let backup = log_dir.join("khived.log.1");
        assert!(backup.exists(), "oversized log must rotate to .log.1");
        assert_eq!(
            std::fs::metadata(&backup).expect("backup metadata").len(),
            20,
            "backup must retain the original oversized content"
        );
        assert_eq!(
            std::fs::metadata(&log_path).expect("log metadata").len(),
            0,
            "post-rotation log must start fresh"
        );
    }

    #[test]
    fn prepare_daemon_log_file_rotation_replaces_prior_backup() {
        let dir = tempfile::tempdir().expect("tempdir");
        let log_dir = dir.path().join("logs");
        std::fs::create_dir_all(&log_dir).expect("create log dir");
        let log_path = log_dir.join("khived.log");
        let backup = log_dir.join("khived.log.1");
        std::fs::write(&backup, b"stale backup").expect("seed stale backup");
        std::fs::write(&log_path, vec![9u8; 20]).expect("seed oversized log");

        let file = prepare_daemon_log_file_with_cap(&log_path, 10);

        assert!(file.is_some());
        let backup_content = std::fs::read(&backup).expect("read backup");
        assert_eq!(
            backup_content,
            vec![9u8; 20],
            "rotation must replace the prior .log.1, not merge with it"
        );
    }

    #[test]
    fn prepare_daemon_log_file_returns_none_when_dir_creation_fails() {
        let dir = tempfile::tempdir().expect("tempdir");
        // Put a regular FILE where the log directory needs to be, so
        // create_dir_all cannot create a directory at that path.
        let blocker = dir.path().join("logs");
        std::fs::write(&blocker, b"not a directory").expect("seed blocker file");
        let log_path = blocker.join("khived.log");

        assert!(prepare_daemon_log_file_with_cap(&log_path, DAEMON_LOG_MAX_BYTES).is_none());
    }

    // ── #645: remove_daemon_paths_if_still_stale ownership recheck ───────────
    //
    // These mirror the existing `shutdown_cleanup_skips_when_*` tests in
    // `khive-runtime/src/daemon.rs` (owner-checked cleanup on the daemon's own
    // shutdown path) but exercise the CLIENT's stale-daemon cleanup instead:
    // between observing a PID as stale and reaching the unlink, a concurrent
    // starter that could not rely on the recovery lock alone (e.g. it failed
    // to acquire it) may have already claimed the rendezvous. Deterministic
    // state fabrication (not a sleep-based race) proves each skip condition.

    #[test]
    #[serial]
    fn remove_daemon_paths_if_still_stale_removes_when_pid_unchanged() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let pid_file = dir.path().join("khived.pid");
        let sock = dir.path().join("khived.sock");
        std::env::set_var("KHIVE_SOCKET", &sock);
        std::fs::write(&pid_file, "4242").expect("write pid file");
        std::fs::write(&sock, "stale socket placeholder").expect("write stale sock placeholder");

        remove_daemon_paths_if_still_stale(&pid_file, Some(4242));

        assert!(!pid_file.exists(), "unchanged pid file must be removed");
        assert!(!sock.exists(), "stale socket must be removed");
        clear_daemon_env();
    }

    #[test]
    #[serial]
    fn remove_daemon_paths_if_still_stale_skips_when_pid_file_changed() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let pid_file = dir.path().join("khived.pid");
        let sock = dir.path().join("khived.sock");
        std::env::set_var("KHIVE_SOCKET", &sock);
        // A replacement daemon already wrote its own (different) pid here.
        std::fs::write(&pid_file, "5555").expect("write replacement pid file");
        std::fs::write(&sock, "replacement socket placeholder").expect("write sock placeholder");

        remove_daemon_paths_if_still_stale(&pid_file, Some(4242));

        assert!(
            pid_file.exists(),
            "replacement daemon's pid file must survive when it no longer \
             matches the expected (pre-SIGTERM) pid"
        );
        assert!(
            sock.exists(),
            "replacement daemon's socket must survive alongside its pid file"
        );
        clear_daemon_env();
    }

    #[test]
    #[serial]
    fn remove_daemon_paths_if_still_stale_skips_when_socket_has_a_live_listener() {
        clear_daemon_env();
        let dir = tempfile::tempdir().expect("tempdir");
        let pid_file = dir.path().join("khived.pid");
        let sock = dir.path().join("khived.sock");
        std::env::set_var("KHIVE_SOCKET", &sock);
        std::fs::write(&pid_file, "4242").expect("write pid file matching expected_pid");
        // A replacement daemon already bound the socket path — even though the
        // pid file on disk has not been overwritten yet (e.g. its bind landed
        // just before its own pid-write).
        let _listener = std::os::unix::net::UnixListener::bind(&sock).expect("bind live socket");

        remove_daemon_paths_if_still_stale(&pid_file, Some(4242));

        assert!(
            sock.exists(),
            "a socket with a live listener must never be unlinked"
        );
        assert!(
            pid_file.exists(),
            "pid file must be left alone alongside the live socket"
        );
        clear_daemon_env();
    }

    // ── bridge self-heal on ProtocolMismatch (#714) ───────────────────────────

    #[test]
    fn resumed_generation_from_args_absent_is_none() {
        let argv = vec![
            "kkernel".to_string(),
            "mcp".to_string(),
            "--daemon".to_string(),
        ];
        assert_eq!(resumed_generation_from_args(argv.into_iter()), None);
    }

    #[test]
    fn resumed_generation_from_args_present_parses_value() {
        let argv = vec![
            "kkernel".to_string(),
            "mcp".to_string(),
            "--resumed-generation=1".to_string(),
        ];
        assert_eq!(resumed_generation_from_args(argv.into_iter()), Some(1));
    }

    #[test]
    fn resumed_generation_from_args_malformed_value_is_none() {
        let argv = vec![
            "kkernel".to_string(),
            "--resumed-generation=notanumber".to_string(),
        ];
        assert_eq!(resumed_generation_from_args(argv.into_iter()), None);
    }

    #[test]
    fn resumed_generation_from_args_takes_the_last_occurrence() {
        // Defensive: `reexec_in_place` filters any pre-existing marker before
        // appending its own, so duplicates should never occur in practice —
        // but if one ever did, the last one (the freshest exec's own marker)
        // must win, not the first.
        let argv = vec![
            "kkernel".to_string(),
            "--resumed-generation=1".to_string(),
            "--resumed-generation=2".to_string(),
        ];
        assert_eq!(resumed_generation_from_args(argv.into_iter()), Some(2));
    }

    // Cold-start non-regression (issue #714, self-heal test plan item 4): a
    // real `cargo test` process's own argv never carries the marker, so the
    // production entry point must resolve to `None` — the same guarantee
    // `KhiveMcpServer::serve_stdio` relies on to keep using the normal
    // `.serve()` handshake for every ordinary session.
    #[test]
    fn resumed_generation_is_none_for_a_normal_test_process() {
        assert_eq!(resumed_generation(), None);
    }

    // Loop-breaker guard rail (issue #714, self-heal test plan item 2): a
    // resumed generation that observes ProtocolMismatch again must take the
    // fallback, never a second exec. Pure decision function — no live
    // process needed.
    #[test]
    fn decide_mismatch_recovery_first_generation_schedules_reexec() {
        assert_eq!(
            decide_mismatch_recovery(None),
            MismatchRecovery::ReexecScheduled
        );
    }

    #[test]
    fn decide_mismatch_recovery_resumed_generation_drains_and_exits() {
        assert_eq!(
            decide_mismatch_recovery(Some(1)),
            MismatchRecovery::DrainAndExit
        );
    }

    // Ordering regression (issue #714, self-heal test plan item 1): arming a
    // self-heal action must never fire it — only `fire_pending_self_heal`
    // (the stdio transport's post-flush hook, exercised for real in
    // `crates/kkernel/tests/mcp_bridge_reexec_protocol_mismatch.rs`) may take
    // the armed action, and only after a flush has actually completed. This
    // is the exact bug the original toy-server evidence reproduced (an
    // in-flight response lost because `execv()` ran before the response
    // bytes were flushed) — these tests pin the "arm never fires" half of
    // that contract; the "fire only after a real flush" half is what the
    // live integration test proves.

    #[test]
    #[serial]
    fn schedule_reexec_on_mismatch_arms_without_firing() {
        reset_self_heal_counters();
        schedule_reexec_on_mismatch();
        assert_eq!(
            REEXEC_INVOKED_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "arming must never exec synchronously or eagerly"
        );
        fire_pending_self_heal();
        assert_eq!(
            REEXEC_INVOKED_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the armed action must fire exactly once it is taken"
        );
    }

    #[test]
    #[serial]
    fn schedule_drain_and_exit_arms_without_firing() {
        reset_self_heal_counters();
        schedule_drain_and_exit();
        assert_eq!(
            DRAIN_EXIT_INVOKED_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0,
            "arming must never exit synchronously or eagerly"
        );
        fire_pending_self_heal();
        assert_eq!(
            DRAIN_EXIT_INVOKED_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "the armed action must fire exactly once it is taken"
        );
    }

    #[test]
    #[serial]
    fn fire_pending_self_heal_is_a_no_op_when_nothing_is_armed() {
        reset_self_heal_counters();
        fire_pending_self_heal();
        assert_eq!(
            REEXEC_INVOKED_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0
        );
        assert_eq!(
            DRAIN_EXIT_INVOKED_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            0
        );
    }

    #[test]
    #[serial]
    fn fire_pending_self_heal_takes_the_armed_action_exactly_once() {
        reset_self_heal_counters();
        schedule_reexec_on_mismatch();
        fire_pending_self_heal();
        // A second flush completing after the action was already taken must
        // not re-fire it — `PENDING_SELF_HEAL` is `take()`n, not just read.
        fire_pending_self_heal();
        assert_eq!(
            REEXEC_INVOKED_COUNT.load(std::sync::atomic::Ordering::SeqCst),
            1,
            "must fire exactly once even if fire_pending_self_heal is called again"
        );
    }
}