khive-runtime 0.12.0

//! khived daemon server — persistent warm runtime over a Unix socket.
//!
//! The daemon binds `~/.khive/khived.sock`, accepts length-prefixed request
//! frames, dispatches them through a [`DaemonDispatch`] implementor, and serves
//! results back. It is transport-agnostic: the MCP crate provides the dispatch
//! impl, but any future client (CLI, HTTP gateway) can reuse this server.
//!
//! The client side (forwarding, auto-spawn) lives in the transport crate
//! (e.g. `khive-mcp`), not here.

use std::sync::Arc;

mod locks;
#[cfg(all(test, unix))]
use locks::open_lock_file;
#[cfg(unix)]
pub use locks::{
    acquire_daemon_boot_guard, acquire_recovery_lock, try_acquire_daemon_boot_guard_until,
    try_acquire_recoverer_lock_until, DaemonBootGuard,
};
#[cfg(unix)]
pub(crate) use locks::{peer_uid, uid_is_permitted};
#[cfg(unix)]
use locks::{socket_identity, SocketIdentity};

mod store_guard;
mod store_identity;
#[cfg(unix)]
use store_guard::ensure_claimed_parent_identity;
#[cfg(unix)]
pub use store_guard::{acquire_daemon_store_guards, bind_daemon_store_files, claim_stores};
pub use store_guard::{assert_daemon_store_identities, DaemonStoreGuard};
#[cfg(unix)]
pub use store_identity::claimed_daemon_store_identity;
mod supervisor_marker;
#[cfg(unix)]
pub use supervisor_marker::supervisor_marker_path;

#[cfg(unix)]
use std::io::Write as _;
#[cfg(unix)]
use std::os::unix::fs::{MetadataExt, PermissionsExt};
#[cfg(unix)]
use std::os::unix::io::AsRawFd;
#[cfg(all(test, unix))]
use std::path::PathBuf;

#[cfg(unix)]
use async_trait::async_trait;
#[cfg(unix)]
use libc;
#[cfg(unix)]
use serde::Deserialize;
#[cfg(unix)]
use tokio::io::AsyncReadExt;
#[cfg(all(test, unix))]
use tokio::io::AsyncWriteExt;
#[cfg(unix)]
use tokio::net::{UnixListener, UnixStream};

#[cfg(unix)]
use crate::pack::RequestIdentity;
#[cfg(unix)]
use khive_db::{run_checkpoint_task, CheckpointConfig, CheckpointLifecycleOwner, ConnectionPool};

// Unix-only: gated inside the file, because the error-constructor census
// admits only attribute-free submodule declarations.
mod exit_record;
mod lifecycle;
#[cfg(unix)]
use lifecycle::khive_dir;
#[cfg(test)]
use lifecycle::khive_root_from;
#[cfg(unix)]
use lifecycle::{
    current_supervisor_claim, default_socket_path, ensure_rendezvous_overrides_paired,
    read_supervisor_marker_claim, DaemonLifecycle, PID_PATH_ENV, SOCKET_PATH_ENV,
};
pub use lifecycle::{
    lock_path, volume_lock_dir, DaemonLifecyclePhase, DaemonLifecycleSnapshot, DaemonLifetime,
    DaemonOptions, DaemonShutdownReason, DaemonStartupReport, DEFAULT_DEMAND_IDLE_SECS,
};
#[cfg(unix)]
pub use lifecycle::{
    pid_path, recoverer_lock_path, rendezvous_split, socket_path, RendezvousSplit,
    SUPERVISOR_CLAIM_ENV,
};

mod load_limits;
#[cfg(unix)]
use load_limits::{admit_or_refuse_busy, ConnectionAdmission};
pub use load_limits::{
    recall_ledger_snapshot, track_recall_ledger_task, ConnectionCapSnapshot, RecallLedgerSnapshot,
};

/// Maximum frame size accepted in either direction.
pub const MAX_FRAME_BYTES: usize = 8 * 1024 * 1024;

/// Wire protocol version for the daemon IPC framing.
///
/// Increment this constant whenever the request or response frame shape
/// changes in a backward-incompatible way. The client sends its version
/// in every request; the daemon rejects mismatches with an explicit error
/// that names both sides so the operator knows exactly what to do
/// (`make local` rebuilds the client binary).
/// See `docs/api/daemon.md#protocol_version` for the version-by-version history.
pub const PROTOCOL_VERSION: u32 = 8;

/// Internal signal carried in a dispatch result until the daemon moves it to
/// response-frame metadata. It must never be sent in `result`: older v8
/// clients publish that string without inspecting its contents.
#[doc(hidden)]
pub const DAEMON_LEXICAL_TIMEOUT_MARKER: &str = "__khive_daemon_lexical_timeout";

const DEFAULT_DRAIN_TIMEOUT_SECS: u64 = 10;
/// An accepted local socket must finish its first frame within this window.
/// Dispatch deadlines start only after decoding, so they cannot reap peers
/// that connect and then stop sending request bytes.
#[cfg(unix)]
const INITIAL_FRAME_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

#[cfg(unix)]
fn next_accept_error_backoff(previous: Option<std::time::Duration>) -> std::time::Duration {
    previous
        .map(|delay| delay.saturating_mul(2))
        .unwrap_or_else(|| std::time::Duration::from_millis(10))
        .min(std::time::Duration::from_secs(1))
}

// ── wire types ────────────────────────────────────────────────────────────────

mod config_id;
#[cfg(test)]
use config_id::parse_config_id;
pub use config_id::{
    config_id_extra_embedder_exclusions, config_ids_compatible, first_config_mismatch_field,
};

mod wire_frames;
#[cfg(unix)]
pub use wire_frames::{read_frame, write_frame};
#[cfg(unix)]
use wire_frames::{read_initial_frame, take_daemon_lexical_timeout_marker};
pub use wire_frames::{
    CheckpointStoreMetrics, DaemonDispatchError, DaemonRequestFrame, DaemonResponseFrame,
    MetricsSnapshot, ERROR_DETAIL_NESTING_DEPTH_LIMIT,
};

// ── dispatch trait ────────────────────────────────────────────────────────────

/// Transport-agnostic dispatch interface for the daemon server.
///
/// The MCP crate implements this by dispatching through the shared request body
/// while honoring [`DaemonRequestFrame::from_wire`] (so subhandler visibility is
/// gated by request origin, not by transport); any future transport can do the
/// same.
#[cfg(unix)]
#[async_trait]
pub trait DaemonDispatch: Clone + Send + Sync + 'static {
    /// Named retained resources or unknown inventory that prevents idle exit.
    /// An implementor must explicitly account for its resources before retiring.
    fn idle_retirement_blockers(&self) -> Vec<String> {
        vec!["dispatcher_resource_inventory_unknown".to_owned()]
    }

    /// Describe syntax and loaded catalog membership without dispatching.
    fn plan(&self, ops: &str) -> String;

    /// Dispatch a verb-DSL request string and return the rendered result.
    ///
    /// `from_wire` carries the origin discriminator from
    /// [`DaemonRequestFrame::from_wire`]: when `true`, the implementor enforces
    /// verb visibility (rejects `Visibility::Subhandler` verbs); when `false`,
    /// the request is from a trusted operator surface and subhandlers pass.
    ///
    /// `identity` is the per-request identity context threaded from the frame
    /// (ADR-096): `Some(..)` when serving a request forwarded over the
    /// daemon socket (built from `frame.namespace` / `frame.actor_id` /
    /// `frame.visible_namespaces` by the connection handler), `None` for any
    /// other dispatch path. Implementors should mint the storage/gate token from
    /// `identity` when present and fall back to their own construction-baked
    /// identity when absent, so pure local (non-daemon) dispatch is unchanged.
    #[allow(clippy::too_many_arguments)]
    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<RequestIdentity>,
    ) -> Result<String, String>;

    /// Read-deadline ceiling for one request. The default is the operator
    /// ceiling; an implementor that understands the request may grant a
    /// longer bounded allowance (a long poll's declared wait plus a margin).
    fn request_read_timeout(&self, _ops: &str) -> std::time::Duration {
        khive_storage::request_read_timeout_from_env()
    }

    /// Preserve structured dispatch errors without breaking string-only implementors.
    #[allow(clippy::too_many_arguments)]
    async fn dispatch_with_error_detail(
        &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<RequestIdentity>,
    ) -> Result<String, DaemonDispatchError> {
        self.dispatch(
            ops,
            presentation,
            presentation_per_op,
            format,
            format_per_op,
            from_wire,
            identity,
        )
        .await
        .map_err(|message| DaemonDispatchError::new(message, None))
    }

    /// Warm every pack's in-memory state (ANN indexes, etc.).
    async fn warm_all(&self);

    /// The namespace this dispatcher was configured for.
    fn namespace(&self) -> &str;

    /// Fingerprint of this dispatcher's resolved runtime config (packs, db
    /// target, embedders). Used to reject forwarded requests from clients whose
    /// config differs, so a restricted client cannot dispatch through a broader
    /// daemon.
    fn config_id(&self) -> &str;

    /// Return the pool to use for background WAL checkpointing, if available.
    ///
    /// Implementors backed by a file-based SQLite database should return
    /// `Some(pool_arc)`. In-memory or test dispatchers that have no pool
    /// return `None` and the checkpoint task is not spawned.
    ///
    /// The default implementation returns `None`.
    fn pool_for_checkpoint(&self) -> Option<Arc<ConnectionPool>> {
        None
    }

    /// File-backed backend pools beyond [`Self::pool_for_checkpoint`]'s pool
    /// (ADR-091 Amendment 3): one checkpoint task is spawned per entry here,
    /// in addition to the one spawned for the primary pool, so a
    /// multi-backend deployment gets PASSIVE/TRUNCATE checkpointing and
    /// sidecar enumeration on every file-backed backend it wired, not only
    /// the main one.
    ///
    /// The default implementation returns an empty `Vec` — an implementor
    /// with only one backend (or none) needs no override.
    fn secondary_pools_for_checkpoint(&self) -> Vec<Arc<ConnectionPool>> {
        Vec::new()
    }

    /// Return the audit `EventStore` the checkpoint task should append
    /// ADR-094 lifecycle events (`CheckpointOutcomeRecorded`) to, if any.
    ///
    /// Mirrors [`Self::pool_for_checkpoint`]'s default-`None` shape: an
    /// implementor with no configured event store (or no pool at all) simply
    /// gets a checkpoint task that never appends events — the checkpoint
    /// task itself remains fully functional either way.
    ///
    /// The default implementation returns `None`.
    fn event_store_for_checkpoint(&self) -> Option<Arc<dyn khive_storage::EventStore>> {
        None
    }
}

#[cfg(unix)]
struct CheckpointTaskSpec {
    pool: Arc<ConnectionPool>,
    lifecycle_owner: Option<CheckpointLifecycleOwner>,
    is_main: bool,
}

/// Build checkpoint-task fan-out and designate one lifecycle owner.
///
/// The main checkpoint task owns lifecycle emission when it exists. If the
/// main backend is in-memory and therefore has no checkpoint task, the first
/// file-backed secondary owns emission instead. All remaining tasks are
/// explicit non-owners.
#[cfg(unix)]
fn checkpoint_task_specs(
    main_pool: Option<Arc<ConnectionPool>>,
    secondary_pools: Vec<Arc<ConnectionPool>>,
    event_store: Option<Arc<dyn khive_storage::EventStore>>,
    namespace: String,
) -> Vec<CheckpointTaskSpec> {
    let mut tasks = Vec::with_capacity(usize::from(main_pool.is_some()) + secondary_pools.len());
    if let Some(pool) = main_pool {
        tasks.push(CheckpointTaskSpec {
            pool,
            lifecycle_owner: None,
            is_main: true,
        });
    }
    tasks.extend(secondary_pools.into_iter().map(|pool| CheckpointTaskSpec {
        pool,
        lifecycle_owner: None,
        is_main: false,
    }));

    if let (Some(task), Some(event_store)) = (tasks.first_mut(), event_store) {
        task.lifecycle_owner = Some(CheckpointLifecycleOwner::new(event_store, namespace));
    }
    tasks
}

