ai_memory/hooks/

executor.rs

// Copyright 2026 AlphaOne LLC
// SPDX-License-Identifier: Apache-2.0
//
// v0.7 Track G — Task G3: subprocess hook executor (exec + daemon modes).
//
// G1 (PR #554) shipped the `hooks.toml` schema + SIGHUP hot-reload
// plumbing. G2 (PR #563) attached payload structs to every variant
// of `HookEvent`. G3 wires the runtime: given a `HookConfig` and a
// JSON payload, fire the configured subprocess and parse a
// `HookDecision` back from its stdout.
//
// # Two modes
//
// * [`ExecExecutor`] — `tokio::process::Command` per fire. The
//   payload is written to stdin; stdin is then closed; the child's
//   stdout is read to EOF and parsed as a single JSON object.
//   Cheapest model to reason about; ideal for cold or low-rate
//   events (`pre_governance_decision`, `pre_archive`).
//
// * [`DaemonExecutor`] — one long-lived child per `HookConfig`.
//   Frames newline-delimited JSON over stdin/stdout (NDJSON, see
//   "Framing choice" below). Cheap per-fire amortized cost; the
//   right pick for hot events (`post_recall`, `post_search`) where
//   we need to preserve the v0.6.3 50ms recall budget. Reconnects
//   on child crash with exponential backoff (100ms → 5s, capped at
//   5 attempts).
//
// # Framing choice — NDJSON
//
// Newline-delimited JSON. One JSON object per line, no embedded
// newlines (`serde_json::to_writer` + `b'\n'`). Picked over
// length-prefixed for three reasons:
//
//   1. Hook authors can `read_line()` from any language stdlib —
//      no varint or 4-byte-BE length to decode.
//   2. The same stdio pipe is greppable / pipeable for debugging:
//      `tail -f /var/log/hook.log | jq` Just Works.
//   3. Our payloads (`MemoryDelta`, `RecallResult`) never embed
//      raw newlines — `serde_json` encodes `\n` inside strings as
//      `\\n`, so the framing invariant holds without escaping
//      gymnastics on either side of the pipe.
//
// The trade-off is a single malformed line corrupts the rest of
// the stream — but on the daemon path we already reconnect on
// any framing error, so the operator-visible behaviour is the same
// as a child crash: log + exponential backoff + retry.
//
// # Backpressure
//
// The daemon executor wraps each in-flight fire in a `tokio::time::timeout`
// keyed off `HookConfig.timeout_ms`. If the child can't keep up, the
// per-fire deadline trips and we drop the request with a `tracing::warn!`
// + `events_dropped` counter bump. "Oldest first" is a property of the
// single-flight serialization: each fire holds the connection mutex for
// at most `timeout_ms`, so the queue ahead of a slow fire is bounded by
// `timeout_ms × queue_depth` — which is exactly the deadline-drain shape
// the prompt asked for.
//
// # G4 update — HookDecision lifted into `src/hooks/decision.rs`
//
// G3 shipped a local `Allow + Deny` stub of `HookDecision` here so
// the executor had something to deserialize against. G4 replaces
// the stub with the full four-variant enum
// (`Allow / Modify(MemoryDelta) / Deny / AskUser`) in the
// dedicated `decision.rs` module. This file now imports the
// canonical type and routes parse errors through the executor's
// `Decode` variant — failure modes the operator sees (warning
// log + degrade-to-Allow on the dispatcher path) are unchanged.
//
// # Cross-references (G5/G6 shipped post-G3)
//
// * G5 chain ordering / first-deny-wins lives in `hooks/chain.rs`.
//   The executor here is the per-config fire primitive; `chain.rs`
//   composes a sequence of executors into the per-event chain.
// * G6 per-event-class deadlines + multiplier path live in
//   `hooks/timeouts.rs`. This executor still honours
//   `HookConfig.timeout_ms` as the per-fire ceiling; the class
//   deadlines in `timeouts.rs` resolve the budget the chain hands
//   each fire (via the class-multiplier table).
// * G7-G11 firing at the actual memory operation points is wired
//   through the per-event hook sites under `hooks/recall.rs`,
//   `hooks/pre_store/`, and `hooks/post_reflect/`, composed via the
//   chain entry points in `hooks/chain.rs`.

use std::collections::VecDeque;
use std::io;
use std::process::Stdio;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};

use serde::Serialize;
use serde_json::Value;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tokio::time::timeout;

/// Capacity of the per-child stderr ring buffer (in bytes). Sized to
/// `4 KiB` — large enough to surface a typical Python/Node traceback
/// (the diagnostic shape operators reach for first), small enough to
/// keep the per-daemon overhead irrelevant. The drain task pops from
/// the front whenever the buffer would exceed this on push, so the
/// invariant is "the most recent 4 KiB of stderr is always retained".
const STDERR_RING_CAPACITY: usize = 4 * 1024;

// ---------------------------------------------------------------------------
// Spawn-retry backoff (issue #1207) — handle transient fork(2) errnos
// ---------------------------------------------------------------------------
//
// Under parallel test load on macOS (and on any kernel under fork/FD
// pressure), `fork+exec` of a hook child can fail with one of three
// transient errnos:
//
//   * `EAGAIN` — kernel can't allocate a new process / hit RLIMIT_NPROC
//     (libc::EAGAIN: 35 on macOS, 11 on Linux).
//   * `ENOMEM` — kernel running low on memory (libc::ENOMEM = 12).
//   * `EMFILE` — process FD table full (libc::EMFILE = 24).
//
// We also keep the pre-existing `ETXTBSY` (26) retry inline below
// because it's the same shape: transient, kernel-mediated, resolved
// by a few-ms backoff. Issue #1207 added EAGAIN/ENOMEM/EMFILE to the
// same retry loop so the executor stops surfacing transient kernel
// pressure to callers as a hard `ExecutorError::Spawn` — which under
// `FailMode::Open` silently degrades to `ChainResult::Allow` and
// masquerades as a passing test when the child never actually ran
// (the v0.7 G8 `on_index_eviction_fires_with_full_payload` flake).

/// Spawn-retry backoff ladder. Each attempt sleeps for the listed
/// duration BEFORE retrying; the first attempt has no pre-sleep.
/// Total wall-clock budget across all retries is ~1.26s, which keeps
/// the executor well under the typical hook `timeout_ms` ceiling
/// (5_000ms in the v0.7 G8 test fixture).
const SPAWN_RETRY_BACKOFF_MS: &[u64] = &[10, 50, 200, 1_000];

/// Returns `true` iff the io::Error is a transient errno the kernel
/// will likely recover from on a short backoff. On non-unix platforms
/// always returns `false` (Windows doesn't expose these errnos; the
/// caller still gets the original error via the standard Spawn path).
#[cfg(unix)]
fn is_transient_spawn_errno(err: &io::Error) -> bool {
    let Some(errno) = err.raw_os_error() else {
        return false;
    };
    // libc::EAGAIN, ENOMEM, EMFILE, ETXTBSY (the v0.6 race) — all
    // transient under parallel-load conditions. ETXTBSY is the
    // exec-races-write window that PR #563 originally guarded.
    errno == libc::EAGAIN
        || errno == libc::ENOMEM
        || errno == libc::EMFILE
        || errno == libc::ETXTBSY
}

#[cfg(not(unix))]
fn is_transient_spawn_errno(_err: &io::Error) -> bool {
    false
}

/// Fault-injection hook for `spawn_with_transient_retry` unit tests.
/// When set (in test builds only), the next `expected` spawn attempts
/// will be forced to return a synthesized EAGAIN regardless of what
/// the inner closure would return. Each call decrements the counter
/// by one; the closure's real result is returned once the counter
/// hits zero.
///
/// Production callers never read this — the constant
/// `AI_MEMORY_TEST_FORCE_SPAWN_EAGAIN` env var is the only ingress
/// path and the env-var read itself is gated on `cfg(test)`. Even if
/// the env var is set in a release binary, the helper short-circuits
/// at compile time.
#[cfg(all(test, unix))]
fn test_force_eagain_remaining() -> u32 {
    use std::sync::atomic::{AtomicU32, Ordering};
    static REMAINING: AtomicU32 = AtomicU32::new(u32::MAX);
    // Lazy-init from env on first call per process.
    static INIT: std::sync::Once = std::sync::Once::new();
    INIT.call_once(|| {
        let n = std::env::var("AI_MEMORY_TEST_FORCE_SPAWN_EAGAIN")
            .ok()
            .and_then(|s| s.parse::<u32>().ok())
            .unwrap_or(0);
        REMAINING.store(n, Ordering::SeqCst);
    });
    let cur = REMAINING.load(Ordering::SeqCst);
    if cur == 0 {
        return 0;
    }
    REMAINING.fetch_sub(1, Ordering::SeqCst);
    cur
}