// ── tracked background tasks ─────────────────────────────────────────────────
//
// Pack handlers (e.g. memory.recall's ADR-081 serve-ledger append) fire
// fire-and-forget `tokio::spawn`ed work off the response path so the caller
// never waits on a cross-pack dispatch or a SQL write. Left untracked, that
// work is invisible to `drain()`: a SIGTERM landing between the response
// returning and the spawned task completing can abort it mid-flight with no
// log and no row. `track_background_task` gives such spawns a process-wide
// presence that `drain()` waits on, exactly like the `active` counter does
// for in-flight connections: the caller still only pays for the spawn +
// counter increment, never the task's own work.
/// Set once by the boot path that takes the daemon role, and never cleared: a
/// process that is not the warm daemon has no path to becoming one except exec.
static WARM_INDEX_HOST: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

/// Declare this process the warm index host. Called by the serve path as soon as
/// the daemon role is decided, before any runtime is built, so nothing warms
/// under the wrong answer.
pub fn mark_warm_index_host() {
    WARM_INDEX_HOST.store(true, std::sync::atomic::Ordering::Release);
}

/// Whether this process is the warm index host.
///
/// Building an ANN index from the full corpus is minutes of CPU and hundreds of
/// megabytes of segment rewrite, and it pays for itself only across a process
/// that outlives the request. A short-lived client that does it pays the whole
/// cost, discards the result at exit, and publishes a checkpoint that every
/// other reader on the root must then re-read. Consumers use this to decide
/// whether to build or to serve degraded and let the daemon build.
pub fn is_warm_index_host() -> bool {
    WARM_INDEX_HOST.load(std::sync::atomic::Ordering::Acquire)
}

static BACKGROUND_TASKS: std::sync::OnceLock<Arc<std::sync::atomic::AtomicUsize>> =
    std::sync::OnceLock::new();

fn background_tasks() -> &'static Arc<std::sync::atomic::AtomicUsize> {
    BACKGROUND_TASKS.get_or_init(|| Arc::new(std::sync::atomic::AtomicUsize::new(0)))
}

/// Decrements the shared background-task counter from `Drop`, so the count
/// comes back down whether the tracked future returns normally, panics, or
/// is cancelled — a plain post-`await` `fetch_sub` only covers the return
/// path and leaks the count forever on a panic, since unwinding skips every
/// statement after the panic point.
// ── outstanding background-task names ────────────────────────────────────────
//
// Names of the tasks the background counter is currently holding, so a drain
// timeout can say which ones held it open instead of printing a bare count.
// Registered and released at exactly the points the counter is incremented and
// decremented, and in the order that keeps the counter authoritative: the name
// goes in after the increment and comes out before the decrement, so a reported
// name always belongs to a task the counter already holds. The reverse ordering
// would let the warning name a task that had already finished, which is the one
// reading that would send someone looking in the wrong place.
static BACKGROUND_TASK_NAMES: std::sync::OnceLock<
    std::sync::Mutex<std::collections::HashMap<&'static str, usize>>,
> = std::sync::OnceLock::new();

fn background_task_names_registry(
) -> &'static std::sync::Mutex<std::collections::HashMap<&'static str, usize>> {
    BACKGROUND_TASK_NAMES.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
}

/// Name recorded for tasks spawned through the unnamed entry points, which
/// stay on the public API of a published crate.
pub const UNNAMED_BACKGROUND_TASK: &str = "unnamed";

fn register_background_task_name(name: &'static str) {
    let mut names = background_task_names_registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    *names.entry(name).or_insert(0) += 1;
}

fn release_background_task_name(name: &'static str) {
    let mut names = background_task_names_registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if let Some(count) = names.get_mut(name) {
        *count -= 1;
        if *count == 0 {
            names.remove(name);
        }
    }
}

/// Names of the in-flight tracked background tasks, sorted and deduplicated.
/// A diagnostic beside [`background_task_count`], never a substitute for it:
/// the count is what drain waits on.
pub fn background_task_names() -> Vec<String> {
    let names = background_task_names_registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let mut out: Vec<String> = names.keys().map(|name| (*name).to_string()).collect();
    out.sort();
    out
}

#[cfg(unix)]
fn idle_retirement_blockers<D: DaemonDispatch>(dispatcher: &D) -> Vec<String> {
    let mut blockers = dispatcher.idle_retirement_blockers();
    if !khive_storage::tx_registry::snapshot().is_empty() {
        blockers.push("open_sql_transaction".to_owned());
    }
    blockers.extend(
        active_phase_names()
            .into_iter()
            .map(|name| format!("active_phase:{name}")),
    );
    let count = background_task_count();
    let names = background_task_names_registry()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if names.values().sum::<usize>() != count {
        blockers.push("tracked_worker_inventory_unsettled".to_owned());
    }
    // Only these inspected loops maintain replaceable caches/checkpoint state.
    // Their tracked lifetime still participates in the final drain.
    for name in names.keys() {
        if !matches!(
            *name,
            "wal_checkpoint" | "memory_ann_rotation_watch" | "knowledge_ann_rotation_watch"
        ) {
            blockers.push(format!("unsettled_worker:{name}"));
        }
    }
    blockers.sort();
    blockers.dedup();
    blockers
}

#[cfg(unix)]
async fn wait_for_idle<D: DaemonDispatch>(dispatcher: &D, lifecycle: &DaemonLifecycle) {
    if lifecycle.options.lifetime == DaemonLifetime::Persistent {
        std::future::pending::<()>().await;
    }
    loop {
        if lifecycle.try_idle(|| idle_retirement_blockers(dispatcher)) {
            return;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
}

struct BackgroundTaskGuard {
    counter: Arc<std::sync::atomic::AtomicUsize>,
    name: &'static str,
}

impl Drop for BackgroundTaskGuard {
    fn drop(&mut self) {
        release_background_task_name(self.name);
        self.counter
            .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
    }
}

/// Spawn a task that daemon shutdown's `drain()` waits for and return its join
/// handle. Retaining the handle lets boot coordinators form an explicit barrier;
/// dropping it deliberately detaches the task while the background counter still
/// keeps daemon drain aware of its lifetime. The decrement happens via
/// `BackgroundTaskGuard`'s `Drop`, including panic and cancellation paths.
pub fn spawn_tracked_task<F, T>(fut: F) -> tokio::task::JoinHandle<T>
where
    F: std::future::Future<Output = T> + Send + 'static,
    T: Send + 'static,
{
    spawn_named_tracked_task(UNNAMED_BACKGROUND_TASK, fut)
}

/// [`spawn_tracked_task`] with a name that a drain timeout can print.
///
/// The name is a short static string describing the task, never a formatted or
/// caller-supplied value: it is read by an operator staring at a shutdown that
/// would not finish, so it is a label for a call site, not a record of one
/// occurrence.
pub fn spawn_named_tracked_task<F, T>(name: &'static str, fut: F) -> tokio::task::JoinHandle<T>
where
    F: std::future::Future<Output = T> + Send + 'static,
    T: Send + 'static,
{
    background_tasks().fetch_add(1, std::sync::atomic::Ordering::SeqCst);
    register_background_task_name(name);
    let guard = BackgroundTaskGuard {
        counter: background_tasks().clone(),
        name,
    };
    tokio::spawn(async move {
        let _guard = guard;
        fut.await
    })
}

/// Spawn a fire-and-forget task through [`spawn_tracked_task`].
///
/// Callers that need a boot or shutdown barrier should retain and await the
/// returned handle from [`spawn_tracked_task`] instead of detaching it here.
pub fn track_background_task<F>(fut: F)
where
    F: std::future::Future<Output = ()> + Send + 'static,
{
    track_named_background_task(UNNAMED_BACKGROUND_TASK, fut);
}

/// [`track_background_task`] with a name that a drain timeout can print. See
/// [`spawn_named_tracked_task`] for what belongs in the name.
pub fn track_named_background_task<F>(name: &'static str, fut: F)
where
    F: std::future::Future<Output = ()> + Send + 'static,
{
    drop(spawn_named_tracked_task(name, fut));
}

/// Current count of in-flight tasks started via [`track_background_task`].
/// Exposed for tests; `drain()` reads the shared counter directly.
pub fn background_task_count() -> usize {
    background_tasks().load(std::sync::atomic::Ordering::SeqCst)
}

/// Process-wide daemon shutdown signal (ADR-119).
///
/// Cancelled exactly once, when the daemon's unified shutdown future resolves
/// — before `drain()` begins waiting on tracked tasks — so long-running
/// daemon components supervised outside this module observe shutdown through
/// the same path the daemon itself does, rather than inventing their own.
/// Clones share the underlying token; child tokens derived from it are
/// cancelled transitively.
///
/// In non-daemon processes the token simply never fires.
pub fn daemon_shutdown_token() -> tokio_util::sync::CancellationToken {
    static TOKEN: std::sync::OnceLock<tokio_util::sync::CancellationToken> =
        std::sync::OnceLock::new();
    TOKEN
        .get_or_init(tokio_util::sync::CancellationToken::new)
        .clone()
}

// ── active background phase names (ADR-103) ──────────────────────────────────
//
// A lightweight, best-effort process-wide gauge of which named background
// phases (e.g. `ann_warm`) are in flight right now, read by `comm.health`'s
// resource self-report so a caller can see "what is the daemon doing" at a
// glance without correlating timestamps across the event log itself. Counted
// per name rather than boolean, since more than one occurrence of the same
// named phase can legitimately overlap (e.g. two embedding models warming
// concurrently) — the name only drops out of the reported set once every
// concurrent occurrence has ended.
static ACTIVE_PHASES: std::sync::OnceLock<
    std::sync::Mutex<std::collections::HashMap<String, usize>>,
> = std::sync::OnceLock::new();

fn active_phases() -> &'static std::sync::Mutex<std::collections::HashMap<String, usize>> {
    ACTIVE_PHASES.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
}

/// RAII guard for one occurrence of a named background phase. Increments the
/// phase's count on creation (see [`register_active_phase`]); decrements on
/// `Drop`, so the count comes back down whether the guarded work returns
/// normally, panics, or is cancelled — the same rationale as
/// `BackgroundTaskGuard` above.
pub struct PhaseGuard {
    name: String,
}

impl Drop for PhaseGuard {
    fn drop(&mut self) {
        let mut map = active_phases()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if let Some(count) = map.get_mut(&self.name) {
            *count -= 1;
            if *count == 0 {
                map.remove(&self.name);
            }
        }
    }
}

/// Register one occurrence of a named background phase as currently active.
/// Returns a guard: drop it (or let it fall out of scope) when the phase
/// ends. Best-effort process-wide gauge only, read by `comm.health` — never
/// load-bearing for correctness.
pub fn register_active_phase(name: &str) -> PhaseGuard {
    let mut map = active_phases()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    *map.entry(name.to_string()).or_insert(0) += 1;
    PhaseGuard {
        name: name.to_string(),
    }
}

/// Currently active background-phase names, sorted for deterministic output.
/// Empty when no tracked phase is in flight.
pub fn active_phase_names() -> Vec<String> {
    let map = active_phases()
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    let mut names: Vec<String> = map.keys().cloned().collect();
    names.sort();
    names
}

// ── server ────────────────────────────────────────────────────────────────────

/// Build a point-in-time [`MetricsSnapshot`] of this process's server-side
/// gauges. Called only from `handle_conn`'s `metrics_only` arm — a
/// process-global, read-only assembly with no side effects of its own.
/// See `docs/api/daemon.md#build_metrics_snapshot` for where each gauge is sourced
/// from and why.
#[cfg(unix)]
fn build_metrics_snapshot<D: DaemonDispatch>(dispatcher: &D) -> MetricsSnapshot {
    let open_tx_count = khive_storage::tx_registry::snapshot().len();
    // ADR-091 Amendment 3: deliberately the process-wide aggregate, not a
    // backend-attributed view — this gauge reports "oldest pinned tx in this
    // process" across every wired backend, not an attribution claim about
    // which database it belongs to. Attribution consumers (the session sweep,
    // the per-backend checkpoint tasks) use the scoped `oldest_for` views in
    // khive-db instead.
    let (oldest_pinned_tx_micros, oldest_pinned_tx_label) =
        match khive_storage::tx_registry::oldest() {
            Some((_id, age, label)) => (Some(age.as_micros() as u64), label),
            None => (None, None),
        };

    let checkpoint_pool = dispatcher.pool_for_checkpoint();
    let mut secondary_index = 0;
    let wal_checkpoint_stores = checkpoint_task_specs(
        checkpoint_pool.clone(),
        dispatcher.secondary_pools_for_checkpoint(),
        None,
        String::new(),
    )
    .into_iter()
    .map(|task| {
        let (store_id, role) = if task.is_main {
            ("main".to_string(), "main".to_string())
        } else {
            let store_id = format!("secondary:{secondary_index}");
            secondary_index += 1;
            (store_id, "secondary".to_string())
        };
        CheckpointStoreMetrics {
            store_id,
            role,
            database: task
                .pool
                .canonical_path()
                .and_then(std::path::Path::file_name)
                .map(|name| name.to_string_lossy().into_owned()),
            timing: khive_db::checkpoint::checkpoint_timing(&task.pool),
        }
    })
    .collect();
    let routine_wal = checkpoint_pool
        .as_deref()
        .and_then(khive_db::checkpoint::routine_wal_observation);
    let writer_stages = checkpoint_pool
        .as_deref()
        .and_then(khive_db::writer_task::last_writer_stage_observation);
    let (write_queue_depth, write_queue_capacity) = checkpoint_pool
        .as_ref()
        .and_then(|pool| pool.writer_task_handle().ok().flatten())
        .map(|handle| (Some(handle.queue_depth()), Some(handle.capacity())))
        .unwrap_or((None, None));

    MetricsSnapshot {
        lifecycle: None,
        wal_pages: routine_wal.as_ref().map(|sample| sample.log_frames),
        wal_log_frames: routine_wal.as_ref().map(|sample| sample.log_frames),
        wal_checkpointed_frames: routine_wal
            .as_ref()
            .map(|sample| sample.checkpointed_frames),
        wal_pending_frames: routine_wal.as_ref().map(|sample| sample.pending_frames),
        wal_physical_bytes: routine_wal
            .as_ref()
            .and_then(|sample| sample.physical_wal_bytes),
        wal_observed_at_unix_ms: routine_wal
            .as_ref()
            .map(|sample| sample.observed_at_unix_ms),
        wal_checkpoint_stores,
        wal_truncate_attempts: khive_db::checkpoint::truncate_attempts(),
        wal_truncate_consecutive_failures: khive_db::checkpoint::truncate_consecutive_failures(),
        wal_checkpoint_skipped_ticks: khive_db::checkpoint::checkpoint_skipped_ticks(),
        wal_checkpoint_consecutive_skips: khive_db::checkpoint::checkpoint_consecutive_skips(),
        wal_checkpoint_last_skip_wal_pages: khive_db::checkpoint::checkpoint_last_skip_wal_pages(),
        oldest_pinned_tx_micros,
        oldest_pinned_tx_label,
        open_tx_count,
        write_queue_depth,
        write_queue_capacity,
        write_last_queue_wait_micros: writer_stages
            .as_ref()
            .map(|sample| sample.queue_wait_micros),
        write_last_transaction_acquire_micros: writer_stages
            .as_ref()
            .map(|sample| sample.transaction_acquire_micros),
        write_last_body_micros: writer_stages.as_ref().map(|sample| sample.body_micros),
        write_last_commit_micros: writer_stages.as_ref().map(|sample| sample.commit_micros),
        write_last_total_micros: writer_stages.as_ref().map(|sample| sample.total_micros),
        write_last_observed_at_unix_ms: writer_stages
            .as_ref()
            .map(|sample| sample.observed_at_unix_ms),
        connections: None,
        recall_ledger: Some(recall_ledger_snapshot()),
    }
}

#[cfg(unix)]
async fn write_response_frame<W>(stream: &mut W, payload: &[u8]) -> std::io::Result<()>
where
    W: tokio::io::AsyncWrite + Unpin,
{
    tokio::time::timeout(INITIAL_FRAME_READ_TIMEOUT, write_frame(stream, payload))
        .await
        .map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                "daemon response write timed out",
            )
        })?
}

#[cfg(unix)]
async fn wait_for_peer_disconnect(read: &mut tokio::net::unix::OwnedReadHalf) {
    let mut byte = [0u8; 1];
    // One request is admitted per connection. EOF, a read error, or any
    // subsequent byte all make this connection no longer a valid response
    // peer, so the first completed read is the entire observation.
    let _ = read.read(&mut byte).await;
}

#[cfg(all(unix, test))]
async fn handle_conn<D: DaemonDispatch>(stream: UnixStream, dispatcher: D) {
    handle_conn_with_shutdown(
        stream,
        dispatcher,
        None,
        tokio::time::Instant::now() + INITIAL_FRAME_READ_TIMEOUT,
    )
    .await;
}

#[cfg(all(unix, feature = "fault-injection"))]
#[doc(hidden)]
pub async fn handle_conn_for_test<D: DaemonDispatch>(stream: UnixStream, dispatcher: D) {
    handle_conn_with_shutdown(
        stream,
        dispatcher,
        None,
        tokio::time::Instant::now() + INITIAL_FRAME_READ_TIMEOUT,
    )
    .await;
}

#[cfg(unix)]
fn plan_frame_companion(raw: &[u8]) -> Option<&'static str> {
    let value: serde_json::Value = serde_json::from_slice(raw).ok()?;
    if value.get("plan").and_then(serde_json::Value::as_bool) != Some(true)
        || value
            .get("protocol_version")
            .and_then(serde_json::Value::as_u64)
            != Some(u64::from(PROTOCOL_VERSION))
    {
        return None;
    }
    [
        "presentation",
        "presentation_per_op",
        "format",
        "format_per_op",
        "request_id",
    ]
    .into_iter()
    .find(|field| value.get(*field).is_some())
}

#[cfg(all(
    unix,
    any(test, feature = "fault-injection", feature = "test-internals")
))]
async fn handle_conn_with_shutdown<D: DaemonDispatch>(
    stream: UnixStream,
    dispatcher: D,
    shutdown: Option<tokio::sync::watch::Receiver<bool>>,
    initial_frame_deadline: tokio::time::Instant,
) {
    handle_conn_with_lifecycle(stream, dispatcher, shutdown, initial_frame_deadline, None).await;
}