/// Spawn a child with bounded retry on transient kernel errnos. The
/// closure is invoked at least once and at most `SPAWN_RETRY_BACKOFF_MS.len() + 1`
/// times. Between attempts we sleep per the backoff ladder. After
/// exhausting retries we surface the original `io::Error` verbatim so
/// the caller can wrap it in the appropriate `ExecutorError::Spawn`
/// (or another variant) and operators still see the precise errno.
///
/// The closure is `FnMut(): io::Result<Child>` rather than
/// `FnOnce` because each retry needs to invoke `Command::spawn()`
/// afresh — `Command` is consumed by `spawn()` so the caller wraps
/// the build-and-spawn pair in the closure.
async fn spawn_with_transient_retry<F>(mut spawn_once: F) -> io::Result<Child>
where
    F: FnMut() -> io::Result<Child>,
{
    let total_attempts = SPAWN_RETRY_BACKOFF_MS.len() + 1;
    let mut last_err: Option<io::Error> = None;
    for attempt in 0..total_attempts {
        if attempt > 0 {
            // Backoff index is (attempt - 1) since attempt 0 has no
            // pre-sleep. attempt 1 → BACKOFF_MS[0] = 10ms, etc.
            let sleep_ms = SPAWN_RETRY_BACKOFF_MS[attempt - 1];
            tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
        }

        // Test-only fault injection — synthesize EAGAIN N times then
        // pass through. Compiled out of release builds entirely. Gated
        // on `unix` because `libc::EAGAIN` lives in the cfg(unix)-only
        // `libc` dep; on Windows the fault injection is a no-op (the
        // transient-spawn-errno classifier is unix-only too, so there
        // is nothing to fault-inject for).
        #[cfg(all(test, unix))]
        {
            if test_force_eagain_remaining() > 0 {
                last_err = Some(io::Error::from_raw_os_error(libc::EAGAIN));
                continue;
            }
        }

        match spawn_once() {
            Ok(child) => return Ok(child),
            Err(e) if is_transient_spawn_errno(&e) => {
                tracing::debug!(
                    attempt = attempt + 1,
                    max = total_attempts,
                    errno = ?e.raw_os_error(),
                    "hooks: transient spawn errno, retrying after backoff"
                );
                last_err = Some(e);
                continue;
            }
            Err(e) => return Err(e),
        }
    }
    Err(last_err.unwrap_or_else(|| io::Error::other("spawn retries exhausted with no error")))
}

/// Bounded ring buffer used by the daemon-mode stderr drain. Wrapped
/// in an `Arc<Mutex<…>>` so the drain task and the executor (which
/// snapshots the contents on timeout / drop) can both reach it without
/// fighting over ownership of the underlying `ChildStderr` handle.
type StderrRing = Arc<Mutex<VecDeque<u8>>>;

/// Spawn a background task that drains `stderr` into `ring`, bounding
/// the retained bytes at [`STDERR_RING_CAPACITY`]. The task exits when
/// the child closes its stderr (EOF) or on any read error — both
/// terminal cases the executor will rediscover on the next exchange,
/// so silent task exit here does not desync the caller.
///
/// Returning the `JoinHandle` lets the caller await/abort on shutdown
/// if it wants to (the daemon executor stores it on the connection so
/// the buffered bytes are still snapshot-able after the child dies).
fn spawn_stderr_drain(mut stderr: ChildStderr, ring: StderrRing) -> JoinHandle<()> {
    tokio::spawn(async move {
        // 4 KiB read buffer matches the ring capacity — one read at
        // most fills the ring, so the worst case is a single
        // pop-from-front pass per chunk.
        let mut chunk = [0u8; 4 * 1024];
        loop {
            match stderr.read(&mut chunk).await {
                Ok(0) => break, // EOF — child closed stderr.
                Ok(n) => {
                    let mut guard = ring.lock().await;
                    guard.extend(&chunk[..n]);
                    // Drop oldest bytes until we're back under cap.
                    // VecDeque's pop_front is O(1) amortised so this
                    // stays cheap even on a flood.
                    while guard.len() > STDERR_RING_CAPACITY {
                        guard.pop_front();
                    }
                }
                Err(_) => break, // Read error — surface as silent EOF; the
                                 // executor will rediscover the failure on
                                 // the next stdout exchange.
            }
        }
    })
}

/// Redact patterns that look like secrets from a stderr tail before
/// it reaches the operator log. Hook authors are already trusted at
/// filesystem scope, but a hostile hook running `printenv >&2; exit 1`
/// should not be able to exfiltrate environment variables (or other
/// secret-shaped strings) into the operator log feed, which may be
/// ingested by less-trusted aggregation systems.
///
/// Two filters: (1) replace the value half of any `VAR=value`
/// assignment where the variable name is shell-identifier-shaped,
/// (2) drop any line matching one of the well-known secret keywords.
/// Conservative — favours over-redaction over leaking.
fn redact_stderr_tail(tail: &str) -> String {
    const SECRET_KEYWORDS: &[&str] = &[
        "secret",
        "password",
        "passwd",
        "token",
        "api_key",
        "apikey",
        "bearer",
        "private_key",
        "private-key",
        " auth",
        "credential",
        "cookie",
        "x-amz-",
        "aws_",
        "ssh-rsa",
        "ssh-ed25519",
        "begin private",
        "begin rsa",
        "begin ec",
        "begin openssh",
    ];
    tail.lines()
        .map(|line| {
            let lower = line.to_ascii_lowercase();
            if let Some(eq_pos) = line.find('=') {
                let prefix: &str = &line[..eq_pos];
                if !prefix.is_empty()
                    && prefix
                        .chars()
                        .all(|c| c.is_ascii_alphanumeric() || c == '_')
                    && prefix
                        .chars()
                        .next()
                        .is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
                {
                    return format!("{prefix}=<redacted>");
                }
            }
            if SECRET_KEYWORDS.iter().any(|kw| lower.contains(kw)) {
                return "<redacted: matched secret-keyword filter>".to_string();
            }
            line.to_string()
        })
        .collect::<Vec<_>>()
        .join("\n")
}

/// Emit a WARN log carrying the redacted stderr tail iff non-empty.
/// Free function (rather than a method) so it can be called from the
/// `exchange` failure arms without re-borrowing `&self` or the
/// connection guard — the borrow-checker won't let us hold `conn`
/// (a `&mut DaemonConnection` from `guard.as_mut()`) and call a
/// method on `&self` that would also need to read `self.config` while
/// `conn` is live.
fn warn_stderr_tail(command: &std::path::Path, stage: &str, tail: &str) {
    if !tail.is_empty() {
        let redacted = redact_stderr_tail(tail);
        tracing::warn!(
            command = %command.display(),
            stage,
            stderr_tail = %redacted,
            "hooks: daemon child stderr at failure (redacted — env-var-shaped values + secret-keyword lines stripped)"
        );
    }
}

/// Snapshot the current contents of a stderr ring as a UTF-8 string.
/// Lossy decoding so a stray non-UTF-8 byte (a hook author's binary
/// dump, an emoji split mid-sequence by the ring's pop_front) doesn't
/// suppress the diagnostic — operators get the closest readable
/// rendering of what the child actually wrote.
async fn snapshot_stderr_ring(ring: &StderrRing) -> String {
    let guard = ring.lock().await;
    let bytes: Vec<u8> = guard.iter().copied().collect();
    String::from_utf8_lossy(&bytes).into_owned()
}

use super::config::HookConfig;
use super::decision::HookDecision;
use super::events::HookEvent;

/// Adapter from the G4 strict-parse path into the executor's
/// existing `Decode` error surface. Keeps `drive_exec_child` /
/// `exchange` callers using one error type while letting the
/// dispatcher (G5) reach for `DecisionParseError`'s named
/// variants when it wants to log the precise failure mode.
fn parse_decision_line(line: &str) -> Result<HookDecision> {
    HookDecision::parse(line).map_err(|e| ExecutorError::Decode {
        reason: e.to_string(),
    })
}

// ---------------------------------------------------------------------------
// HookExecutor trait
// ---------------------------------------------------------------------------