#[cfg(unix)]
async fn handle_conn_with_lifecycle<D: DaemonDispatch>(
    mut stream: UnixStream,
    dispatcher: D,
    shutdown: Option<tokio::sync::watch::Receiver<bool>>,
    initial_frame_deadline: tokio::time::Instant,
    lifecycle: Option<Arc<DaemonLifecycle>>,
) {
    let mut ordinary_admission = None;
    let production_shutdown = shutdown.is_some();
    // A handover is delivered over the probed connection. The production
    // listener enforces same-uid admission, and direct handler tests cannot
    // self-signal because they do not supply the daemon shutdown receiver.
    let handover_peer_allowed = peer_uid(&stream)
        .ok()
        .is_some_and(|uid| uid == unsafe { libc::geteuid() } as u32);
    let (local_shutdown_tx, local_shutdown_rx) = tokio::sync::watch::channel(false);
    let shutdown = shutdown.unwrap_or(local_shutdown_rx);
    // Keeps the fallback receiver open in direct/test calls. Production owns
    // a sender at the daemon-run scope and passes its receiver above.
    let _local_shutdown_tx = local_shutdown_tx;
    let raw = match read_initial_frame(&mut stream, initial_frame_deadline).await {
        Ok(r) => r,
        Err(e) => {
            tracing::debug!(error = %e, "failed to read daemon request frame");
            return;
        }
    };
    #[derive(Deserialize)]
    struct SupervisorRequestEnvelope {
        #[serde(flatten)]
        frame: DaemonRequestFrame,
        #[serde(default)]
        supervisor_handover: bool,
    }
    let decoded: Result<SupervisorRequestEnvelope, _> = serde_json::from_slice(&raw);
    if decoded.as_ref().ok().is_none_or(|item| item.frame.plan) {
        if let Some(field) = plan_frame_companion(&raw) {
            let response = DaemonResponseFrame {
                ok: false,
                result: None,
                error: Some(format!(
                    "invalid_params: plan=true cannot be combined with {field}"
                )),
                error_detail: Some(serde_json::json!({
                    "kind": "protocol",
                    "code": "invalid_params",
                    "message": format!("plan=true cannot be combined with {field}"),
                    "domain_disposition": crate::DomainDisposition::NotCommitted.as_str(),
                })),
                namespace_mismatch: false,
                config_mismatch: false,
                served_config_id: Some(dispatcher.config_id().to_string()),
                version_mismatch: false,
                daemon_protocol_version: PROTOCOL_VERSION,
                metrics: None,
                request_id: None,
            };
            if let Ok(payload) = serde_json::to_vec(&response) {
                if let Err(error) = write_response_frame(&mut stream, &payload).await {
                    tracing::debug!(%error, "failed to write plan envelope refusal");
                }
            }
            return;
        }
    }
    let (frame, handover_requested) = match decoded {
        Ok(item) => (item.frame, item.supervisor_handover),
        Err(e) => {
            tracing::debug!(error = %e, "failed to decode daemon request frame");
            return;
        }
    };
    let supervisor_probe = frame.probe_only;
    let handover_accepted = handover_requested
        && supervisor_probe
        && production_shutdown
        && handover_peer_allowed
        && frame.protocol_version == PROTOCOL_VERSION
        && !frame.plan
        && !frame.metrics_only
        && frame.ops.is_empty()
        && read_supervisor_marker_claim().is_some_and(|(pid, _)| pid != std::process::id());
    let (mut peer_read, mut peer_write) = stream.into_split();

    let served_config_id = Some(dispatcher.config_id().to_string());
    let resp = if frame.protocol_version != PROTOCOL_VERSION {
        let msg = format!(
            "daemon protocol mismatch: client={} daemon={} — \
             rebuild/update the client binary (make local)",
            frame.protocol_version, PROTOCOL_VERSION,
        );
        tracing::warn!(
            client_version = frame.protocol_version,
            daemon_version = PROTOCOL_VERSION,
            "daemon protocol version mismatch"
        );
        DaemonResponseFrame {
            ok: false,
            result: None,
            error: Some(msg.clone()),
            error_detail: Some(serde_json::json!({
                "kind": "protocol",
                "code": "version_mismatch",
                "message": msg,
                "domain_disposition": crate::DomainDisposition::Unknown.as_str(),
            })),
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id,
            // A client below this protocol is a bridge that predates the binary this
            // daemon was spawned from. Through protocol 5 the bridge treats an
            // explicit `version_mismatch` from a higher-numbered daemon as a terminal
            // error it repeats on every request, and re-execs itself onto the on-disk
            // binary only for the implicit shape: an unequal `daemon_protocol_version`
            // with the flag clear. Answering older clients in that shape, still
            // refused and still carrying the code in `error_detail`, lets every
            // pre-swap bridge replace itself on its first request instead of staying
            // refused until a person reconnects the session. A client above this
            // protocol keeps the explicit flag. Remove once no live bridge predates
            // the two-direction re-exec in khive-mcp (an inode census of `kkernel mcp`
            // processes before the swap): bridges built with it no longer read the
            // flag, but a bump while older bridges still run must keep this shape so
            // they replace themselves too.
            version_mismatch: frame.protocol_version > PROTOCOL_VERSION,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
            request_id: frame.request_id,
        }
    } else if handover_requested {
        if handover_accepted {
            DaemonResponseFrame {
                ok: true,
                result: None,
                error: None,
                error_detail: None,
                namespace_mismatch: false,
                config_mismatch: false,
                served_config_id,
                version_mismatch: false,
                daemon_protocol_version: PROTOCOL_VERSION,
                metrics: None,
                request_id: frame.request_id,
            }
        } else {
            DaemonResponseFrame {
                ok: false,
                result: None,
                error: Some("supervisor handover refused".to_string()),
                error_detail: Some(serde_json::json!({
                    "kind": "protocol",
                    "code": "supervisor_handover_refused",
                    "message": "supervisor handover refused",
                    "domain_disposition": crate::DomainDisposition::NotCommitted.as_str(),
                })),
                namespace_mismatch: false,
                config_mismatch: false,
                served_config_id,
                version_mismatch: false,
                daemon_protocol_version: PROTOCOL_VERSION,
                metrics: None,
                request_id: frame.request_id,
            }
        }
    } else if frame.metrics_only && !frame.plan {
        // Process-global gauge read: namespace/config-agnostic, so this is
        // handled BEFORE the `config_id` equality reject below (unlike every
        // other arm) — a metrics probe must work regardless of which
        // client's config is asking, since it never touches the dispatcher's
        // packs/db/embed registry. READ-ONLY: builds a snapshot and returns
        // immediately, never reaching the ops-dispatch arm.
        DaemonResponseFrame {
            ok: true,
            result: None,
            error: None,
            error_detail: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id,
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: Some({
                let mut metrics = build_metrics_snapshot(&dispatcher);
                metrics.lifecycle = lifecycle.as_ref().map(|state| {
                    let mut snapshot = state.snapshot();
                    snapshot.idle_blockers = idle_retirement_blockers(&dispatcher);
                    snapshot
                });
                metrics.connections = lifecycle.as_ref().map(|state| state.connections.snapshot());
                metrics
            }),
            request_id: frame.request_id,
        }
    // There is no `frame.namespace != dispatcher.namespace()` reject here.
    // The daemon accepts and serves the request under the frame's own
    // identity (namespace / actor / visible_namespaces, built into a
    // `RequestIdentity` below) over its one shared warm registry, rather
    // than rejecting a differently-attributed same-uid connection to a cold
    // local-dispatch fallback. `config_id`: which governs packs/db/embed
    // coherence for the shared warm engine: remains a hard reject for every
    // field; only legacy IDs allow a superset of the client's extra embedders.
    } else if !config_ids_compatible(&frame.config_id, dispatcher.config_id()) {
        DaemonResponseFrame {
            ok: false,
            result: None,
            error: None,
            error_detail: Some(serde_json::json!({
                "kind": "protocol",
                "code": "config_mismatch",
                "message": "daemon configuration does not match the request",
                "domain_disposition": crate::DomainDisposition::NotCommitted.as_str(),
            })),
            namespace_mismatch: false,
            config_mismatch: true,
            served_config_id,
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
            request_id: frame.request_id,
        }
    } else if frame.plan {
        DaemonResponseFrame {
            ok: true,
            result: Some(dispatcher.plan(&frame.ops)),
            error: None,
            error_detail: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id,
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
            request_id: None,
        }
    } else if frame.probe_only {
        // Probe-only request: identity checks passed; return immediately without
        // dispatching any verb. The client uses this to confirm the daemon is
        // alive and identity-matching without triggering any mutation.
        DaemonResponseFrame {
            ok: true,
            result: None,
            error: None,
            error_detail: None,
            namespace_mismatch: false,
            config_mismatch: false,
            served_config_id,
            version_mismatch: false,
            daemon_protocol_version: PROTOCOL_VERSION,
            metrics: None,
            request_id: frame.request_id,
        }
    } else {
        if let Some(lifecycle) = &lifecycle {
            ordinary_admission = lifecycle.admit();
            if ordinary_admission.is_none() {
                let refusal = DaemonResponseFrame {
                    ok: false,
                    result: None,
                    error: Some("daemon is draining; request was not admitted".to_owned()),
                    error_detail: Some(serde_json::json!({
                        "kind": "runtime", "code": "daemon_draining",
                        "domain_disposition": crate::DomainDisposition::NotCommitted.as_str(),
                    })),
                    request_id: frame.request_id,
                    daemon_protocol_version: PROTOCOL_VERSION,
                    namespace_mismatch: false,
                    config_mismatch: false,
                    served_config_id: Some(dispatcher.config_id().to_owned()),
                    version_mismatch: false,
                    metrics: None,
                };
                if let Ok(payload) = serde_json::to_vec(&refusal) {
                    let _ = write_response_frame(&mut peer_write, &payload).await;
                }
                return;
            }
        }
        // Build the per-request identity context from the frame so the
        // implementor mints the storage/gate token from the CALLER's
        // identity, not the dispatcher's own construction-baked scalars.
        // This is always `Some` here: every frame that reaches this arm
        // carries a `namespace` (required on the wire) plus whatever
        // `actor_id`/`visible_namespaces` the client resolved (defaulting to
        // `None`/`vec![]` for an older, field-absent payload, which is
        // exactly the prior anonymous/no-extra-visibility behavior).
        // The caller's actor namespace joins default reads where the registry
        // mints the token (ADR-007 Rev 4 Rule 3b), the one seam every identity
        // path shares; the frame's list is forwarded as sent.
        let identity = RequestIdentity {
            namespace: frame.namespace.clone(),
            actor_id: frame.actor_id.clone(),
            visible_namespaces: frame.visible_namespaces.clone(),
            process_ref: frame.process_ref.clone(),
            request_id: frame.request_id,
        };
        tracing::debug!(
            request_id = frame.request_id,
            "daemon RequestIdentity constructed"
        );
        let (read_cancel_tx, read_cancel_rx) = tokio::sync::watch::channel(false);
        // The connection's own ceiling nests inside the dispatcher's, and a
        // nested scope keeps the earlier deadline, so the allowance must be
        // granted here or a long poll times out at the operator ceiling.
        let read_timeout = dispatcher.request_read_timeout(&frame.ops);
        let excluded_embedder_names =
            config_id_extra_embedder_exclusions(&frame.config_id, dispatcher.config_id())
                .expect("compatible configuration ids must expose their extra embedder sets");
        let dispatch = crate::runtime::scope_request_embedder_exclusions(
            excluded_embedder_names,
            khive_storage::scope_request_read_cancellation(
                shutdown,
                khive_storage::scope_request_read_cancellation(
                    read_cancel_rx,
                    khive_storage::scope_request_read_deadline(
                        read_timeout,
                        dispatcher.dispatch_with_error_detail(
                            frame.ops,
                            frame.presentation,
                            frame.presentation_per_op,
                            frame.format,
                            frame.format_per_op,
                            frame.from_wire,
                            Some(identity),
                        ),
                    ),
                ),
            ),
        );
        tokio::pin!(dispatch);
        let dispatch_result = tokio::select! {
            result = &mut dispatch => result,
            _ = wait_for_peer_disconnect(&mut peer_read) => {
                let _ = read_cancel_tx.send(true);
                dispatch.await
            }
        };
        match dispatch_result {
            Ok(result) => {
                let (result, detail) = take_daemon_lexical_timeout_marker(result);
                DaemonResponseFrame {
                    ok: true,
                    result: Some(result),
                    error: None,
                    error_detail: detail,
                    namespace_mismatch: false,
                    config_mismatch: false,
                    served_config_id,
                    version_mismatch: false,
                    daemon_protocol_version: PROTOCOL_VERSION,
                    metrics: None,
                    request_id: frame.request_id,
                }
            }
            Err(error) => {
                let error = DaemonDispatchError::new(error.message, Some(error.error_detail));
                DaemonResponseFrame {
                    ok: false,
                    result: None,
                    error: Some(error.message),
                    error_detail: Some(error.error_detail),
                    namespace_mismatch: false,
                    config_mismatch: false,
                    served_config_id,
                    version_mismatch: false,
                    daemon_protocol_version: PROTOCOL_VERSION,
                    metrics: None,
                    request_id: frame.request_id,
                }
            }
        }
    };

    let payload = if supervisor_probe {
        serde_json::to_value(&resp).and_then(|mut value| {
            if let Some(claim) = current_supervisor_claim() {
                value["supervisor_claim"] = serde_json::Value::String(claim);
            }
            if handover_accepted {
                value["supervisor_handover_accepted"] = serde_json::Value::Bool(true);
            }
            serde_json::to_vec(&value)
        })
    } else {
        serde_json::to_vec(&resp)
    };
    let mut handover_ack_written = false;
    match payload {
        Ok(payload) => {
            if payload.len() > MAX_FRAME_BYTES {
                // The serialized response exceeds the IPC frame cap.  Send a
                // small explicit error frame so the client can distinguish a
                // per-request payload-size failure from a daemon crash.  A
                // client that receives this error frame will NOT trigger
                // stale-daemon kill/respawn (ParseFailure requires a read_frame
                // error, not an ok=false result).
                tracing::warn!(
                    bytes = payload.len(),
                    limit = MAX_FRAME_BYTES,
                    "daemon response exceeds MAX_FRAME_BYTES; sending explicit error frame"
                );
                let message = format!(
                    "response too large: {} bytes exceeds {} byte IPC cap",
                    payload.len(),
                    MAX_FRAME_BYTES,
                );
                // One frame may aggregate successful, failed, and aborted operations.
                let err_resp = DaemonResponseFrame {
                    ok: false,
                    result: None,
                    error: Some(message.clone()),
                    error_detail: Some(serde_json::json!({
                        "kind": "transport",
                        "code": "response_frame_size_limit",
                        "message": message,
                        "domain_disposition": crate::DomainDisposition::Unknown.as_str(),
                    })),
                    namespace_mismatch: false,
                    config_mismatch: false,
                    served_config_id: resp.served_config_id,
                    version_mismatch: false,
                    daemon_protocol_version: PROTOCOL_VERSION,
                    metrics: None,
                    request_id: resp.request_id,
                };
                if let Ok(err_payload) = serde_json::to_vec(&err_resp) {
                    if let Err(e) = write_response_frame(&mut peer_write, &err_payload).await {
                        tracing::debug!(error = %e, "failed to write oversized-response error frame");
                    }
                }
            } else {
                match write_response_frame(&mut peer_write, &payload).await {
                    Ok(()) => handover_ack_written = true,
                    Err(e) => tracing::debug!(error = %e, "failed to write daemon response frame"),
                }
            }
        }
        Err(e) => tracing::warn!(error = %e, "failed to serialize daemon response frame"),
    }
    if handover_accepted && handover_ack_written {
        // Signal this process, not a PID observed earlier over a socket.
        // Production installed its SIGTERM handler before binding the socket.
        if unsafe { libc::raise(libc::SIGTERM) } != 0 {
            tracing::error!(error = %std::io::Error::last_os_error(), "self-directed handover signal failed");
        }
    }
    drop(ordinary_admission);
}

/// An accepted connection owns a drain slot from the accept loop until its
/// handler future is dropped. The claim happens before spawning so shutdown
/// cannot observe zero in the gap between `accept()` and the task's first poll.
#[cfg(unix)]
struct ActiveConnectionGuard {
    active: Arc<std::sync::atomic::AtomicUsize>,
}

#[cfg(unix)]
impl ActiveConnectionGuard {
    fn claim(active: Arc<std::sync::atomic::AtomicUsize>) -> Self {
        active.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        Self { active }
    }
}

#[cfg(unix)]
impl Drop for ActiveConnectionGuard {
    fn drop(&mut self) {
        self.active
            .fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
    }
}

#[cfg(unix)]
fn spawn_connection_task<F>(
    active: Arc<std::sync::atomic::AtomicUsize>,
    future: F,
) -> tokio::task::JoinHandle<()>
where
    F: std::future::Future<Output = ()> + Send + 'static,
{
    let guard = ActiveConnectionGuard::claim(active);
    tokio::spawn(async move {
        let _guard = guard;
        future.await;
    })
}

/// Run the daemon: bind the socket, warm in the background, serve request
/// frames until SIGTERM/SIGINT.
///
/// Fatally acquires its own startup lock, which only protects
/// cleanup→pid-claim→bind — `dispatcher` has already run migrations and
/// applied pack schema plans while constructing itself, unguarded. Production
/// boot must go through [`run_daemon_with_boot_guard`] instead, which extends
/// the same lock back over construction. This entry point is for callers
/// (and tests) that build the dispatcher and start serving as one atomic
/// step with no separate boot-guard window to protect.
#[cfg(unix)]
pub async fn run_daemon<D: DaemonDispatch>(dispatcher: D) -> anyhow::Result<()> {
    let boot_guard = Some(acquire_daemon_boot_guard()?);
    run_daemon_with_boot_guard_inner(
        dispatcher,
        boot_guard,
        false,
        DaemonOptions::default(),
        |_| DaemonStartupReport::default(),
    )
    .await
}

/// Run a real daemon server for an in-process multi-launch test.
///
/// Separate production daemon candidates have distinct PIDs, so the boot fence
/// recognizes a live incumbent and makes later candidates exit. Parallel
/// test launchers share one OS process and therefore one PID; this explicit
/// fault-injection entry point preserves the production fence semantics by
/// allowing a live same-PID incumbent to win. Ordinary daemon startup
/// continues to treat a same-PID rendezvous as stale, protecting PID-reuse
/// cleanup behavior.
///
/// A losing candidate still follows the ordinary daemon-exit path and cancels
/// the process-wide component shutdown token. Callers must therefore use a
/// component-free dispatcher; this seam validates socket/PID ownership, not
/// multi-candidate component lifecycle.
#[cfg(all(unix, any(test, feature = "fault-injection")))]
#[doc(hidden)]
pub async fn run_daemon_in_process_test<D: DaemonDispatch>(dispatcher: D) -> anyhow::Result<()> {
    let boot_guard = Some(acquire_daemon_boot_guard()?);
    run_daemon_with_boot_guard_inner(
        dispatcher,
        boot_guard,
        true,
        DaemonOptions::default(),
        |_| DaemonStartupReport::default(),
    )
    .await
}

#[cfg(unix)]
#[derive(Clone, Copy, PartialEq, Eq)]
enum RendezvousPathRole {
    Socket,
    PidFile,
}

#[cfg(unix)]
impl RendezvousPathRole {
    fn env_name(self) -> &'static str {
        match self {
            Self::Socket => SOCKET_PATH_ENV,
            Self::PidFile => PID_PATH_ENV,
        }
    }

    fn directory_name(self) -> &'static str {
        match self {
            Self::Socket => "socket directory",
            Self::PidFile => "PID-file directory",
        }
    }

    fn path_name(self) -> &'static str {
        match self {
            Self::Socket => "socket path",
            Self::PidFile => "PID-file path",
        }
    }

    fn path_component_name(self) -> &'static str {
        match self {
            Self::Socket => "socket-path",
            Self::PidFile => "PID-file-path",
        }
    }
}

/// Vet the socket's parent directory, re-permissioning it only when it is the
/// directory khive owns by convention.
///
/// `KHIVE_SOCKET` takes an arbitrary path, and the previous unconditional
/// chmod-0700 of its parent was wrong in both directions: pointed at a shared
/// parent like `/tmp` it either failed outright for an ordinary user, or —
/// worse — succeeded when privileged and stripped access for every other
/// process on the machine. A directory we did not create is never modified.
///
/// What the directory must actually prevent is a *takeover of the socket
/// path*: the connection gate is the 0600 socket plus the accept-time
/// peer-uid check, but both defend the daemon's own socket — neither helps
/// once another local user can put *their* listener at the path clients
/// resolve. There are two ways a shared directory allows that, and the
/// sticky bit closes neither: a writer can *pre-bind* the predictable path
/// before this daemon starts (the sticky bit restricts unlinking, not
/// creating), and the directory's *owner* can unlink and rebind even in a
/// 1777 directory (sticky exempts the directory owner). So a caller-chosen
/// directory is served only when it is trusted end to end: owned by this
/// daemon's euid or root, and not writable by group or other at all. The
/// umask-default 0755 stays acceptable; shared sticky directories like
/// `/tmp` do not.
#[cfg(unix)]
pub(crate) fn ensure_socket_dir_is_trusted(parent: &std::path::Path) -> anyhow::Result<()> {
    // SAFETY: `geteuid` is always successful and takes no arguments.
    let daemon_euid = unsafe { libc::geteuid() } as u32;
    ensure_rendezvous_dir_is_trusted(parent, RendezvousPathRole::Socket, daemon_euid, true)
}

/// Vet the parent directory of a PID file before reading, locking, or writing
/// it. The file's lock only protects the inode currently named by its path;
/// every directory component must therefore be as swap-resistant as the
/// socket rendezvous.
#[cfg(unix)]
pub fn ensure_pid_file_dir_is_trusted(pid_file: &std::path::Path) -> anyhow::Result<()> {
    let parent = pid_file
        .parent()
        .filter(|parent| !parent.as_os_str().is_empty())
        .unwrap_or_else(|| std::path::Path::new("."));
    // SAFETY: `geteuid` is always successful and takes no arguments.
    let daemon_euid = unsafe { libc::geteuid() } as u32;
    ensure_rendezvous_dir_is_trusted(parent, RendezvousPathRole::PidFile, daemon_euid, false)
}

#[cfg(unix)]
fn ensure_rendezvous_dir_is_trusted(
    parent: &std::path::Path,
    role: RendezvousPathRole,
    daemon_euid: u32,
    repair_owned_default: bool,
) -> anyhow::Result<()> {
    let env_name = role.env_name();
    let directory_name = role.directory_name();

    if repair_owned_default && parent == khive_dir() {
        std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700)).map_err(|e| {
            anyhow::anyhow!(
                "refusing to start: cannot chmod 0700 {}: {e}. The khive directory must be \
                 owner-only as the {directory_name} for {env_name}; it is part of the \
                 same-uid guarantee this daemon enforces.",
                parent.display()
            )
        })?;
        return ensure_rendezvous_path_is_swap_resistant(parent, daemon_euid, role);
    }

    // Fail closed on the stat itself: not being able to read the metadata is
    // not the same as the directory passing.
    let meta = std::fs::metadata(parent).map_err(|e| {
        anyhow::anyhow!(
            "refusing to start: cannot stat {directory_name} {} for {env_name}: {e}. \
             It gates rendezvous-path safety, and unreadable metadata is not a passing state.",
            parent.display()
        )
    })?;

    use std::os::unix::fs::MetadataExt;
    let owner = meta.uid();
    if owner != daemon_euid && owner != 0 {
        anyhow::bail!(
            "refusing to start: {directory_name} {} for {env_name} is owned by uid {owner}, \
             not this daemon's uid ({daemon_euid}) or root. A directory owner can replace \
             the rendezvous path regardless of mode bits. Point {env_name} at a directory \
             you own, or unset it for the default.",
            parent.display()
        );
    }

    let mode = meta.permissions().mode();
    if mode & 0o022 != 0 {
        anyhow::bail!(
            "refusing to start: {directory_name} {} for {env_name} is mode {:04o} — writable \
             by group or other, so another local user could replace the rendezvous path. \
             Use a directory only you can write, or unset {env_name} for the default. \
             This daemon is not changing the permissions of a directory it does not own.",
            parent.display(),
            mode & 0o7777
        );
    }

    ensure_rendezvous_path_is_swap_resistant(parent, daemon_euid, role)
}

/// Socket-role form of [`ensure_rendezvous_path_is_swap_resistant`], used by the
/// socket-path tests.
#[cfg(all(unix, test))]
fn ensure_socket_path_is_swap_resistant(
    parent: &std::path::Path,
    daemon_euid: u32,
) -> anyhow::Result<()> {
    ensure_rendezvous_path_is_swap_resistant(parent, daemon_euid, RendezvousPathRole::Socket)
}

/// Walk the socket directory path exactly as the kernel will traverse it at
/// bind time — component by component, following symlinks — and refuse any
/// node another local user could swap after this validation.
///
/// Checking only the canonicalized result is not enough: bind and every
/// client traverse the *original* path, so a symlink component owned by
/// another user can resolve somewhere trusted while it is being validated
/// and be retargeted before the socket is bound. Validating the nodes the
/// traversal actually visits closes that gap: a symlink component is
/// acceptable only when its owner — the only party besides root who can
/// retarget it — is this daemon's euid or root, and every directory
/// component is held to the ancestor rule below.
///
/// The directory rule is deliberately weaker than the immediate parent's.
/// The parent hosts the socket file, where the threat is *creation* of the
/// predictable path — the sticky bit does not restrict creating, so no
/// sticky exception is sound there. An ancestor only threatens via
/// *rename/unlink of an existing entry we own*, which the sticky bit does
/// restrict: in a sticky directory, only the entry's owner, the directory's
/// owner, or root may rename it. So a root-owned `/tmp` (1777) is an
/// acceptable ancestor of a user-owned 0700 socket directory, while a
/// non-sticky group/other-writable ancestor, or one owned by a third uid
/// (who could rename the entry, or chmod the directory first), is refused.
/// Every stat failure fails closed.
#[cfg(unix)]
fn ensure_rendezvous_path_is_swap_resistant(
    parent: &std::path::Path,
    daemon_euid: u32,
    role: RendezvousPathRole,
) -> anyhow::Result<()> {
    use std::os::unix::fs::MetadataExt;

    let env_name = role.env_name();
    let directory_name = role.directory_name();
    let path_name = role.path_name();
    let component_name = role.path_component_name();

    let absolute = if parent.is_absolute() {
        parent.to_path_buf()
    } else {
        std::env::current_dir()
            .map_err(|e| {
                anyhow::anyhow!(
                    "refusing to start: cannot resolve the working directory to absolutize \
                     {directory_name} {} for {env_name}: {e}.",
                    parent.display()
                )
            })?
            .join(parent)
    };

    fn push_components(stack: &mut Vec<std::ffi::OsString>, path: &std::path::Path) {
        let components: Vec<_> = path
            .components()
            .map(|c| c.as_os_str().to_os_string())
            .collect();
        stack.extend(components.into_iter().rev());
    }

    let mut stack: Vec<std::ffi::OsString> = Vec::new();
    push_components(&mut stack, &absolute);
    let mut resolved = std::path::PathBuf::new();
    let mut symlinks_followed = 0u32;

    while let Some(component) = stack.pop() {
        if component == "/" {
            resolved = std::path::PathBuf::from("/");
            continue;
        }
        if component == "." {
            continue;
        }
        if component == ".." {
            resolved.pop();
            continue;
        }
        let candidate = resolved.join(&component);
        let meta = std::fs::symlink_metadata(&candidate).map_err(|e| {
            anyhow::anyhow!(
                "refusing to start: cannot stat {component_name} component {} for {env_name}: \
                 {e}. An unreadable component is not a passing one.",
                candidate.display()
            )
        })?;
        let owner = meta.uid();

        if meta.file_type().is_symlink() {
            symlinks_followed += 1;
            if symlinks_followed > 40 {
                anyhow::bail!(
                    "refusing to start: {path_name} for {env_name} resolves through more than \
                     40 symlinks at {} — treating this as a loop.",
                    candidate.display()
                );
            }
            if owner != daemon_euid && owner != 0 {
                anyhow::bail!(
                    "refusing to start: {component_name} symlink component {} for {env_name} \
                     is owned by uid {owner}, not this daemon's uid ({daemon_euid}) or root — \
                     its owner could retarget it after this check and re-root the {path_name}. \
                     Point {env_name} somewhere trusted end to end, or unset it for the default.",
                    candidate.display()
                );
            }
            let target = std::fs::read_link(&candidate).map_err(|e| {
                anyhow::anyhow!(
                    "refusing to start: cannot read {component_name} symlink component {} \
                     for {env_name}: {e}.",
                    candidate.display()
                )
            })?;
            push_components(&mut stack, &target);
            continue;
        }

        if meta.is_dir() {
            let mode = meta.permissions().mode();
            let sticky = mode & 0o1000 != 0;
            if owner != daemon_euid && owner != 0 {
                anyhow::bail!(
                    "refusing to start: {component_name} ancestor {} for {env_name} is owned by \
                     uid {owner}, not this daemon's uid ({daemon_euid}) or root — its owner \
                     could rename the next path component and re-root the {path_name}. Point \
                     {env_name} somewhere trusted end to end, or unset it for the default.",
                    candidate.display()
                );
            }
            if mode & 0o022 != 0 && !sticky {
                anyhow::bail!(
                    "refusing to start: {component_name} ancestor {} for {env_name} is mode \
                     {:04o} — writable by group or other without the sticky bit, so another \
                     local user could rename the next path component and re-root the {path_name}. \
                     Point {env_name} somewhere trusted end to end, or unset it for the default.",
                    candidate.display(),
                    mode & 0o7777
                );
            }
            resolved = candidate;
            continue;
        }

        anyhow::bail!(
            "refusing to start: {component_name} component {} for {env_name} is neither a \
             directory nor a symlink — the {path_name} cannot traverse it.",
            candidate.display()
        );
    }

    Ok(())
}