/// Errors surfaced by the executor layer. Hand-rolled `Display +
/// Error` per the v0.7 lesson (no `thiserror` in this crate's hot
/// dependency tree).
#[derive(Debug)]
pub enum ExecutorError {
    /// The configured `command` could not be spawned (missing
    /// binary, permissions, etc.).
    Spawn { command: String, source: io::Error },
    /// I/O failure talking to the child's stdio pipes.
    Io(io::Error),
    /// The child returned non-zero or closed its stdout without
    /// writing a decision.
    ChildExit { code: Option<i32>, stderr: String },
    /// The child wrote a payload we could not parse as a
    /// [`HookDecision`].
    Decode { reason: String },
    /// The fire deadline (`HookConfig.timeout_ms`) elapsed before
    /// the child returned a decision.
    Timeout { ms: u64 },
    /// The daemon child crashed or was unreachable after exhausting
    /// the reconnect budget.
    DaemonUnavailable { attempts: u32 },
    /// v0.7.0 (issue #691 fold-1) — the governance pre-action wire
    /// hook refused the `ProcessSpawn` action. `reason` carries the
    /// operator-authored explanation from the matched rule. Surfaced
    /// to the chain runner so the cascade policy can treat it as a
    /// distinct outcome from a Spawn / Io error.
    GovernanceRefused { command: String, reason: String },
}

impl std::fmt::Display for ExecutorError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ExecutorError::Spawn { command, source } => {
                write!(f, "hook spawn failed for {command}: {source}")
            }
            ExecutorError::Io(e) => write!(f, "hook io error: {e}"),
            ExecutorError::ChildExit { code, stderr: _ } => {
                // P2 (#628 agent-5): the stderr tail is in the operator
                // log (redacted via `redact_stderr_tail`) — do not
                // include it here. `Display for ExecutorError` flows
                // into `ChainResult::Deny.reason`, which is sent back
                // to the JSON-RPC caller; surfacing raw stderr there
                // is a credential-exfiltration vector for hostile
                // hooks (`printenv >&2; exit 1`).
                let code_str = code.map_or_else(|| "<signaled>".into(), |c| c.to_string());
                write!(
                    f,
                    "hook child exited (code {code_str}); see operator log for redacted stderr"
                )
            }
            ExecutorError::Decode { reason } => {
                write!(f, "hook decision decode failed: {reason}")
            }
            ExecutorError::Timeout { ms } => {
                write!(f, "hook timed out after {ms}ms")
            }
            ExecutorError::DaemonUnavailable { attempts } => {
                write!(
                    f,
                    "hook daemon unavailable after {attempts} reconnect attempts"
                )
            }
            ExecutorError::GovernanceRefused { command, reason } => {
                write!(
                    f,
                    "hook spawn refused by governance for {command}: {reason}"
                )
            }
        }
    }
}

impl std::error::Error for ExecutorError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            ExecutorError::Spawn { source, .. } | ExecutorError::Io(source) => Some(source),
            _ => None,
        }
    }
}

impl From<io::Error> for ExecutorError {
    fn from(value: io::Error) -> Self {
        ExecutorError::Io(value)
    }
}

/// `Result` alias used across the executor surface.
pub type Result<T> = std::result::Result<T, ExecutorError>;

/// Trait every executor implementation satisfies. `fire` is the
/// single hot-path method G5 will iterate over when stitching
/// chains together.
///
/// `Send + Sync` is mandatory: the registry hands out
/// `Arc<dyn HookExecutor>` and the chain runner (G5) drives fires
/// from arbitrary tokio worker threads.
pub trait HookExecutor: Send + Sync {
    /// Fire the hook for `event` with `payload`. Returns the
    /// child's [`HookDecision`] or an [`ExecutorError`] on
    /// spawn / IO / decode / timeout failure.
    ///
    /// This is `async` via the `BoxFuture` shape because trait
    /// objects + `async fn in trait` is still rough on stable
    /// (the auto-trait inference for `Send` doesn't carry across
    /// the dyn boundary). `BoxFuture<'_, Result<HookDecision>>`
    /// is the same shape `tower::Service` settled on.
    fn fire<'a>(
        &'a self,
        event: HookEvent,
        payload: Value,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<HookDecision>> + Send + 'a>>;

    /// Snapshot of executor metrics. Surfaced by `ai-memory doctor
    /// --tokens --hooks` (see `src/cli/doctor.rs`).
    fn metrics(&self) -> ExecutorMetrics;
}

// ---------------------------------------------------------------------------
// ExecutorMetrics — backpressure observability
// ---------------------------------------------------------------------------

/// Per-executor metrics surfaced by `ai-memory doctor`.
///
/// These are *snapshots*; the executor accumulates raw counters
/// internally and projects to this struct on demand. See
/// [`MetricsCounters`] for the live atomics.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub struct ExecutorMetrics {
    pub events_fired: u64,
    pub events_dropped: u64,
    pub mean_latency_us: u64,
}

#[derive(Debug, Default)]
struct MetricsCounters {
    fired: AtomicU64,
    dropped: AtomicU64,
    latency_sum_us: AtomicU64,
    latency_n: AtomicU64,
}

impl MetricsCounters {
    fn record_fire(&self, latency: Duration) {
        self.fired.fetch_add(1, Ordering::Relaxed);
        let us = u64::try_from(latency.as_micros()).unwrap_or(u64::MAX);
        self.latency_sum_us.fetch_add(us, Ordering::Relaxed);
        self.latency_n.fetch_add(1, Ordering::Relaxed);
    }

    fn record_drop(&self) {
        self.dropped.fetch_add(1, Ordering::Relaxed);
    }

    fn snapshot(&self) -> ExecutorMetrics {
        let fired = self.fired.load(Ordering::Relaxed);
        let dropped = self.dropped.load(Ordering::Relaxed);
        let n = self.latency_n.load(Ordering::Relaxed);
        let sum = self.latency_sum_us.load(Ordering::Relaxed);
        let mean_latency_us = if n == 0 { 0 } else { sum / n };
        ExecutorMetrics {
            events_fired: fired,
            events_dropped: dropped,
            mean_latency_us,
        }
    }
}

// ---------------------------------------------------------------------------
// FireEnvelope — wire shape sent to the child
// ---------------------------------------------------------------------------

/// JSON envelope written to a hook subprocess on every fire.
///
/// Keeping `event` separate from `payload` lets the child route
/// without re-parsing the payload bag — useful for daemon mode,
/// where one child might subscribe to several events.
#[derive(Debug, Serialize)]
struct FireEnvelope<'a> {
    event: HookEvent,
    payload: &'a Value,
}

// ---------------------------------------------------------------------------
// ExecExecutor — subprocess per fire
// ---------------------------------------------------------------------------

/// Subprocess-per-fire executor. Spawns a fresh child for every
/// event; closes stdin to signal "no more input"; reads stdout to
/// EOF and parses a single [`HookDecision`].
///
/// Cheapest mental model. Right pick for low-rate events. Hot
/// events should configure `mode = "daemon"` instead.
pub struct ExecExecutor {
    config: HookConfig,
    metrics: MetricsCounters,
}

/// SEC-17 (Cluster D, issue #767) — convert an `OsStr` (the raw
/// command path) into the wire-check `AgentAction::ProcessSpawn.binary`
/// string with the highest fidelity available on the host. On Unix
/// the underlying bytes are forwarded verbatim through
/// `OsStrExt::as_bytes` + `String::from_utf8_lossy` (the lossy
/// conversion is a substitution, not a truncation — every non-UTF-8
/// byte becomes a U+FFFD REPLACEMENT CHARACTER, preserving the byte
/// count so a downstream substring match cannot accidentally collide
/// with a sanitised value). On Windows / wasm the standard
/// `OsStr::to_string_lossy` path is the best the platform exposes.
///
/// Why not `display().to_string()` everywhere? `PathBuf::display` is
/// designed for human-facing output and may collapse or reshape
/// non-UTF-8 sequences depending on the platform. The matcher engine
/// is a security boundary — it must see the same bytes the kernel
/// will exec, with no platform-specific rewriting.
#[cfg(unix)]
fn os_str_lossless_for_wire_check(s: &std::ffi::OsStr) -> String {
    use std::os::unix::ffi::OsStrExt;
    // `from_utf8_lossy` substitutes invalid bytes with U+FFFD WITHOUT
    // shrinking the byte length (each invalid byte becomes one
    // replacement char), so a substring matcher sees a stable shape.
    String::from_utf8_lossy(s.as_bytes()).into_owned()
}