/// Run the daemon using a startup lock acquired by the caller *before*
/// building `dispatcher`, so a second process racing to boot (e.g. two
/// `kkernel mcp --daemon` spawns before either has bound its socket) cannot
/// run migrations/FTS DDL concurrently against the same database file.
/// `boot_guard` is only `None` on non-unix targets, where there is no
/// advisory boot lock to hold in the first place; every unix daemon-mode
/// caller passes `Some`.
///
/// The guard is held across cleanup → pid-claim → bind, then dropped. The
/// caller must not still be holding a *different* handle to the same lock
/// file when this function is entered — see the "Deadlock note" on the
/// `_startup_lock` binding below for why that would self-deadlock on `flock`.
#[cfg(unix)]
pub async fn run_daemon_with_boot_guard<D: DaemonDispatch>(
    dispatcher: D,
    boot_guard: Option<std::fs::File>,
) -> anyhow::Result<()> {
    run_daemon_with_boot_guard_inner(
        dispatcher,
        boot_guard,
        false,
        DaemonOptions::default(),
        |_| DaemonStartupReport::default(),
    )
    .await
}

/// Run the daemon and start host-owned background work only after the socket
/// is bound, permissions are restricted, and this process owns the PID file.
/// The callback runs once while the startup lock and teardown guard are held;
/// setup failures never invoke it. Work started by the callback must use the
/// daemon shutdown token and tracked-task drain contract.
#[cfg(unix)]
pub async fn run_daemon_with_boot_guard_and_start<D, F>(
    dispatcher: D,
    boot_guard: Option<std::fs::File>,
    start: F,
) -> anyhow::Result<()>
where
    D: DaemonDispatch,
    F: FnOnce(&D) + Send,
{
    run_daemon_with_options_and_boot_guard_and_start(
        dispatcher,
        boot_guard,
        DaemonOptions::default(),
        |dispatcher| {
            start(dispatcher);
            DaemonStartupReport::default()
        },
    )
    .await
}

/// Start with an explicit launch mode and collect the host's startup inventory.
#[cfg(unix)]
pub async fn run_daemon_with_options_and_boot_guard_and_start<D, F>(
    dispatcher: D,
    boot_guard: Option<std::fs::File>,
    options: DaemonOptions,
    start: F,
) -> anyhow::Result<()>
where
    D: DaemonDispatch,
    F: FnOnce(&D) -> DaemonStartupReport + Send,
{
    anyhow::ensure!(
        !options.idle_interval.is_zero(),
        "daemon idle interval must be positive"
    );
    run_daemon_with_boot_guard_inner(dispatcher, boot_guard, false, options, start).await
}

#[cfg(unix)]
async fn run_daemon_with_boot_guard_inner<D, F>(
    dispatcher: D,
    boot_guard: Option<std::fs::File>,
    allow_same_process_incumbent: bool,
    options: DaemonOptions,
    start: F,
) -> anyhow::Result<()>
where
    D: DaemonDispatch,
    F: FnOnce(&D) -> DaemonStartupReport + Send,
{
    // Cancel on every exit, including setup failure and unwinding from the
    // post-ownership startup callback. The guard precedes all fallible work
    // so even a process that never becomes the daemon relinquishes its
    // process-lifetime shutdown token; restarting requires exec.
    struct ComponentTeardown;
    impl Drop for ComponentTeardown {
        fn drop(&mut self) {
            daemon_shutdown_token().cancel();
        }
    }
    let _component_teardown = ComponentTeardown;

    // Placed after the teardown guard so this refusal keeps the ADR-119
    // contract every other pre-bind error path has, and before the paths are
    // resolved so a split rendezvous never reaches cleanup/bind/pid-write.
    ensure_rendezvous_overrides_paired()?;

    let sock = socket_path();
    let pid_file = pid_path();
    let socket_parent = sock.parent();
    let pid_parent = pid_file.parent();

    if let Some(parent) = socket_parent {
        std::fs::create_dir_all(parent)?;
        ensure_socket_dir_is_trusted(parent)?;
    }
    // Identical parent paths traverse the same components, so the socket
    // check above also vets the PID-file parent. Aliased paths are checked
    // independently because each original path is traversed by file access.
    if pid_parent != socket_parent {
        ensure_pid_file_dir_is_trusted(&pid_file)?;
    }

    // Hold the startup lock across cleanup → pid-claim → bind so a concurrent
    // client's kill_and_respawn (which also holds this lock) cannot remove the
    // rendezvous paths during setup. The PID file's own lock is retained after
    // this shared startup lock is released, including while the listener drains
    // during shutdown.
    //
    // Deadlock note: the client holds this lock only during kill+spawn and
    // releases it before the spawned daemon process starts (the lock guard is
    // dropped when kill_and_respawn returns, before the readiness probe loop).
    // The daemon holds exactly one handle to this lock for its whole boot
    // sequence (received as `boot_guard`, extended from before `dispatcher`
    // was constructed) — never a second, independently-acquired handle in the
    // same process, which would self-deadlock on `flock`.
    let _startup_lock = boot_guard;

    // A second daemon must refuse loudly rather than exit successfully while
    // another daemon owns this rendezvous. Only `Stale` lets the caller proceed.
    match cleanup_stale_daemon(
        &sock,
        &pid_file,
        allow_same_process_incumbent,
        dispatcher.config_id(),
    )
    .await
    {
        Incumbent::Serving(incumbent_pid) => {
            tracing::error!(
                pid = incumbent_pid,
                socket = ?sock,
                "refusing to start: a khived instance is already serving this socket"
            );
            anyhow::bail!(
                "refusing to start: khived is already running as pid {incumbent_pid}, \
                 serving socket {}. Stop that instance first if you intend to replace it.",
                sock.display()
            );
        }
        Incumbent::Live(incumbent_pid) => {
            tracing::error!(
                pid = incumbent_pid,
                socket = ?sock,
                "refusing to start: a live process owns the PID file but no khived answered"
            );
            anyhow::bail!(
                "refusing to start: pid {incumbent_pid} owns the daemon PID file and is alive, \
                 but nothing answered the khived protocol on {}. It may be draining. Nothing \
                 was removed; stop that process first if you intend to replace it.",
                sock.display()
            );
        }
        Incumbent::Stale => {}
    }

    // Register the sender recorder before the shutdown streams below. The
    // signal registry runs a signal's actions in registration order, so the
    // recorder stores the sender before tokio's action wakes the shutdown
    // path; registered after it, the exit record can say sender_pid=unknown.
    if let Err(error) = exit_record::capture_signal_senders() {
        tracing::warn!(target: "khive.boot", error = %error, "signal sender capture unavailable; the exit record will say sender_pid=unknown");
    }
    // Install signal streams before publishing either rendezvous file. A
    // supervisor may stop us as soon as connect/pid checks succeed, before
    // the accept loop or shutdown future has been polled. Keep these streams
    // alive so a signal during the rest of startup reaches normal cleanup.
    let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
    let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;

    let pid_file_guard = match write_pid_file_exclusive(&pid_file) {
        Ok(guard) => guard,
        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
            // A PID file appeared between cleanup and our claim. Never touch
            // the winner's files; defer only if it already answers as khived.
            if pid_file_names_a_reachable_daemon(
                &pid_file,
                &sock,
                allow_same_process_incumbent,
                dispatcher.config_id(),
            )
            .await
            {
                tracing::info!(
                    "a replacement khived already claimed the pid/socket rendezvous; exiting"
                );
                return Ok(());
            }
            anyhow::bail!(
                "failed to claim daemon pid file at {pid_file:?}: it already exists \
                 and does not name a reachable daemon"
            );
        }
        Err(e) => return Err(e.into()),
    };
    // After the claim, never before: only the PID file's owner writes the record.
    let exit_record = exit_record::ExitRecord::claim(&pid_file);

    let listener = match UnixListener::bind(&sock) {
        Ok(listener) => listener,
        Err(e) => {
            exit_record.exiting(exit_record::ExitReason::Error);
            remove_pid_file_if_owned(&pid_file, &pid_file_guard);
            return Err(e.into());
        }
    };
    // Fail closed, same reason as the directory above. If this chmod fails the
    // socket is world-reachable in a way the accepted design never covered, so
    // the bound listener is dropped and the entry removed rather than served.
    if let Err(e) = std::fs::set_permissions(&sock, std::fs::Permissions::from_mode(0o600)) {
        drop(listener);
        let _ = std::fs::remove_file(&sock);
        exit_record.exiting(exit_record::ExitReason::Error);
        remove_pid_file_if_owned(&pid_file, &pid_file_guard);
        return Err(anyhow::anyhow!(
            "refusing to start: cannot chmod 0600 {}: {e}. The daemon socket must be owner-only \
             — it is half of the single-principal guarantee this daemon enforces.",
            sock.display()
        ));
    }
    // Captured while still holding the startup lock, immediately after
    // bind, so shutdown cleanup can later prove "this is still the same socket
    // I bound" rather than trusting the path alone.
    let bound_identity = socket_identity(&sock);

    let lifecycle = Arc::new(DaemonLifecycle::new(options, start(&dispatcher)));

    // Release the shared startup lock now that the listener is bound. The
    // locked PID file continues to identify this daemon through shutdown.
    drop(_startup_lock);
    tracing::info!(
        socket = ?sock,
        pid = std::process::id(),
        source_revision = crate::BUILD_INFO.source_revision,
        build_time = crate::BUILD_INFO.build_time,
        "khived listening"
    );

    {
        let warm = dispatcher.clone();
        track_named_background_task("daemon_warmup", async move {
            warm.warm_all().await;
        });
    }

    // The checkpoint task's own strong-count-based exit is unreachable
    // whenever `event_store_for_checkpoint()` returns `Some` (the ordinary
    // production shape), because the `SqlEventStore` it wraps retains its
    // own clone of the same pool. An explicit watch channel replaces that
    // mechanism: the sender is held for the remainder of this function's
    // scope and signalled as the first action once shutdown is observed,
    // below.
    let (checkpoint_shutdown_tx, checkpoint_shutdown_rx) = tokio::sync::watch::channel(());
    // ADR-091 Amendment 3: one checkpoint task per file-backed backend the
    // dispatcher wired — the primary pool plus every entry
    // `secondary_pools_for_checkpoint` returns — sharing this one shutdown
    // channel (the sender broadcasts to every receiver clone), so the single
    // send below stops every spawned task before `drain()`.
    let checkpoint_tasks = checkpoint_task_specs(
        dispatcher.pool_for_checkpoint(),
        dispatcher.secondary_pools_for_checkpoint(),
        dispatcher.event_store_for_checkpoint(),
        dispatcher.namespace().to_string(),
    );
    if !checkpoint_tasks.is_empty() {
        let cfg = CheckpointConfig::from_env();
        let checkpoint_task_count = checkpoint_tasks.len();
        for task in checkpoint_tasks {
            track_named_background_task(
                "wal_checkpoint",
                run_checkpoint_task(
                    task.pool,
                    cfg.clone(),
                    task.lifecycle_owner,
                    checkpoint_shutdown_rx.clone(),
                    task.is_main,
                ),
            );
        }
        tracing::info!(checkpoint_task_count, "WAL checkpoint task(s) started");
    }

    let active = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let connection_tasks = Arc::new(std::sync::Mutex::new(
        Vec::<tokio::task::JoinHandle<()>>::new(),
    ));
    let (request_shutdown_tx, request_shutdown_rx) = tokio::sync::watch::channel(false);

    let shutdown = async {
        let signal = tokio::select! {
            _ = sigterm.recv() => "SIGTERM",
            _ = sigint.recv() => "SIGINT",
        };
        tracing::info!("received {signal}");
        // Tokio retains its process-wide handlers after the streams are dropped.
        // This daemon cannot restart without exec; a repeat signal must terminate
        // even if shutdown is blocked in synchronous recovery-lock acquisition.
        for signal in [libc::SIGTERM, libc::SIGINT] {
            // SAFETY: setting SIG_DFL for these valid signals needs no handler
            // pointer or shared Rust state and applies to the whole process.
            if unsafe { libc::signal(signal, libc::SIG_DFL) } == libc::SIG_ERR {
                return Err(std::io::Error::last_os_error());
            }
        }
        Ok::<&'static str, std::io::Error>(signal)
    };
    tokio::pin!(shutdown);
    lifecycle.ready();

    // SAFETY: `geteuid` is always successful and takes no arguments.
    let daemon_euid = unsafe { libc::geteuid() } as u32;

    let reason = tokio::select! {
        _ = async {
            let mut accept_error_backoff = None;
            let mut last_accept_error_log: Option<std::time::Instant> = None;
            loop {
                match listener.accept().await {
                    Ok((mut stream, _)) => {
                        let initial_frame_deadline =
                            tokio::time::Instant::now() + INITIAL_FRAME_READ_TIMEOUT;
                        accept_error_backoff = None;
                        last_accept_error_log = None;
                        // Refuse a foreign uid before any frame is read.
                        // Fails CLOSED: an error reading peer credentials is
                        // "cannot prove same-uid", which is the same answer as
                        // "is not same-uid" — never a pass.
                        //
                        // Scope: this is a same-UID check, which is strictly
                        // weaker than same-PRINCIPAL. Several distinct actors
                        // running under one uid all pass here, so this does not
                        // by itself establish that the daemon serves a single
                        // principal, and nothing downstream refuses or degrades
                        // on observing more than one. Do not describe it as a
                        // multi-principal guard — it bounds the process boundary,
                        // not the identity one.
                        match peer_uid(&stream) {
                            Ok(peer) if uid_is_permitted(peer, daemon_euid) => {}
                            Ok(peer) => {
                                tracing::error!(
                                    peer_uid = peer,
                                    daemon_euid,
                                    "refusing connection from a foreign uid: this daemon accepts \
                                     only peers running as its own uid"
                                );
                                drop(stream);
                                continue;
                            }
                            Err(e) => {
                                tracing::error!(
                                    error = %e,
                                    "refusing connection: cannot read peer credentials, so \
                                     same-uid cannot be proven"
                                );
                                drop(stream);
                                continue;
                            }
                        }
                        // One permit per connection, taken before the task is
                        // spawned. Past the cap the peer is answered with a busy
                        // error and the stream is dropped; connections already
                        // admitted keep their permits and are not affected.
                        let Some(permit) = admit_or_refuse_busy(
                            &lifecycle.connections,
                            &mut stream,
                            dispatcher.config_id(),
                        )
                        .await
                        else {
                            continue;
                        };
                        // Keep the acceptance-time deadline across the
                        // credential check and connection-task scheduling.
                        let d = dispatcher.clone();
                        let shutdown = request_shutdown_rx.clone();
                        let lifecycle = Arc::clone(&lifecycle);
                        let handle = spawn_connection_task(Arc::clone(&active), async move {
                            // Released when the handler ends, whether it returns,
                            // panics or is aborted at shutdown.
                            let _permit = permit;
                            handle_conn_with_lifecycle(
                                stream,
                                d,
                                Some(shutdown),
                                initial_frame_deadline,
                                Some(lifecycle),
                            )
                            .await;
                        });
                        let mut tasks = connection_tasks
                            .lock()
                            .unwrap_or_else(std::sync::PoisonError::into_inner);
                        tasks.retain(|task| !task.is_finished());
                        tasks.push(handle);
                    }
                    Err(e) => {
                        let delay = next_accept_error_backoff(accept_error_backoff);
                        accept_error_backoff = Some(delay);
                        let capacity_exhausted = matches!(
                            e.raw_os_error(),
                            Some(libc::EMFILE) | Some(libc::ENFILE)
                        );
                        if last_accept_error_log.is_none_or(|last| {
                            last.elapsed() >= std::time::Duration::from_secs(30)
                        }) {
                            tracing::error!(
                                error = %e,
                                capacity_exhausted,
                                retry_ms = delay.as_millis(),
                                "daemon accept failed; retrying with bounded backoff"
                            );
                            last_accept_error_log = Some(std::time::Instant::now());
                        }
                        tokio::time::sleep(delay).await;
                    }
                }
            }
        } => DaemonShutdownReason::Signal,
        result = &mut shutdown => {
            exit_record.exiting(exit_record::ExitReason::Signal(result?));
            DaemonShutdownReason::Signal
        },
        _ = wait_for_idle(&dispatcher, &lifecycle) => {
            exit_record.exiting(exit_record::ExitReason::Idle);
            DaemonShutdownReason::Idle
        },
    };

    lifecycle.draining(reason);

    // A listening backlog is not admitted work. Close it before draining so
    // new clients cannot finish writing to a socket nobody will accept.
    drop(listener);

    // Signal the checkpoint task to exit before draining, so `drain()`
    // actually waits on it via `track_background_task` rather than the
    // task outliving the drain window (or the process) unsignalled.
    let _ = checkpoint_shutdown_tx.send(());

    // Per-run signal: read scopes stop promptly, admitted writes ignore it and
    // retain the rest of the configured drain window to commit or roll back.
    if reason == DaemonShutdownReason::Signal {
        let _ = request_shutdown_tx.send(true);
    }

    // Same ordering contract for ADR-119 daemon components: cancel before
    // drain, so each component's supervisor (itself a tracked task) can run
    // its bounded shutdown inside the drain wait.
    daemon_shutdown_token().cancel();

    let drained = if reason == DaemonShutdownReason::Idle {
        tokio::select! {
            _ = drain_for_idle(&active, drain_timeout()) => true,
            result = &mut shutdown => {
                exit_record.exiting(exit_record::ExitReason::Signal(result?));
                lifecycle.draining(DaemonShutdownReason::Signal);
                let _ = request_shutdown_tx.send(true);
                drain(&active).await
            }
        }
    } else {
        drain(&active).await
    };
    let tasks = {
        let mut retained = connection_tasks
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        std::mem::take(&mut *retained)
    };
    finish_connection_tasks(tasks, drained).await;

    // A concurrent client's `kill_and_respawn` may have already decided
    // this daemon looked stale, killed it, and spawned a replacement that
    // bound the same socket/PID paths while this daemon was draining above.
    // Reacquire the recovery lock (the same one that serializes startup) and
    // only unlink if the PID file still names this process AND the socket at
    // `sock` is still the exact one this daemon bound — otherwise a
    // replacement daemon owns those paths now and unlinking would delete its
    // live socket/PID out from under it.
    match acquire_recovery_lock() {
        Some(_shutdown_lock) => {
            shutdown_cleanup_if_owned(&sock, &pid_file, bound_identity);
        }
        None => {
            tracing::warn!(
                "could not acquire recovery lock for shutdown cleanup; \
                 skipping unlink to avoid deleting a replacement daemon's paths"
            );
        }
    }
    lifecycle.stopped();
    tracing::info!("khived stopped");
    Ok(())
}

/// Remove `sock`/`pid_file` only if they still belong to this process: the PID
/// file must name `std::process::id()` AND the socket currently at `sock` must
/// still be the exact one identified by `bound_identity` (dev/ino, not path).
///
/// Returns `true` if cleanup ran, `false` if it was skipped because a
/// replacement daemon already owns those paths. The caller must hold
/// the recovery lock across this call — the same lock daemon startup holds
/// across cleanup+bind+pid-write — so no replacement can bind between this
/// function's checks and its unlinks.
#[cfg(unix)]
fn shutdown_cleanup_if_owned(
    sock: &std::path::Path,
    pid_file: &std::path::Path,
    bound_identity: Option<SocketIdentity>,
) -> bool {
    let pid_is_ours = std::fs::read_to_string(pid_file)
        .ok()
        .and_then(|s| s.trim().parse::<u32>().ok())
        == Some(std::process::id());
    let socket_is_ours = bound_identity.is_some() && socket_identity(sock) == bound_identity;
    if pid_is_ours && socket_is_ours {
        let _ = std::fs::remove_file(sock);
        let _ = std::fs::remove_file(pid_file);
        true
    } else {
        tracing::warn!(
            socket = ?sock,
            pid_file = ?pid_file,
            "skipping shutdown cleanup — a replacement daemon already owns this socket/PID"
        );
        false
    }
}

// ── helpers ───────────────────────────────────────────────────────────────────

/// Liveness verdict for a `kill(pid, 0)` probe.
#[cfg(unix)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PidLiveness {
    /// errno 0 — signal delivery succeeded, the process exists and this
    /// caller may signal it.
    Alive,
    /// ESRCH (or any other non-EPERM errno) — no such process.
    Dead,
    /// EPERM — the process exists but this caller lacks permission to
    /// signal it. Unknown-safe: treated as running so stale-daemon cleanup
    /// never unlinks a live daemon's socket/PID file just because it is
    /// owned by a different user/uid.
    PermissionDenied,
}

#[cfg(unix)]
impl PidLiveness {
    fn is_running(self) -> bool {
        !matches!(self, PidLiveness::Dead)
    }
}

/// Maps a `kill(pid, 0)` outcome (return code + errno) to a [`PidLiveness`].
/// Pure and side-effect-free so the errno mapping can be unit tested without
/// a real process probe.
#[cfg(unix)]
fn classify_kill_result(rc: i32, errno: i32) -> PidLiveness {
    if rc == 0 {
        return PidLiveness::Alive;
    }
    match errno {
        libc::EPERM => PidLiveness::PermissionDenied,
        _ => PidLiveness::Dead,
    }
}

#[cfg(unix)]
fn is_process_running(pid: u32) -> bool {
    let Ok(pid) = i32::try_from(pid) else {
        return false;
    };
    if pid <= 0 {
        return false;
    }
    // SAFETY: signal 0 is an existence/permission probe with no side effects.
    let rc = unsafe { libc::kill(pid, 0) };
    let errno = std::io::Error::last_os_error().raw_os_error().unwrap_or(0);
    classify_kill_result(rc, errno).is_running()
}

/// Whether a PID may identify an incumbent from this candidate's point of view.
///
/// Production rejects the current PID so a stale rendezvous left by a prior
/// process whose PID was reused cannot protect an unrelated socket. The
/// in-process daemon harness opts in to the same-PID case because all of its
/// otherwise independent boot candidates necessarily share one OS process.
#[cfg(unix)]
fn pid_can_name_incumbent(pid: u32, current_pid: u32, allow_same_process_incumbent: bool) -> bool {
    allow_same_process_incumbent || pid != current_pid
}

/// Bounded timeout for the protocol-identity probe used by duplicate-daemon
/// detection. Short enough that a hung or foreign listener does not stall
/// startup; long enough for a live khived under normal load to answer a
/// `probe_only` frame.
#[cfg(unix)]
const DUPLICATE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(500);