/// Non-Unix fallback — `to_string_lossy` is the best the platform
/// surface exposes. On Windows the path is WTF-8 internally so the
/// lossy conversion is generally a no-op for legitimate paths.
#[cfg(not(unix))]
fn os_str_lossless_for_wire_check(s: &std::ffi::OsStr) -> String {
    s.to_string_lossy().into_owned()
}

impl ExecExecutor {
    #[must_use]
    pub fn new(config: HookConfig) -> Self {
        Self {
            config,
            metrics: MetricsCounters::default(),
        }
    }

    async fn fire_inner(&self, event: HookEvent, payload: Value) -> Result<HookDecision> {
        // v0.7.0 (issue #691 fold-1) — wire the ProcessSpawn governance
        // gate BEFORE the Command::new(...).spawn() call. The closure
        // installed by bootstrap_serve consults the governance_rules
        // table for a refusal verdict (e.g. R004 — cargo forbidden on
        // low-disk system). Refusal short-circuits cleanly with a
        // typed ExecutorError::GovernanceRefused so the chain runner
        // can apply the cascade policy without confusing it with a
        // legitimate spawn IO failure.
        //
        // SEC-13 / SEC-17 (Cluster D, issue #767) — build the binary
        // identifier from the raw `OsStr` (NOT
        // `display().to_string()`, which is lossy on non-UTF-8 path
        // injections). The lossy conversion is reserved for log
        // surfaces / error messages where readability outweighs the
        // fidelity loss. The matcher engine sees the full OS-string
        // representation so a non-UTF-8 path injection does not
        // sneak past a substring-match rule. We pass an empty argv
        // here because the hook executor never spawns the child with
        // positional args (HookConfig has no `args` field); the
        // matcher's optional `args_contain` field is a no-op for the
        // hooks path but ready for the next caller (CLI shell-out,
        // skill-export, etc.) that adds positional args.
        let binary = os_str_lossless_for_wire_check(self.config.command.as_os_str());
        let command_str = self.config.command.display().to_string();
        let spawn_action = crate::governance::agent_action::AgentAction::ProcessSpawn {
            binary,
            args: Vec::new(),
        };
        if let Err(refusal) = crate::governance::wire_check::check(&spawn_action) {
            return Err(ExecutorError::GovernanceRefused {
                command: command_str,
                reason: refusal.reason,
            });
        }
        let envelope = FireEnvelope {
            event,
            payload: &payload,
        };
        let envelope_bytes = serde_json::to_vec(&envelope).map_err(|e| ExecutorError::Decode {
            reason: format!("envelope encode: {e}"),
        })?;

        // Spawn-retry-with-backoff (issue #1207). Covers:
        //   * Linux/macOS ETXTBSY ("Text file busy") — exec() races
        //     with another process that still holds the file open for
        //     write. The pre-#1207 ETXTBSY loop is now folded into
        //     `spawn_with_transient_retry`.
        //   * EAGAIN/ENOMEM/EMFILE — kernel under fork/FD pressure
        //     under parallel test load. The v0.7 G8 flake
        //     (`on_index_eviction_fires_with_full_payload`) was the
        //     trigger; pre-#1207 the executor surfaced EAGAIN to the
        //     caller, `FailMode::Open` masked it as `ChainResult::Allow`,
        //     and the child never actually ran.
        let command_path = self.config.command.clone();
        let child = spawn_with_transient_retry(|| {
            Command::new(&command_path)
                .stdin(Stdio::piped())
                .stdout(Stdio::piped())
                .stderr(Stdio::piped())
                .kill_on_drop(true)
                .spawn()
        })
        .await
        .map_err(|source| ExecutorError::Spawn {
            command: self.config.command.display().to_string(),
            source,
        })?;

        let started = Instant::now();
        let deadline = Duration::from_millis(u64::from(self.config.timeout_ms));
        let command_str = self.config.command.display().to_string();

        let driven = timeout(
            deadline,
            drive_exec_child(child, envelope_bytes, &command_str),
        )
        .await;

        match driven {
            Ok(Ok(decision)) => {
                self.metrics.record_fire(started.elapsed());
                Ok(decision)
            }
            Ok(Err(e)) => {
                // Even on child error, we count the latency we
                // actually paid — the budget is a wall-clock figure.
                self.metrics.record_fire(started.elapsed());
                Err(e)
            }
            Err(_elapsed) => {
                self.metrics.record_drop();
                // The Child has been moved into drive_exec_child; that
                // future is dropped on timeout, which fires the
                // `kill_on_drop` knob set above and reaps the process.
                Err(ExecutorError::Timeout {
                    ms: u64::from(self.config.timeout_ms),
                })
            }
        }
    }
}

impl HookExecutor for ExecExecutor {
    fn fire<'a>(
        &'a self,
        event: HookEvent,
        payload: Value,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<HookDecision>> + Send + 'a>>
    {
        Box::pin(self.fire_inner(event, payload))
    }

    fn metrics(&self) -> ExecutorMetrics {
        self.metrics.snapshot()
    }
}

async fn drive_exec_child(
    mut child: Child,
    envelope: Vec<u8>,
    command: &str,
) -> Result<HookDecision> {
    // Write the envelope, then close stdin so the child knows it
    // can finish. `take()` drops the handle on the way out.
    if let Some(mut stdin) = child.stdin.take() {
        stdin.write_all(&envelope).await?;
        stdin.write_all(b"\n").await?;
        stdin.shutdown().await?;
    }

    let output = child.wait_with_output().await?;
    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr).into_owned();
        return Err(ExecutorError::ChildExit {
            code: output.status.code(),
            stderr,
        });
    }

    // H9 fix — surface stderr on the success path too. Operators
    // debugging a flapping hook need to see whatever diagnostics the
    // child wrote even when the decision parses cleanly; silently
    // dropping stderr here was the v0.7.0 review #628 blocker. Cap
    // the logged slice at the same `STDERR_RING_CAPACITY` the daemon
    // path uses so an over-eager hook can't pin tracing's allocator.
    if !output.stderr.is_empty() {
        let trimmed_len = output.stderr.len().min(STDERR_RING_CAPACITY);
        let start = output.stderr.len() - trimmed_len;
        let stderr_tail = String::from_utf8_lossy(&output.stderr[start..]);
        tracing::debug!(
            command,
            stderr_bytes = output.stderr.len(),
            stderr = %stderr_tail,
            "hooks: exec child wrote stderr on success path"
        );
    }

    let stdout = String::from_utf8_lossy(&output.stdout);
    // The child may write multiple newlines; the decision is the
    // last non-empty line. Hooks that print debug output to stdout
    // before their decision Just Work this way.
    let decision_line = stdout
        .lines()
        .filter(|l| !l.trim().is_empty())
        .next_back()
        .unwrap_or("");
    parse_decision_line(decision_line)
}

// ---------------------------------------------------------------------------
// DaemonExecutor — long-lived child + NDJSON framing
// ---------------------------------------------------------------------------

/// Per-`HookConfig` long-lived child. One executor owns one child
/// at a time; fires are serialized through a single connection
/// mutex (NDJSON request → NDJSON response). On framing error or
/// child exit, the connection is dropped and the next fire
/// reconnects with exponential backoff.
pub struct DaemonExecutor {
    config: HookConfig,
    conn: Mutex<Option<DaemonConnection>>,
    metrics: MetricsCounters,
}

struct DaemonConnection {
    child: Child,
    stdin: ChildStdin,
    stdout: BufReader<ChildStdout>,
    /// Bounded ring of the most recent [`STDERR_RING_CAPACITY`] bytes
    /// the child wrote to stderr. Populated by `stderr_drain` in the
    /// background; snapshotted by the executor on timeout / error /
    /// drop so operators see the child's diagnostics next to the
    /// failure that surfaced them.
    stderr_buf: StderrRing,
    /// Handle to the stderr-drain task. Held so we can `abort()` it
    /// when the connection is torn down — without this the task
    /// would linger until the kernel closes stderr after the child's
    /// reaping, which is racy under stress.
    stderr_task: JoinHandle<()>,
}

impl DaemonConnection {
    /// Snapshot the buffered stderr (lossy UTF-8) for inclusion in a
    /// failure log. Only called on the slow / failure paths; the
    /// happy path never touches the ring.
    async fn stderr_tail(&self) -> String {
        snapshot_stderr_ring(&self.stderr_buf).await
    }
}

impl Drop for DaemonConnection {
    fn drop(&mut self) {
        // Abort the drain task — the ChildStderr it owns is going
        // out of scope and the OS will close the pipe; no point
        // waiting for the EOF roundtrip.
        self.stderr_task.abort();
    }
}

impl DaemonExecutor {
    #[must_use]
    pub fn new(config: HookConfig) -> Self {
        Self {
            config,
            conn: Mutex::new(None),
            metrics: MetricsCounters::default(),
        }
    }

    async fn fire_inner(&self, event: HookEvent, payload: Value) -> Result<HookDecision> {
        let envelope = FireEnvelope {
            event,
            payload: &payload,
        };
        let mut envelope_bytes =
            serde_json::to_vec(&envelope).map_err(|e| ExecutorError::Decode {
                reason: format!("envelope encode: {e}"),
            })?;
        envelope_bytes.push(b'\n');

        let started = Instant::now();
        let deadline = Duration::from_millis(u64::from(self.config.timeout_ms));

        let driven = timeout(deadline, self.exchange(&envelope_bytes)).await;
        match driven {
            Ok(Ok(decision)) => {
                self.metrics.record_fire(started.elapsed());
                Ok(decision)
            }
            Ok(Err(e)) => {
                self.metrics.record_fire(started.elapsed());
                Err(e)
            }
            Err(_elapsed) => {
                self.metrics.record_drop();
                // Drop the connection so the next fire reconnects;
                // a slow daemon may still write the response into
                // the pipe after we've moved on, which would desync
                // subsequent fires.
                //
                // H9 fix — before tearing down the connection, snapshot
                // the buffered stderr so the operator log carries
                // whatever the child was complaining about right up
                // to the deadline. Without this, a hook that's
                // genuinely failing inside its handler (panic
                // backtrace on stderr, but no decision on stdout)
                // would just look like a generic "Timeout" with no
                // diagnostic context.
                let mut guard = self.conn.lock().await;
                if let Some(conn) = guard.as_ref() {
                    let tail = conn.stderr_tail().await;
                    if !tail.is_empty() {
                        tracing::warn!(
                            command = %self.config.command.display(),
                            timeout_ms = self.config.timeout_ms,
                            stderr_tail = %tail,
                            "hooks: daemon child stderr at timeout"
                        );
                    }
                }
                *guard = None;
                Err(ExecutorError::Timeout {
                    ms: u64::from(self.config.timeout_ms),
                })
            }
        }
    }

    /// Write one envelope, read one decision line. On any IO /
    /// framing error the connection is dropped before returning so
    /// the next fire goes through `connect_with_backoff`.
    async fn exchange(&self, envelope: &[u8]) -> Result<HookDecision> {
        let mut guard = self.conn.lock().await;
        if guard.is_none() {
            *guard = Some(self.connect_with_backoff().await?);
        }
        // Safe: just inserted if missing.
        let conn = guard.as_mut().expect("connection just inserted");

        if let Err(e) = conn.stdin.write_all(envelope).await {
            // H9 fix — snapshot buffered stderr before tearing down
            // so the operator log carries the child's diagnostics.
            let tail = conn.stderr_tail().await;
            warn_stderr_tail(&self.config.command, "stdin write", &tail);
            *guard = None;
            return Err(ExecutorError::Io(e));
        }
        if let Err(e) = conn.stdin.flush().await {
            let tail = conn.stderr_tail().await;
            warn_stderr_tail(&self.config.command, "stdin flush", &tail);
            *guard = None;
            return Err(ExecutorError::Io(e));
        }

        let mut line = String::new();
        match conn.stdout.read_line(&mut line).await {
            Ok(0) => {
                // EOF — child closed its stdout (likely crashed).
                // H9 fix — surface buffered stderr in the error so
                // the operator sees the child's last words instead
                // of a generic "closed stdout".
                let tail = conn.stderr_tail().await;
                let stderr = if tail.is_empty() {
                    "daemon child closed stdout".into()
                } else {
                    format!("daemon child closed stdout; stderr tail: {tail}")
                };
                *guard = None;
                Err(ExecutorError::ChildExit { code: None, stderr })
            }
            Ok(_) => match parse_decision_line(&line) {
                Ok(d) => Ok(d),
                Err(e) => {
                    // Framing error — reset the connection so the
                    // next fire doesn't read into a half-consumed
                    // envelope. Log buffered stderr at WARN so the
                    // operator can correlate the bad framing with
                    // any hook-side panic that produced it.
                    let tail = conn.stderr_tail().await;
                    warn_stderr_tail(&self.config.command, "framing error", &tail);
                    *guard = None;
                    Err(e)
                }
            },
            Err(e) => {
                let tail = conn.stderr_tail().await;
                warn_stderr_tail(&self.config.command, "stdout read", &tail);
                *guard = None;
                Err(ExecutorError::Io(e))
            }
        }
    }

    /// Spawn the child with exponential backoff (100ms → 5s, max 5
    /// attempts). Returns the connected handles or
    /// [`ExecutorError::DaemonUnavailable`] on exhaustion.
    async fn connect_with_backoff(&self) -> Result<DaemonConnection> {
        const MAX_ATTEMPTS: u32 = 5;
        const BASE_BACKOFF_MS: u64 = 100;
        const MAX_BACKOFF_MS: u64 = 5_000;

        let mut last_err: Option<ExecutorError> = None;
        for attempt in 0..MAX_ATTEMPTS {
            if attempt > 0 {
                // Exponential backoff: 100, 200, 400, 800, 1600… capped.
                let pow = 1u64 << (attempt - 1);
                let backoff_ms = (BASE_BACKOFF_MS.saturating_mul(pow)).min(MAX_BACKOFF_MS);
                tokio::time::sleep(Duration::from_millis(backoff_ms)).await;
            }
            match self.spawn_one().await {
                Ok(conn) => return Ok(conn),
                Err(e) => {
                    tracing::warn!(
                        attempt = attempt + 1,
                        max = MAX_ATTEMPTS,
                        error = %e,
                        "hooks: daemon spawn attempt failed"
                    );
                    last_err = Some(e);
                }
            }
        }
        Err(last_err.unwrap_or(ExecutorError::DaemonUnavailable {
            attempts: MAX_ATTEMPTS,
        }))
    }

    async fn spawn_one(&self) -> Result<DaemonConnection> {
        // v0.7.0 (issue #691 fold-1) — same ProcessSpawn gate as the
        // exec-mode path above. Daemon-mode hooks spawn at most once
        // per (process, hook) so this fires at start-up rather than
        // per-event; a governance refusal here aborts the daemon
        // connection attempt cleanly.
        let command_str = self.config.command.display().to_string();
        let spawn_action = crate::governance::agent_action::AgentAction::ProcessSpawn {
            binary: command_str.clone(),
            args: Vec::new(),
        };
        if let Err(refusal) = crate::governance::wire_check::check(&spawn_action) {
            return Err(ExecutorError::GovernanceRefused {
                command: command_str,
                reason: refusal.reason,
            });
        }
        // Spawn-retry-with-backoff (issue #1207) for the daemon path
        // too. The outer `connect_with_backoff` retries on any
        // ExecutorError (5 attempts, 100ms → 5s); this inner loop
        // catches the narrow transient-errno class first so a brief
        // fork-pressure window doesn't even consume a reconnect slot.
        let command_path = self.config.command.clone();
        let mut child = spawn_with_transient_retry(|| {
            Command::new(&command_path)
                .stdin(Stdio::piped())
                .stdout(Stdio::piped())
                .stderr(Stdio::piped())
                .spawn()
        })
        .await
        .map_err(|source| ExecutorError::Spawn {
            command: self.config.command.display().to_string(),
            source,
        })?;
        let stdin = child.stdin.take().ok_or_else(|| {
            ExecutorError::Io(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "child stdin not piped",
            ))
        })?;
        let stdout = child.stdout.take().ok_or_else(|| {
            ExecutorError::Io(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "child stdout not piped",
            ))
        })?;
        // H9 fix — drain stderr in the background so a verbose hook
        // can't fill the OS pipe buffer (typically 64 KiB on Linux,
        // ~16 KiB on macOS) and deadlock the child on its next
        // `write(2)` to stderr — which would in turn deadlock us
        // waiting for the next NDJSON response on stdout.
        let stderr = child.stderr.take().ok_or_else(|| {
            ExecutorError::Io(io::Error::new(
                io::ErrorKind::BrokenPipe,
                "child stderr not piped",
            ))
        })?;
        let stderr_buf: StderrRing =
            Arc::new(Mutex::new(VecDeque::with_capacity(STDERR_RING_CAPACITY)));
        let stderr_task = spawn_stderr_drain(stderr, Arc::clone(&stderr_buf));
        Ok(DaemonConnection {
            child,
            stdin,
            stdout: BufReader::new(stdout),
            stderr_buf,
            stderr_task,
        })
    }
}