/// Whether the listener at `sock` actually speaks the khived wire protocol
/// **as a khived this process can defer to** — identified by a configuration
/// compatible with `expected_config_id`.
///
/// A live PID plus an accepting Unix socket is not proof of khived: any
/// unrelated process that happens to have bound the same path also answers
/// `connect()`. Nor is any well-formed [`DaemonResponseFrame`] proof: a
/// `config_mismatch`/`version_mismatch` response, a `metrics_only` snapshot
/// response, or a legacy pre-probe daemon that falls through to normal
/// dispatch on the empty `ops` string, all deserialize cleanly without being
/// the unambiguous "yes, alive and identity-matching" answer this check
/// needs — the daemon's `metrics_only` arm in particular echoes the same
/// `ok=true, result=None, error=None`, all-mismatch-flags-false, matching
/// protocol version and `served_config_id` shape as the probe-ack arm, and
/// is distinguished only by carrying `metrics: Some(...)`. This sends a
/// bounded `probe_only` frame (the same identity probe the client-side
/// recovery path uses, `crates/khive-mcp/src/daemon.rs::probe_daemon_identity`)
/// carrying this process's own `config_id`, and requires a compatible
/// probe-branch shape back: `ok=true`, `result=None`, `error=None`,
/// `metrics=None`, `request_id=None` (this probe frame never sets one), no
/// mismatch flags, matching protocol version, and matching
/// `served_config_id` — mirroring the client probe's `is_probe_ack` check so
/// both sides of the protocol agree on what "alive" means. Connect, write,
/// and read are all inside the one bounded timeout: `UnixStream::connect`
/// itself awaits write readiness, so a listener with a saturated accept
/// backlog could otherwise hold this call open past the advertised bound.
/// A connect that succeeds but never answers, times out, or answers with
/// non-protocol bytes, a mismatched identity, a `metrics_only` snapshot, or
/// any other non-probe-shaped response is not treated as the same khived
/// and falls through to the stale-socket recovery path instead.
#[cfg(unix)]
async fn socket_speaks_khived_protocol(sock: &std::path::Path, expected_config_id: &str) -> bool {
    let probe = DaemonRequestFrame {
        probe_only: true,
        protocol_version: PROTOCOL_VERSION,
        config_id: expected_config_id.to_string(),
        ..Default::default()
    };
    let Ok(payload) = serde_json::to_vec(&probe) else {
        return false;
    };
    let response = tokio::time::timeout(DUPLICATE_PROBE_TIMEOUT, async {
        let mut stream = UnixStream::connect(sock).await.ok()?;
        write_frame(&mut stream, &payload).await.ok()?;
        let raw = read_frame(&mut stream).await.ok()?;
        serde_json::from_slice::<DaemonResponseFrame>(&raw).ok()
    })
    .await
    .ok()
    .flatten();

    let Some(resp) = response else {
        return false;
    };
    let is_probe_ack = resp.ok
        && resp.result.is_none()
        && resp.error.is_none()
        && resp.metrics.is_none()
        && resp.request_id.is_none();
    is_probe_ack
        && !resp.version_mismatch
        && !resp.namespace_mismatch
        && !resp.config_mismatch
        && resp.daemon_protocol_version == PROTOCOL_VERSION
        && resp
            .served_config_id
            .as_deref()
            .is_some_and(|served| config_ids_compatible(expected_config_id, served))
}

/// Whether connecting to an existing socket path is definitely unreachable.
/// Timeouts and other errors remain ambiguous so cleanup fails closed.
#[cfg(unix)]
async fn socket_is_unreachable(sock: &std::path::Path) -> bool {
    match tokio::time::timeout(DUPLICATE_PROBE_TIMEOUT, UnixStream::connect(sock)).await {
        Ok(Err(error)) => matches!(
            error.kind(),
            std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused
        ),
        _ => false,
    }
}

/// What owns the daemon PID file, from the point of view of a process that wants
/// to start. A live owner is never cleaned up: a draining incumbent closes its
/// listener before it releases writers, so an unanswered socket is ambiguous and
/// deleting its PID file is how two daemons end up on one store.
#[cfg(unix)]
enum Incumbent {
    /// A live process that answered the khived protocol on the socket.
    Serving(u32),
    /// A live PID still has an active or ambiguous rendezvous. Nothing removed.
    Live(u32),
    /// Nothing live owns the store; the socket and PID file were removed.
    Stale,
}

/// Check whether `pid_file`/`sock` already name a live daemon and, if not,
/// remove the stale rendezvous files so the caller may bind fresh.
///
/// A live protocol responder, reachable socket, or live holder of the PID-file
/// lock means the caller must refuse to start. An unlocked PID with no listener
/// is a reused PID and may be reclaimed.
#[cfg(unix)]
async fn cleanup_stale_daemon(
    sock: &std::path::Path,
    pid_file: &std::path::Path,
    allow_same_process_incumbent: bool,
    expected_config_id: &str,
) -> Incumbent {
    let mut stale_pid_file_guard = None;
    if let Ok(pid_str) = std::fs::read_to_string(pid_file) {
        if let Ok(pid) = pid_str.trim().parse::<u32>() {
            if pid_can_name_incumbent(pid, std::process::id(), allow_same_process_incumbent)
                && is_process_running(pid)
            {
                if sock.exists() && socket_speaks_khived_protocol(sock, expected_config_id).await {
                    return Incumbent::Serving(pid);
                }
                if sock.exists() && !socket_is_unreachable(sock).await {
                    return Incumbent::Live(pid);
                }
                match try_acquire_pid_file_lock(pid_file) {
                    Ok(Some(guard)) => stale_pid_file_guard = Some(guard),
                    Ok(None) => return Incumbent::Live(pid),
                    Err(e) => {
                        tracing::warn!(
                            error = %e,
                            path = ?pid_file,
                            "cannot check daemon PID-file lock"
                        );
                        return Incumbent::Live(pid);
                    }
                }
            }
        }
    }
    if sock.exists() {
        if let Err(e) = std::fs::remove_file(sock) {
            tracing::warn!(error = %e, path = ?sock, "failed to remove stale socket");
        }
    }
    if pid_file.exists() {
        if let Err(e) = std::fs::remove_file(pid_file) {
            tracing::warn!(error = %e, path = ?pid_file, "failed to remove stale PID file");
        }
    }
    drop(stale_pid_file_guard);
    Incumbent::Stale
}

/// Create and lock `pid_file` exclusively (`O_EXCL`) and write this process's PID.
///
/// Uses `create_new(true)` rather than `create(true).truncate(true)` so
/// this can never silently overwrite a PID file another process created —
/// the held file lock also identifies a starting or draining daemon when its
/// socket is not yet reachable.
#[cfg(unix)]
fn write_pid_file_exclusive(pid_file: &std::path::Path) -> std::io::Result<std::fs::File> {
    use std::os::unix::fs::OpenOptionsExt;

    let mut opts = std::fs::OpenOptions::new();
    opts.write(true).create_new(true).mode(0o600);
    let mut f = opts.open(pid_file)?;
    // SAFETY: flock is a POSIX advisory lock with no memory side effects.
    let rc = unsafe { libc::flock(f.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
    if rc != 0 {
        return Err(std::io::Error::last_os_error());
    }
    f.write_all(std::process::id().to_string().as_bytes())?;
    Ok(f)
}

/// Try to lock an existing PID file without creating it. `Some(file)` means
/// there is no daemon lock holder; `None` means a daemon still owns the file.
#[cfg(unix)]
fn try_acquire_pid_file_lock(pid_file: &std::path::Path) -> std::io::Result<Option<std::fs::File>> {
    let file = std::fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open(pid_file)?;
    // SAFETY: flock is a POSIX advisory lock with no memory side effects.
    let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
    if rc == 0 {
        return Ok(Some(file));
    }
    let error = std::io::Error::last_os_error();
    if error.kind() == std::io::ErrorKind::WouldBlock
        || error.raw_os_error() == Some(libc::EWOULDBLOCK)
    {
        Ok(None)
    } else {
        Err(error)
    }
}

/// Remove the PID file on setup failure only while its path still names the
/// file this start attempt created.
#[cfg(unix)]
fn remove_pid_file_if_owned(pid_file: &std::path::Path, guard: &std::fs::File) {
    let Ok(owned) = guard.metadata() else {
        return;
    };
    let Ok(current) = std::fs::metadata(pid_file) else {
        return;
    };
    if owned.dev() == current.dev() && owned.ino() == current.ino() {
        if let Err(e) = std::fs::remove_file(pid_file) {
            tracing::warn!(error = %e, path = ?pid_file, "failed to remove unbound PID file");
        }
    }
}

/// Return `true` if `pid_file` currently names an eligible live process that
/// still answers on `sock` — i.e. a daemon already owns this rendezvous and it
/// is safe to defer to it rather than treat the `AlreadyExists` PID-file
/// collision as a boot failure. Eligibility requires a different PID in
/// production; the explicit in-process harness may allow the current PID.
#[cfg(unix)]
async fn pid_file_names_a_reachable_daemon(
    pid_file: &std::path::Path,
    sock: &std::path::Path,
    allow_same_process_incumbent: bool,
    expected_config_id: &str,
) -> bool {
    let Ok(pid_str) = std::fs::read_to_string(pid_file) else {
        return false;
    };
    let Ok(pid) = pid_str.trim().parse::<u32>() else {
        return false;
    };
    pid_can_name_incumbent(pid, std::process::id(), allow_same_process_incumbent)
        && is_process_running(pid)
        && sock.exists()
        && socket_speaks_khived_protocol(sock, expected_config_id).await
}

#[cfg(unix)]
async fn drain(active: &std::sync::atomic::AtomicUsize) -> bool {
    drain_with_timeout(active, drain_timeout()).await
}

/// Voluntary retirement keeps admitted workers and rendezvous ownership alive.
#[cfg(unix)]
async fn drain_for_idle(active: &std::sync::atomic::AtomicUsize, timeout: std::time::Duration) {
    let deadline = tokio::time::Instant::now() + timeout;
    let mut warned = false;
    while active.load(std::sync::atomic::Ordering::SeqCst) + background_task_count() != 0 {
        if !warned && tokio::time::Instant::now() >= deadline {
            tracing::warn!(
                "idle drain interval elapsed; retaining workers and rendezvous until settled"
            );
            warned = true;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
}

#[cfg(unix)]
async fn drain_with_timeout(
    active: &std::sync::atomic::AtomicUsize,
    timeout: std::time::Duration,
) -> bool {
    use std::sync::atomic::Ordering;
    // One sequentially-consistent order spans connection handoff and tracked
    // task publication. Drain must not observe an ended connection and a
    // stale pre-publication background count as two simultaneous zeroes.
    let remaining = || active.load(Ordering::SeqCst) + background_task_count();
    if remaining() == 0 {
        return true;
    }
    let deadline = tokio::time::Instant::now() + timeout;
    while remaining() > 0 {
        if tokio::time::Instant::now() >= deadline {
            tracing::warn!(
                remaining_connections = active.load(Ordering::SeqCst),
                remaining_background_tasks = background_task_count(),
                outstanding_background_tasks = %background_task_names().join(", "),
                "drain timeout reached; forcing shutdown"
            );
            return false;
        }
        tokio::select! {
            _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {}
            _ = tokio::time::sleep_until(deadline) => {}
        }
    }
    true
}

#[cfg(unix)]
async fn finish_connection_tasks(tasks: Vec<tokio::task::JoinHandle<()>>, drained: bool) {
    if !drained {
        for task in &tasks {
            if !task.is_finished() {
                task.abort();
            }
        }
    }
    for task in tasks {
        let _ = task.await;
    }
}

/// The bound `drain()` waits for tracked background tasks at daemon shutdown
/// (`KHIVE_DRAIN_TIMEOUT_SECS`, default 10s). Public so component supervision
/// can clamp per-component shutdown timeouts against it — a component timeout
/// longer than the drain bound could never complete its abort/state
/// transition before the daemon returns.
pub fn drain_timeout() -> std::time::Duration {
    let secs = khive_db::env::env_parse_or("KHIVE_DRAIN_TIMEOUT_SECS", DEFAULT_DRAIN_TIMEOUT_SECS);
    std::time::Duration::from_secs(secs)
}

/// Returns `true` for non-empty env values that are not `"0"` or `"false"`.
#[cfg(unix)]
pub fn env_truthy(key: &str) -> bool {
    std::env::var(key)
        .map(|v| {
            let v = v.trim();
            !v.is_empty() && v != "0" && !v.eq_ignore_ascii_case("false")
        })
        .unwrap_or(false)
}

include!("daemon_khive_root_tests.rs");

/// Serve one already-admitted test connection through the production frame handler.
///
/// This seam owns no socket path, PID, boot guard, background components, or
/// process-wide shutdown state. The caller owns and joins the connection task.
/// It deliberately does not exercise listener admission or daemon lifecycle.
#[cfg(all(unix, any(test, feature = "test-internals")))]
#[doc(hidden)]
pub async fn serve_connection_for_test<D: DaemonDispatch>(stream: UnixStream, dispatcher: D) {
    handle_conn_with_shutdown(
        stream,
        dispatcher,
        None,
        tokio::time::Instant::now() + INITIAL_FRAME_READ_TIMEOUT,
    )
    .await;
}

#[cfg(all(test, unix))]
#[path = "daemon_tests.rs"]
mod tests;