impl HookExecutor for DaemonExecutor {
    fn fire<'a>(
        &'a self,
        event: HookEvent,
        payload: Value,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<HookDecision>> + Send + 'a>>
    {
        Box::pin(self.fire_inner(event, payload))
    }

    fn metrics(&self) -> ExecutorMetrics {
        self.metrics.snapshot()
    }
}

impl Drop for DaemonExecutor {
    fn drop(&mut self) {
        // Best-effort: kill the child so a reload doesn't leak
        // long-lived processes. tokio::process::Child has a
        // `kill_on_drop` knob but we can't reach it from here
        // without the conn lock; this Drop is the belt to that
        // suspenders.
        if let Ok(mut guard) = self.conn.try_lock() {
            if let Some(conn) = guard.as_mut() {
                let _ = conn.child.start_kill();
            }
        }
    }
}

// ---------------------------------------------------------------------------
// ExecutorRegistry
// ---------------------------------------------------------------------------

/// Maps `HookConfig` → `Arc<dyn HookExecutor>`. Built once per
/// `hooks.toml` load (see `src/hooks/config.rs::spawn_reload_task`)
/// and held behind the same `Arc<RwLock<…>>` that owns the config
/// snapshot. G5's chain runner consumes the registry's outputs.
///
/// The cache is keyed on `HookConfig` (full struct equality) so two
/// different hooks pointing at the same binary still get distinct
/// executors — one daemon child per `[[hook]]` block, never shared.
/// Sharing would let one event's slow fire starve another's
/// connection mutex, which is the opposite of what daemon mode
/// buys us.
///
/// Backed by a `Vec<(HookConfig, …)>` rather than a `HashMap`:
/// `HookConfig` carries a `PathBuf` and a `String` (no `Hash`
/// derivation today), and the cache cardinality is bounded by the
/// number of `[[hook]]` blocks in `hooks.toml` — a linear scan over
/// a few dozen entries is dwarfed by the spawn cost it gates.
pub struct ExecutorRegistry {
    cache: Vec<(HookConfig, Arc<dyn HookExecutor>)>,
}

impl ExecutorRegistry {
    #[must_use]
    pub fn new() -> Self {
        Self { cache: Vec::new() }
    }

    /// Build a registry pre-populated with one executor per entry
    /// in `hooks`. Convenience for the bootstrap path; callers
    /// driving the SIGHUP reload should call [`Self::get`] lazily
    /// instead so dropping a `[[hook]]` from the config tears down
    /// its long-lived child.
    #[must_use]
    pub fn from_hooks(hooks: &[HookConfig]) -> Self {
        let mut me = Self::new();
        for h in hooks {
            let _ = me.get(h);
        }
        me
    }

    /// Return the executor for `hook`, constructing one on first
    /// touch. Subsequent calls with an *equal* `HookConfig` return
    /// the same `Arc`.
    pub fn get(&mut self, hook: &HookConfig) -> Arc<dyn HookExecutor> {
        if let Some((_, existing)) = self.cache.iter().find(|(cfg, _)| cfg == hook) {
            return Arc::clone(existing);
        }
        let executor: Arc<dyn HookExecutor> = match hook.mode {
            super::config::HookMode::Exec => Arc::new(ExecExecutor::new(hook.clone())),
            super::config::HookMode::Daemon => Arc::new(DaemonExecutor::new(hook.clone())),
        };
        self.cache.push((hook.clone(), Arc::clone(&executor)));
        executor
    }

    /// Iterate `(HookConfig, ExecutorMetrics)` pairs. `ai-memory
    /// doctor --tokens --hooks` calls this to render the
    /// per-executor backpressure table.
    pub fn metrics(&self) -> Vec<(HookConfig, ExecutorMetrics)> {
        self.cache
            .iter()
            .map(|(cfg, ex)| (cfg.clone(), ex.metrics()))
            .collect()
    }

    /// Number of cached executors. Cheap accessor for tests.
    #[must_use]
    pub fn len(&self) -> usize {
        self.cache.len()
    }

    /// Whether the registry is empty.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.cache.is_empty()
    }
}

impl Default for ExecutorRegistry {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Tests — unit only; the integration tests live in
// `tests/hooks_executor_test.rs` so they can spawn real subprocesses.
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hooks::config::HookMode;

    fn cfg(mode: HookMode) -> HookConfig {
        HookConfig {
            event: HookEvent::PostStore,
            command: std::path::PathBuf::from("/bin/true"),
            priority: 0,
            timeout_ms: 1_000,
            mode,
            enabled: true,
            namespace: "*".into(),
            fail_mode: crate::hooks::config::FailMode::Open,
        }
    }

    // The G3 stub's parse path now lives in `decision.rs`. These
    // tests cover the executor-side adapter (`parse_decision_line`)
    // which wraps `DecisionParseError` into `ExecutorError::Decode`
    // — the failure mode that surfaces on the daemon stdout path.

    #[test]
    fn parse_decision_line_allow_default_on_empty() {
        assert_eq!(parse_decision_line("").unwrap(), HookDecision::Allow);
        assert_eq!(parse_decision_line("   ").unwrap(), HookDecision::Allow);
        assert_eq!(parse_decision_line("{}").unwrap(), HookDecision::Allow);
    }

    // P2 (#628 agent-5) — stderr secret-redaction.
    #[test]
    fn redact_stderr_strips_env_var_assignments() {
        let raw = "AWS_SECRET_ACCESS_KEY=AKIA1234567890ABCDEF\nDATABASE_URL=postgres://u:p@h/db\nGITHUB_TOKEN=ghp_abcdef\nuser-message-fine\n";
        let red = redact_stderr_tail(raw);
        assert!(!red.contains("AKIA1234567890ABCDEF"));
        assert!(!red.contains("ghp_abcdef"));
        assert!(red.contains("AWS_SECRET_ACCESS_KEY=<redacted>"));
        assert!(red.contains("GITHUB_TOKEN=<redacted>"));
        // Unrelated lines pass through.
        assert!(red.contains("user-message-fine"));
    }

    #[test]
    fn redact_stderr_drops_secret_keyword_lines() {
        let raw = "Authorization: Bearer eyJ.fake.jwt\nset-cookie: session=abc\nmsg=normal\n";
        let red = redact_stderr_tail(raw);
        assert!(!red.contains("Bearer eyJ.fake.jwt"));
        assert!(!red.contains("session=abc"));
    }

    #[test]
    fn child_exit_display_excludes_stderr_content() {
        let err = ExecutorError::ChildExit {
            code: Some(1),
            stderr: "AWS_SECRET_ACCESS_KEY=AKIA-secret-content".to_string(),
        };
        let s = err.to_string();
        // Stderr content must NOT flow into the user-visible Display
        // (which becomes ChainResult::Deny.reason → JSON-RPC caller).
        assert!(!s.contains("AKIA-secret-content"));
        assert!(!s.contains("AWS_SECRET_ACCESS_KEY"));
        // Exit code IS surfaced — operator log carries the full
        // (redacted) stderr separately.
        assert!(s.contains("code 1"));
    }

    #[test]
    fn parse_decision_line_allow_explicit() {
        let d = parse_decision_line(r#"{"action":"allow"}"#).unwrap();
        assert_eq!(d, HookDecision::Allow);
    }

    #[test]
    fn parse_decision_line_deny_with_default_code() {
        let d = parse_decision_line(r#"{"action":"deny","reason":"nope"}"#).unwrap();
        match d {
            HookDecision::Deny { reason, code } => {
                assert_eq!(reason, "nope");
                assert_eq!(code, 403);
            }
            other => panic!("expected Deny, got {other:?}"),
        }
    }

    #[test]
    fn parse_decision_line_deny_with_explicit_code() {
        let d = parse_decision_line(r#"{"action":"deny","reason":"x","code":429}"#).unwrap();
        match d {
            HookDecision::Deny { code, .. } => assert_eq!(code, 429),
            _ => panic!("expected Deny"),
        }
    }

    #[test]
    fn parse_decision_line_unknown_action_wraps_to_decode() {
        // G4 recognises `modify` so the canonical "unknown action"
        // case becomes a deliberately bogus discriminator.
        let err = parse_decision_line(r#"{"action":"explode"}"#).unwrap_err();
        match err {
            ExecutorError::Decode { reason } => {
                assert!(
                    reason.contains("unknown action"),
                    "decode reason should name the failure: {reason}"
                );
            }
            other => panic!("expected Decode, got {other:?}"),
        }
    }

    #[test]
    fn parse_decision_line_modify_now_recognised() {
        // G3's stub rejected `modify`; G4 lifts it into the wire
        // contract, so the executor must round-trip it cleanly.
        let d = parse_decision_line(r#"{"action":"modify","delta":{"priority":7}}"#).unwrap();
        match d {
            HookDecision::Modify(m) => assert_eq!(m.delta.priority, Some(7)),
            other => panic!("expected Modify, got {other:?}"),
        }
    }

    #[test]
    fn metrics_counters_track_fired_dropped_and_mean() {
        let m = MetricsCounters::default();
        m.record_fire(Duration::from_micros(100));
        m.record_fire(Duration::from_micros(300));
        m.record_drop();
        let snap = m.snapshot();
        assert_eq!(snap.events_fired, 2);
        assert_eq!(snap.events_dropped, 1);
        assert_eq!(snap.mean_latency_us, 200);
    }

    #[test]
    fn metrics_counters_zero_when_no_fires() {
        let snap = MetricsCounters::default().snapshot();
        assert_eq!(snap.events_fired, 0);
        assert_eq!(snap.events_dropped, 0);
        assert_eq!(snap.mean_latency_us, 0);
    }

    #[test]
    fn registry_caches_per_hook_config() {
        let mut reg = ExecutorRegistry::new();
        let a = cfg(HookMode::Exec);
        let b = cfg(HookMode::Exec);
        let e1 = reg.get(&a);
        let e2 = reg.get(&b);
        assert_eq!(reg.len(), 1, "equal HookConfigs must dedupe");
        assert!(Arc::ptr_eq(&e1, &e2), "same Arc on cache hit");
    }

    #[test]
    fn registry_distinct_executors_for_distinct_modes() {
        let mut reg = ExecutorRegistry::new();
        let exec_cfg = cfg(HookMode::Exec);
        let mut daemon_cfg = cfg(HookMode::Daemon);
        // Bump priority so the configs are unequal even though
        // command path is identical.
        daemon_cfg.priority = 99;
        reg.get(&exec_cfg);
        reg.get(&daemon_cfg);
        assert_eq!(reg.len(), 2);
    }

    #[test]
    fn registry_from_hooks_prepopulates() {
        let hooks = vec![cfg(HookMode::Exec), {
            let mut d = cfg(HookMode::Daemon);
            d.priority = 1;
            d
        }];
        let reg = ExecutorRegistry::from_hooks(&hooks);
        assert_eq!(reg.len(), 2);
    }

    #[test]
    fn registry_metrics_starts_at_zero() {
        let mut reg = ExecutorRegistry::new();
        let _ = reg.get(&cfg(HookMode::Exec));
        let metrics = reg.metrics();
        assert_eq!(metrics.len(), 1);
        assert_eq!(metrics[0].1.events_fired, 0);
        assert_eq!(metrics[0].1.events_dropped, 0);
    }

    #[test]
    fn executor_error_display_formats_each_variant() {
        let cases: Vec<ExecutorError> = vec![
            ExecutorError::Spawn {
                command: "/bin/x".into(),
                source: io::Error::new(io::ErrorKind::NotFound, "no"),
            },
            ExecutorError::Io(io::Error::new(io::ErrorKind::Other, "boom")),
            ExecutorError::ChildExit {
                code: Some(42),
                stderr: "stderr msg".into(),
            },
            ExecutorError::Decode {
                reason: "bad json".into(),
            },
            ExecutorError::Timeout { ms: 1234 },
            ExecutorError::DaemonUnavailable { attempts: 5 },
            ExecutorError::GovernanceRefused {
                command: "/usr/bin/cargo".into(),
                reason: "R004: cargo forbidden on low-disk".into(),
            },
        ];
        for e in cases {
            let s = e.to_string();
            assert!(!s.is_empty(), "Display empty for {e:?}");
        }
    }

    #[test]
    fn executor_error_governance_refused_display_names_both_fields() {
        // v0.7.0 (issue #691 fold-1) — pin the user-visible message
        // shape for the PE-1 wire-point refusal so an operator-facing
        // log line carries the command path AND the rule's authored
        // reason (no truncation, no PII rewrite). The chain runner's
        // cascade policy may classify on the prefix "hook spawn refused
        // by governance" — change the wording here in lockstep.
        let err = ExecutorError::GovernanceRefused {
            command: "/opt/homebrew/bin/cargo".into(),
            reason: "R004 cargo forbidden".into(),
        };
        let s = err.to_string();
        assert!(
            s.contains("/opt/homebrew/bin/cargo"),
            "Display must surface the command path: {s}"
        );
        assert!(
            s.contains("R004 cargo forbidden"),
            "Display must surface the rule's reason: {s}"
        );
        assert!(
            s.contains("refused by governance"),
            "Display must carry the governance-refusal marker: {s}"
        );
    }

    #[test]
    fn executor_error_governance_refused_source_is_none() {
        // GovernanceRefused has no underlying io error or cause —
        // `Error::source` must return `None` so the anyhow chain
        // doesn't accidentally show an empty "caused by:" line in
        // operator logs. The Spawn / Io variants are the only ones
        // that carry sources; this pins the negative case so a future
        // refactor that adds a wrapped error here is forced to update
        // the test explicitly.
        let err = ExecutorError::GovernanceRefused {
            command: "/bin/x".into(),
            reason: "denied".into(),
        };
        assert!(std::error::Error::source(&err).is_none());
    }

    #[test]
    fn executor_error_from_io_error_wraps_into_io_variant() {
        let io_err = io::Error::new(io::ErrorKind::PermissionDenied, "nope");
        let err: ExecutorError = io_err.into();
        assert!(matches!(err, ExecutorError::Io(_)));
        let s = err.to_string();
        assert!(s.contains("hook io error"));
        assert!(std::error::Error::source(&err).is_some());
    }

    #[test]
    fn executor_error_spawn_source_chain() {
        let io_err = io::Error::new(io::ErrorKind::NotFound, "no such cmd");
        let err = ExecutorError::Spawn {
            command: "/bin/missing".into(),
            source: io_err,
        };
        assert!(std::error::Error::source(&err).is_some());
        let s = err.to_string();
        assert!(s.contains("/bin/missing"));
        assert!(s.contains("no such cmd"));
    }

    #[test]
    fn executor_error_child_exit_with_signaled_code() {
        let err = ExecutorError::ChildExit {
            code: None,
            stderr: "killed".into(),
        };
        let s = err.to_string();
        assert!(s.contains("<signaled>"));
        // P2 (#628 agent-5) / release/v0.7.0 cbe934c: stderr is REDACTED out
        // of the Display impl to avoid credential-exfiltration via hostile
        // hooks (`printenv >&2; exit 1`). The redacted tail still reaches
        // the operator log via `log_redacted_stderr_at_failure`, but it
        // must NOT appear in the caller-facing reason string.
        assert!(!s.contains("killed"));
        assert!(s.contains("see operator log for redacted stderr"));
    }

    #[test]
    fn executor_error_child_exit_stderr_is_truncated_for_display() {
        let big = "x".repeat(1024);
        let err = ExecutorError::ChildExit {
            code: Some(1),
            stderr: big,
        };
        let s = err.to_string();
        // Display previews at most 256 chars of stderr.
        // The total display is "hook child exited (code 1): " (28) + up to 256 chars.
        assert!(s.len() < 1024);
    }

    #[test]
    fn executor_error_decode_display_carries_reason() {
        let err = ExecutorError::Decode {
            reason: "bad parse".into(),
        };
        let s = err.to_string();
        assert!(s.contains("decode failed"));
        assert!(s.contains("bad parse"));
        assert!(std::error::Error::source(&err).is_none());
    }

    #[test]
    fn executor_error_timeout_display_carries_ms() {
        let err = ExecutorError::Timeout { ms: 5000 };
        let s = err.to_string();
        assert!(s.contains("5000ms"));
        assert!(std::error::Error::source(&err).is_none());
    }

    #[test]
    fn executor_error_daemon_unavailable_carries_attempts() {
        let err = ExecutorError::DaemonUnavailable { attempts: 7 };
        let s = err.to_string();
        assert!(s.contains("7"));
        assert!(s.contains("reconnect attempts"));
        assert!(std::error::Error::source(&err).is_none());
    }

    #[test]
    fn executor_metrics_serialize_to_json() {
        let m = ExecutorMetrics {
            events_fired: 42,
            events_dropped: 1,
            mean_latency_us: 250,
        };
        let json = serde_json::to_string(&m).unwrap();
        assert!(json.contains("\"events_fired\":42"));
        assert!(json.contains("\"events_dropped\":1"));
        assert!(json.contains("\"mean_latency_us\":250"));
    }

    #[test]
    fn metrics_counters_overflow_safe_latency() {
        let m = MetricsCounters::default();
        // Use a huge duration; record_fire clamps to u64::MAX via try_from.
        m.record_fire(Duration::from_micros(u64::MAX));
        m.record_fire(Duration::from_micros(0));
        let snap = m.snapshot();
        assert_eq!(snap.events_fired, 2);
    }

    #[test]
    fn registry_default_is_empty_and_default_eq_new() {
        let reg = ExecutorRegistry::default();
        assert!(reg.is_empty());
        assert_eq!(reg.len(), 0);
    }

    #[test]
    fn parse_decision_line_modify_invalid_delta_wraps_to_decode() {
        let err = parse_decision_line(r#"{"action":"modify","delta":99}"#).unwrap_err();
        assert!(matches!(err, ExecutorError::Decode { .. }));
    }

    #[test]
    fn parse_decision_line_array_payload_wraps_to_decode() {
        let err = parse_decision_line(r#"[1,2,3]"#).unwrap_err();
        match err {
            ExecutorError::Decode { reason } => assert!(reason.contains("object")),
            other => panic!("expected Decode, got {other:?}"),
        }
    }

    // -----------------------------------------------------------------
    // Issue #1207 — spawn-retry-with-backoff pin tests.
    // -----------------------------------------------------------------

    /// EAGAIN / ENOMEM / EMFILE / ETXTBSY are the four errnos the
    /// helper must treat as transient. Other errnos (e.g. ENOENT for
    /// a missing binary) must surface immediately.
    #[cfg(unix)]
    #[test]
    fn issue_1207_is_transient_spawn_errno_classification() {
        // Positive cases.
        for errno in [libc::EAGAIN, libc::ENOMEM, libc::EMFILE, libc::ETXTBSY] {
            let err = io::Error::from_raw_os_error(errno);
            assert!(
                is_transient_spawn_errno(&err),
                "errno {errno} should be transient"
            );
        }
        // Negative cases — ENOENT (missing binary), EACCES (perm denied)
        // must NOT be retried; those are operator-actionable.
        for errno in [libc::ENOENT, libc::EACCES, libc::ENOEXEC] {
            let err = io::Error::from_raw_os_error(errno);
            assert!(
                !is_transient_spawn_errno(&err),
                "errno {errno} must NOT be classified as transient"
            );
        }
        // io::Error without an errno (e.g. synthetic Other) is also
        // non-transient — we can't reason about kernel state.
        assert!(!is_transient_spawn_errno(&io::Error::other("oops")));
    }

    /// The helper returns the first `Ok(child)` without sleeping when
    /// the closure succeeds on the first attempt. Unix-only because
    /// the test spawns `/bin/true` (or `/usr/bin/true`); Windows has
    /// no equivalent always-present zero-exit binary in a stable path,
    /// and the spawn-retry helper itself is a no-op on Windows
    /// (`is_transient_spawn_errno` always returns `false` when
    /// `cfg(unix)` is off).
    #[cfg(unix)]
    #[tokio::test(flavor = "current_thread")]
    async fn issue_1207_spawn_retry_first_attempt_succeeds() {
        let started = Instant::now();
        let child = spawn_with_transient_retry(|| {
            Command::new(if std::path::Path::new("/bin/true").exists() {
                "/bin/true"
            } else {
                "/usr/bin/true"
            })
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .kill_on_drop(true)
            .spawn()
        })
        .await
        .expect("first-attempt spawn should succeed");
        // Reap the child to keep the test hermetic.
        drop(child);
        // No backoff sleeps fired — well under the 10ms first-step
        // ladder entry.
        assert!(
            started.elapsed() < Duration::from_millis(10),
            "first-attempt success must not pay any backoff"
        );
    }

    /// A non-transient errno (ENOENT — missing binary) MUST surface
    /// immediately without retries.
    #[cfg(unix)]
    #[tokio::test(flavor = "current_thread")]
    async fn issue_1207_spawn_retry_non_transient_errno_surfaces_immediately() {
        let started = Instant::now();
        let err = spawn_with_transient_retry(|| {
            Command::new("/nonexistent/path/to/binary-xyz-1207")
                .stdin(Stdio::null())
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .kill_on_drop(true)
                .spawn()
        })
        .await
        .expect_err("spawn of /nonexistent should fail");
        assert_eq!(err.raw_os_error(), Some(libc::ENOENT));
        // Should NOT have paid any backoff sleeps — fail-fast on
        // non-transient errno.
        assert!(
            started.elapsed() < Duration::from_millis(10),
            "non-transient errno must surface immediately, took {:?}",
            started.elapsed()
        );
    }

    /// Fault-injected EAGAIN: the helper sleeps through 4 backoff
    /// steps (~10+50+200+1000 = 1260ms) then succeeds on the 5th
    /// attempt when the injected counter hits zero.
    ///
    /// This test sets the `AI_MEMORY_TEST_FORCE_SPAWN_EAGAIN` env var
    /// to force the helper through the full backoff ladder; it MUST
    /// run in a fresh process (via `cargo test`) because the env-var
    /// gate is `Once`-initialized. We use a `Mutex` to serialize
    /// against other tests in this module that might race.
    #[cfg(unix)]
    #[tokio::test(flavor = "current_thread")]
    async fn issue_1207_spawn_retry_recovers_from_transient_eagain() {
        // This test exercises the retry loop by passing a closure
        // that returns a synthesized EAGAIN on its first two calls,
        // then a real successful spawn. We don't use the env-var
        // injection here because the once-init makes it
        // single-shot per process; a per-call counter under our
        // direct control is the more honest unit test for the loop.
        use std::cell::Cell;
        let attempt_count: Cell<u32> = Cell::new(0);
        let started = Instant::now();
        let child = spawn_with_transient_retry(|| {
            let n = attempt_count.get();
            attempt_count.set(n + 1);
            if n < 2 {
                Err(io::Error::from_raw_os_error(libc::EAGAIN))
            } else {
                Command::new(if std::path::Path::new("/bin/true").exists() {
                    "/bin/true"
                } else {
                    "/usr/bin/true"
                })
                .stdin(Stdio::null())
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .kill_on_drop(true)
                .spawn()
            }
        })
        .await
        .expect("spawn should succeed after 2 EAGAIN retries");
        drop(child);
        assert_eq!(
            attempt_count.get(),
            3,
            "closure called 3 times (2 EAGAIN + 1 success)"
        );
        // Paid 10ms + 50ms = 60ms of backoff. Allow generous slop for
        // CI scheduler jitter (especially under cargo's parallel
        // test load — the EXACT condition this fix targets).
        let elapsed = started.elapsed();
        assert!(
            elapsed >= Duration::from_millis(55),
            "should pay at least 10+50=60ms backoff, got {elapsed:?}"
        );
        assert!(
            elapsed < Duration::from_millis(1_500),
            "should not pay the full 1.26s ladder when success comes on attempt 3, got {elapsed:?}"
        );
    }

    /// Exhausting all 5 attempts surfaces the last transient error.
    /// Guarantees the helper doesn't loop forever AND that operators
    /// still see the precise errno when the kernel never recovers.
    #[cfg(unix)]
    #[tokio::test(flavor = "current_thread")]
    async fn issue_1207_spawn_retry_exhaustion_surfaces_last_error() {
        use std::cell::Cell;
        let attempt_count: Cell<u32> = Cell::new(0);
        let err = spawn_with_transient_retry(|| {
            attempt_count.set(attempt_count.get() + 1);
            Err::<Child, _>(io::Error::from_raw_os_error(libc::EMFILE))
        })
        .await
        .expect_err("all attempts return EMFILE → helper must surface");
        assert_eq!(err.raw_os_error(), Some(libc::EMFILE));
        assert_eq!(
            attempt_count.get(),
            (SPAWN_RETRY_BACKOFF_MS.len() + 1) as u32,
            "closure must be called total_attempts times before exhaustion"
        );
    }
}