//! nash-observe — the observation layer for nash (docs/NASH.md §4–§6).
//!
//! Implements the allow-all [`brush_core::gate::Gate`], the shell-batch event
//! model, near-live batching (200 events / 500 ms / exit), and the transport
//! chain: HTTP over a unix socket (`NUCLEIC_SHELL_SOCKET`), then HTTP over TCP
//! (`NUCLEIC_SHELL_HOOK_URL`), then a JSONL disk spool (`NUCLEIC_SHELL_SPOOL`).
//!
//! Failure doctrine (docs/NASH.md §5.5): nothing in this crate may affect a
//! command's outcome. Hook bodies are panic-caught, the event channel drops
//! (with a `dropped` counter) rather than blocks, and transport is
//! fire-and-forget with short timeouts and spool fallback.

use std::collections::HashMap;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc;
use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use serde::ser::SerializeMap;
use serde::Serialize;

const FLUSH_MAX_EVENTS: usize = 200;
const FLUSH_MAX_AGE: Duration = Duration::from_millis(500);
const CHANNEL_BOUND: usize = 4096;
/// How long a command must have been running before nash announces it as still in flight
/// (docs/NASH.md §5.1). Everything shorter is reported once, on exit: a shell that emitted a
/// start event for every `echo` would double the exec traffic to say "running" about commands
/// that were already over before the batch carrying it was flushed.
const ANNOUNCE_AFTER: Duration = Duration::from_millis(400);
/// How often the flusher re-checks for a command that has crossed `ANNOUNCE_AFTER`. Only ticks
/// while an un-announced command is actually in flight; an idle shell parks as it always did.
const ANNOUNCE_TICK: Duration = Duration::from_millis(250);
const IO_TIMEOUT: Duration = Duration::from_millis(1500);
const EXIT_FLUSH_TIMEOUT: Duration = Duration::from_millis(2000);
/// Per-redirection / per-cmdsub preview cap in bytes (docs/NASH.md §5.4).
const PREVIEW_CAP: usize = 64 * 1024;
/// Captured payload one batch may carry, in raw bytes (docs/NASH.md §5.4's per-batch cap).
/// Beyond it the events keep their counts and hashes and lose their previews — a pipeline storm
/// can otherwise put 200 × 64 KiB of preview in one body, which the host then has to parse,
/// base64-decode and hold (docs/NASH_STREAM_PERF_PLAN.md §P4.2).
const BATCH_PREVIEW_BUDGET: usize = 1024 * 1024;

// ---------------------------------------------------------------------------
// Event model (docs/NASH.md §6.1)
// ---------------------------------------------------------------------------

/// The captured half of a data-flow event (`redirect` / `cmdsub` / `pipe`) — everything the wire
/// calls `bytes`, `truncated`, `hash` and `previewB64`.
///
/// It exists so that **none of that encoding happens where the bytes were captured**
/// (docs/NASH_STREAM_PERF_PLAN.md §P3). The interpreter thread, the tee thread and the command's
/// own exit path hand over raw bytes; redaction, base64 and the FNV hash run inside
/// [`Serialize`] — which is only ever called from the flusher thread, off every path a command
/// waits on. The wire shape is unchanged.
enum Capture {
    /// Nothing was captured: a special file (`/dev/null`, a tty, a socket), or a target that
    /// could not be read. Counts as zero, and carries no hash — an empty hash is how the host
    /// tells "no payload" from "a payload that happened to be empty".
    None,
    /// A file range that has not been read back yet; the flusher opens and reads it
    /// (docs/NASH_STREAM_PERF_PLAN.md §P3.2). `offset`/`total` are fixed at the moment the
    /// command exited, so a later command appending to the same file cannot widen this event's
    /// range — only the *reading* is deferred, never the measurement.
    Deferred {
        path: PathBuf,
        offset: u64,
        total: u64,
    },
    /// Bytes in hand, raw and already capped to [`PREVIEW_CAP`].
    Bytes {
        bytes: u64,
        truncated: bool,
        data: Vec<u8>,
    },
    /// A preview dropped by the per-batch budget (docs/NASH.md §5.4): counts and hash survive,
    /// and `truncated` is true because the preview no longer stands for the bytes.
    Counted { bytes: u64, hash: String },
}

impl Capture {
    /// Bytes in hand, capped to the preview cap.
    fn from_bytes(total: u64, data: &[u8]) -> Self {
        let capped = data.len().min(PREVIEW_CAP);
        Self::Bytes {
            bytes: total,
            truncated: total > capped as u64,
            data: data[..capped].to_vec(),
        }
    }

    /// The payload this will preview, in raw bytes — what the per-batch budget is spent on.
    const fn preview_len(&self) -> usize {
        match self {
            Self::Bytes { data, .. } => data.len(),
            // The budget is spent after the deferred reads are resolved, so this arm is only
            // reached for a range that could not be read; charge its bounded worst case anyway
            // rather than letting an unresolved capture spend nothing.
            Self::Deferred { total, .. } => {
                if *total > PREVIEW_CAP as u64 {
                    PREVIEW_CAP
                } else {
                    *total as usize
                }
            }
            Self::None | Self::Counted { .. } => 0,
        }
    }

    /// Read back a deferred range (flusher thread only). A range that has since become
    /// unreadable degrades to [`Capture::None`] — the event still reports what it observed.
    fn resolve(&mut self) {
        let Self::Deferred {
            path,
            offset,
            total,
        } = self
        else {
            return;
        };
        *self = read_range(path, *offset, *total).map_or(Self::None, |(total, data)| Self::Bytes {
            bytes: total,
            truncated: total > data.len() as u64,
            data,
        });
    }

    /// Drop the preview, keeping the counts and the hash (the per-batch budget).
    fn count_only(&mut self) {
        if let Self::Bytes { bytes, data, .. } = self {
            *self = Self::Counted {
                bytes: *bytes,
                hash: fnv1a_hex(data),
            };
        }
    }
}

impl Serialize for Capture {
    /// Flattened into its event, so the four fields sit where they always have. This is where the
    /// hashing, redaction and base64 of every preview nash sends actually happen.
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        let mut map = serializer.serialize_map(Some(4))?;
        match self {
            // A range left unresolved would mean the flusher skipped it; report it as the
            // metadata-only event it effectively is rather than inventing a payload.
            Self::None | Self::Deferred { .. } => {
                map.serialize_entry("bytes", &0u64)?;
                map.serialize_entry("truncated", &false)?;
                map.serialize_entry("hash", "")?;
                map.serialize_entry("previewB64", "")?;
            }
            Self::Bytes {
                bytes,
                truncated,
                data,
            } => {
                map.serialize_entry("bytes", bytes)?;
                map.serialize_entry("truncated", truncated)?;
                // Over the captured prefix: the full stream is never buffered for a pipe, and this
                // is a did-these-bytes-repeat marker rather than an integrity claim (NASH.md §5.2).
                map.serialize_entry("hash", &fnv1a_hex(data))?;
                map.serialize_entry("previewB64", &preview_b64(data))?;
            }
            Self::Counted { bytes, hash } => {
                map.serialize_entry("bytes", bytes)?;
                map.serialize_entry("truncated", &true)?;
                map.serialize_entry("hash", hash)?;
                map.serialize_entry("previewB64", "")?;
            }
        }
        map.end()
    }
}

#[derive(Serialize)]
#[serde(tag = "kind", rename_all = "camelCase")]
enum Event {
    /// A command that has been running longer than [`ANNOUNCE_AFTER`] and has not returned
    /// (docs/NASH.md §5.1). Emitted at most once per command, from the flusher's sweep rather
    /// than the gate, so a short command costs no extra event. `token` is the gate token the
    /// matching [`Event::Exec`] carries, which is what lets a host retire the in-flight row
    /// instead of showing the command twice.
    #[serde(rename = "exec-start", rename_all = "camelCase")]
    ExecStart {
        seq: u64,
        ts: u64,
        token: u64,
        /// How long the command had *already* been running when this went out. A host that showed
        /// elapsed time from the moment it heard about the command would under-report it by the
        /// announce threshold plus however long the batch sat in the flusher; this is what lets
        /// the row count from when the command actually started.
        elapsed_ms: u64,
        argv: Vec<String>,
        cwd: String,
        #[serde(skip_serializing_if = "Option::is_none")]
        pipeline: Option<PipelineRef>,
    },
    #[serde(rename = "exec", rename_all = "camelCase")]
    Exec {
        seq: u64,
        ts: u64,
        /// The gate token this command ran under — the join key for the `exec-start` that may
        /// have announced it (see [`Event::ExecStart`]). Always present; a command that returned
        /// before `ANNOUNCE_AFTER` simply has no start event to join.
        token: u64,
        argv: Vec<String>,
        cwd: String,
        exit_code: i32,
        duration_ms: u64,
        /// Present only for a command running as a stage of `a | b | c`; omitted
        /// otherwise so a standalone command's event keeps its old shape.
        #[serde(skip_serializing_if = "Option::is_none")]
        pipeline: Option<PipelineRef>,
    },
    #[serde(rename = "cd", rename_all = "camelCase")]
    Cd { seq: u64, ts: u64, from: String, to: String },
    #[serde(rename = "export", rename_all = "camelCase")]
    Export {
        seq: u64,
        ts: u64,
        op: String,
        names: Vec<String>,
    },
    #[serde(rename = "fallback", rename_all = "camelCase")]
    Fallback {
        seq: u64,
        ts: u64,
        reason: String,
        input: String,
    },
    #[serde(rename = "redirect", rename_all = "camelCase")]
    Redirect {
        seq: u64,
        ts: u64,
        cmd_seq: u64,
        op: String,
        fd: u32,
        target: Option<String>,
        /// Usually [`Capture::Deferred`] when it goes into the channel and [`Capture::Bytes`] by
        /// the time it is serialized — the file read happens on the flusher (§P3.2).
        #[serde(flatten)]
        capture: Capture,
    },
    #[serde(rename = "cmdsub", rename_all = "camelCase")]
    Cmdsub {
        seq: u64,
        ts: u64,
        #[serde(flatten)]
        capture: Capture,
    },
    #[serde(rename = "pipe", rename_all = "camelCase")]
    Pipe {
        seq: u64,
        ts: u64,
        pipeline_id: u64,
        from_index: u64,
        to_index: u64,
        /// The commands either side of this link, as written — brush-core renders them from the
        /// AST at spawn, so they name a compound stage correctly and arrive with the very first
        /// link rather than waiting on the stages to exit.
        ///
        /// Stage text is source, so it can carry a literal credential the same way argv can;
        /// masking runs at serialization time like every other outbound string (NASH.md §5.4).
        #[serde(serialize_with = "serialize_redacted")]
        from_text: String,
        #[serde(serialize_with = "serialize_redacted")]
        to_text: String,
        #[serde(flatten)]
        capture: Capture,
    },
    #[serde(rename = "dropped", rename_all = "camelCase")]
    Dropped { seq: u64, ts: u64, count: u64 },
    /// An operator lever nash was asked to act on (docs/NASH.md §4.3). `action` is
    /// `ignored` for a lever that arrived through the environment — the channel the
    /// observed process controls, which nash refuses — and `honored` for one the
    /// trusted policy file set. Either way it becomes a feed row, so a shell going
    /// dark is a visible event rather than an absence of events.
    #[serde(rename = "policy", rename_all = "camelCase")]
    Policy {
        seq: u64,
        ts: u64,
        lever: String,
        value: String,
        action: String,
        note: String,
    },
}

/// The pipeline stage an exec event belongs to (docs/NASH.md §5.1). `id` is the same
/// counter the `pipe` events carry, so the host can name the commands either side of
/// a link instead of showing bare stage indices.
#[derive(Serialize, Clone, Copy)]
#[serde(rename_all = "camelCase")]
struct PipelineRef {
    id: u64,
    index: u64,
    len: u64,
}

impl From<brush_core::gate::PipelineSlot> for PipelineRef {
    fn from(slot: brush_core::gate::PipelineSlot) -> Self {
        Self {
            id: slot.id,
            index: slot.index as u64,
            len: slot.len as u64,
        }
    }
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct Batch<'a> {
    #[serde(rename = "type")]
    batch_type: &'static str,
    source: &'static str,
    session_id: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    environment_kind: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    environment_id: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    environment_label: Option<&'a str>,
    shell_id: &'a str,
    parent_shell: Option<&'a str>,
    events: &'a [Event],
}

// ---------------------------------------------------------------------------
// Config
// ---------------------------------------------------------------------------

struct Config {
    socket: Option<PathBuf>,
    hook_url: Option<HookUrl>,
    token: Option<String>,
    session_id: Option<String>,
    environment_kind: Option<String>,
    environment_id: Option<String>,
    environment_label: Option<String>,
    spool: Option<PathBuf>,
}

struct HookUrl {
    host: String,
    port: u16,
    path: String,
}

fn parse_hook_url(url: &str) -> Option<HookUrl> {
    let rest = url.strip_prefix("http://")?;
    let (authority, path) = match rest.find('/') {
        Some(idx) => (&rest[..idx], rest[idx..].to_string()),
        None => (rest, "/shell-event".to_string()),
    };
    let (host, port) = match authority.rsplit_once(':') {
        Some((h, p)) => (h.to_string(), p.parse().ok()?),
        None => (authority.to_string(), 80),
    };
    Some(HookUrl { host, port, path })
}

impl Config {
    /// Read transport config from the environment.
    ///
    /// `require_observation` comes from the trusted policy file (docs/NASH.md §4.3) and
    /// closes the quietest bypass there was: `NUCLEIC_SHELL_CAPTURE=off`, or simply
    /// `env -u NUCLEIC_SHELL_HOOK_URL …`, used to leave nash running as an ordinary
    /// shell that reported nothing — no fallback event, no trace, from a process that
    /// merely rewrote its own environment. Under the policy, `off` from the environment
    /// is refused and a missing transport falls back to `fallback_spool`, which the
    /// host's spool drain already collects. An operator who genuinely wants a surface
    /// silent turns the policy's `require_observation` off (or omits the file) rather
    /// than setting a variable the agent could have set for them.
    fn from_env(require_observation: bool, fallback_spool: &Path) -> Option<Self> {
        if std::env::var("NUCLEIC_SHELL_CAPTURE").as_deref() == Ok("off") && !require_observation {
            return None;
        }
        let socket = std::env::var_os("NUCLEIC_SHELL_SOCKET").map(PathBuf::from);
        let hook_url = std::env::var("NUCLEIC_SHELL_HOOK_URL")
            .ok()
            .and_then(|u| parse_hook_url(&u));
        let mut spool = std::env::var_os("NUCLEIC_SHELL_SPOOL").map(PathBuf::from);
        if socket.is_none() && hook_url.is_none() && spool.is_none() {
            if !require_observation {
                return None;
            }
            spool = Some(fallback_spool.to_path_buf());
        }
        Some(Self {
            socket,
            hook_url,
            token: std::env::var("NUCLEIC_HOOK_TOKEN").ok(),
            session_id: std::env::var("NUCLEIC_SESSION_ID").ok(),
            environment_kind: std::env::var("NUCLEIC_SHELL_ENVIRONMENT_KIND").ok(),
            environment_id: std::env::var("NUCLEIC_SHELL_ENVIRONMENT_ID").ok(),
            environment_label: std::env::var("NUCLEIC_SHELL_ENVIRONMENT_LABEL").ok(),
            spool,
        })
    }
}

// ---------------------------------------------------------------------------
// Observer state
// ---------------------------------------------------------------------------

enum Msg {
    Event(Event),
    FlushSync(mpsc::SyncSender<()>),
    /// A command just started, so the flusher must re-arm its timeout for the in-flight sweep
    /// (§5.1). Without it the thread stays parked on its idle wait — a shell whose only activity
    /// is one long command sends nothing to wake it, which is exactly the case being announced.
    /// Carries nothing and is never batched: a nudge, not an event.
    Wake,
}

struct PendingExec {
    argv: Vec<String>,
    cwd: PathBuf,
    started: Instant,
    redirects: Vec<brush_core::gate::RedirectRecord>,
    pipeline: Option<brush_core::gate::PipelineSlot>,
    /// Whether an [`Event::ExecStart`] has already gone out for this command. The sweep sets it
    /// so a command that outlives many ticks is announced exactly once.
    announced: bool,
}

struct Observer {
    tx: mpsc::SyncSender<Msg>,
    next_token: AtomicU64,
    next_seq: AtomicU64,
    dropped: AtomicU64,
    pending: Mutex<HashMap<u64, PendingExec>>,
}

static OBSERVER: OnceLock<Observer> = OnceLock::new();

impl Observer {
    fn seq(&self) -> u64 {
        self.next_seq.fetch_add(1, Ordering::Relaxed)
    }

    fn send(&self, event: Event) {
        match self.tx.try_send(Msg::Event(event)) {
            Ok(()) => {
                // Piggyback a dropped-counter event whenever capacity returns.
                let dropped = self.dropped.swap(0, Ordering::Relaxed);
                if dropped > 0 {
                    let _ = self.tx.try_send(Msg::Event(Event::Dropped {
                        seq: self.seq(),
                        ts: now_millis(),
                        count: dropped,
                    }));
                }
            }
            Err(_) => {
                self.dropped.fetch_add(1, Ordering::Relaxed);
            }
        }
    }
}

fn now_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| u64::try_from(d.as_millis()).unwrap_or(0))
        .unwrap_or(0)
}

// --- data-flow capture helpers (docs/NASH.md §5.2–5.4) ---------------------

const B64: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";

/// Standard base64 (no external crate — the transport is a hand-rolled HTTP client).
fn b64(input: &[u8]) -> String {
    let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
    for chunk in input.chunks(3) {
        let b = [
            chunk[0],
            *chunk.get(1).unwrap_or(&0),
            *chunk.get(2).unwrap_or(&0),
        ];
        let n = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
        out.push(B64[((n >> 18) & 63) as usize] as char);
        out.push(B64[((n >> 12) & 63) as usize] as char);
        out.push(if chunk.len() > 1 { B64[((n >> 6) & 63) as usize] as char } else { '=' });
        out.push(if chunk.len() > 2 { B64[(n & 63) as usize] as char } else { '=' });
    }
    out
}

/// 64-bit FNV-1a, hex — a cheap content fingerprint for the full (uncapped) data.
fn fnv1a_hex(input: &[u8]) -> String {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for &byte in input {
        h ^= u64::from(byte);
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    format!("{h:016x}")
}

/// Redact obvious credential shapes from a preview before it leaves the guest
/// (docs/NASH.md §5.4). Best-effort pattern masking on whitespace-delimited tokens.
fn redact(text: &str) -> String {
    text.split_inclusive(|c: char| c.is_whitespace())
        .map(|tok| {
            let (word, trailer) = match tok.char_indices().rev().find(|(_, c)| !c.is_whitespace()) {
                Some((i, c)) => tok.split_at(i + c.len_utf8()),
                None => return tok.to_string(),
            };
            if looks_secret(word) {
                format!("«redacted»{trailer}")
            } else {
                tok.to_string()
            }
        })
        .collect()
}

fn looks_secret(word: &str) -> bool {
    // `…` is trimmed alongside the quoting because brush-core appends it when it caps a pipeline
    // stage label (docs/NASH.md §5.3). Glued onto the final token with no separator, it otherwise
    // failed the charset test below and walked a truncated credential straight past the entropy
    // branch — a secret that happened to straddle the cap left the guest in the clear.
    let w = word.trim_matches(|c: char| c == '"' || c == '\'' || c == ',' || c == '…');
    // Common token prefixes (GitHub PATs, Slack, Stripe, AWS, OpenAI/Anthropic, …).
    const PREFIXES: [&str; 10] =
        ["ghp_", "gho_", "ghs_", "github_pat_", "xoxb-", "xoxp-", "sk-", "AKIA", "ASIA", "AIza"];
    if PREFIXES.iter().any(|p| w.starts_with(p)) && w.len() >= 12 {
        return true;
    }
    // Long high-entropy-ish blobs (base64/hex secrets).
    if w.len() >= 32
        && w.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '/' | '=' | '_' | '-'))
        && w.chars().any(|c| c.is_ascii_digit())
        && w.chars().any(|c| c.is_ascii_alphabetic())
    {
        return true;
    }
    false
}

/// What one redirect record captured, measured on the command's own path but **not yet read**
/// (docs/NASH_STREAM_PERF_PLAN.md §P3.2).
///
/// The split is deliberate. The *range* — is this a regular file at all, and which bytes did this
/// command put there — is only true at the moment the command exited: `echo a > f; echo b >> f`
/// would otherwise have the truncating write report both lines, because by the time a batch
/// flushes the file has grown. So the exit path pays one `stat` and nothing else; opening the
/// file, reading up to 64 KiB of it, hashing, redacting and base64-encoding all move to the
/// flusher, which is where the bytes were always going anyway.
fn measure_readback(record: &brush_core::gate::RedirectRecord) -> Capture {
    use brush_core::gate::RedirectReadback;
    if let Some(inline) = &record.inline {
        // Heredoc / here-string: the body was known at setup, so there is nothing to read.
        let full = inline.as_bytes();
        return Capture::from_bytes(full.len() as u64, full);
    }
    let Some(path) = record.path.as_ref() else {
        return Capture::None;
    };
    // Only read back regular files; skip /dev/null, ttys, fifos, sockets, devices.
    let Ok(meta) = std::fs::metadata(path) else {
        return Capture::None;
    };
    if !meta.is_file() {
        return Capture::None;
    }
    let size = meta.len();
    let (offset, total) = match record.readback {
        RedirectReadback::Append => (record.size_before, size.saturating_sub(record.size_before)),
        RedirectReadback::Truncate | RedirectReadback::Input => (0, size),
        RedirectReadback::Inline => return Capture::None,
    };
    Capture::Deferred {
        path: path.clone(),
        offset,
        total,
    }
}

/// Read a measured range back, capped to [`PREVIEW_CAP`]. Returns the range's full length (which
/// may exceed what was read) and the bytes. Runs on the flusher thread only.
fn read_range(path: &Path, offset: u64, total: u64) -> Option<(u64, Vec<u8>)> {
    let mut file = std::fs::File::open(path).ok()?;
    if offset > 0 {
        use std::io::Seek;
        file.seek(std::io::SeekFrom::Start(offset)).ok()?;
    }
    let want = usize::try_from(total.min(PREVIEW_CAP as u64)).unwrap_or(0);
    let mut buf = vec![0u8; want];
    let n = std::io::Read::read(&mut file, &mut buf).unwrap_or(0);
    buf.truncate(n);
    Some((total, buf))
}

/// Resolve a redirection target against the directory its command ran in, so the
/// reported path is meaningful to a host that never saw that cwd. Purely lexical
/// (`.`/`..` collapsed, no symlink resolution, no filesystem access): the target of a
/// truncating write may not exist yet, and observation must never stat its way into a
/// slow or hanging path.
fn absolute_path(path: &Path, cwd: &Path) -> PathBuf {
    let joined = if path.is_absolute() {
        path.to_path_buf()
    } else {
        cwd.join(path)
    };
    let mut out = PathBuf::new();
    for component in joined.components() {
        match component {
            std::path::Component::CurDir => {}
            std::path::Component::ParentDir => {
                out.pop();
            }
            other => out.push(other.as_os_str()),
        }
    }
    out
}

/// Build a preview string (redacted, base64) from captured bytes. Binary data is
/// previewed as a hex head rather than mangled UTF-8.
fn preview_b64(data: &[u8]) -> String {
    let looks_text = std::str::from_utf8(data).is_ok();
    if looks_text {
        // SAFETY-adjacent: validated above.
        let redacted = redact(&String::from_utf8_lossy(data));
        b64(redacted.as_bytes())
    } else {
        let mut hex = String::with_capacity(1024 * 2 + 9);
        hex.push_str("<binary> ");
        for byte in data.iter().take(1024) {
            hex.push(HEX[usize::from(byte >> 4)] as char);
            hex.push(HEX[usize::from(byte & 0xf)] as char);
        }
        b64(hex.as_bytes())
    }
}

const HEX: &[u8; 16] = b"0123456789abcdef";

/// Mask a string on its way out (docs/NASH.md §5.4). Used where redaction is deferred to
/// serialization — the flusher thread — rather than done where the string was captured.
fn serialize_redacted<S: serde::Serializer>(text: &str, serializer: S) -> Result<S::Ok, S::Error> {
    serializer.serialize_str(&redact(text))
}

// ---------------------------------------------------------------------------
// Gate implementation (allow-all; docs/NASH.md §4.2)
// ---------------------------------------------------------------------------

struct RecordingGate;

impl brush_core::gate::Gate for RecordingGate {
    fn on_exec(&self, ev: brush_core::gate::ExecStart) -> brush_core::gate::Verdict {
        let token = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let obs = OBSERVER.get()?;
            let token = obs.next_token.fetch_add(1, Ordering::Relaxed);
            if let Ok(mut pending) = obs.pending.lock() {
                pending.insert(
                    token,
                    PendingExec {
                        argv: ev.argv,
                        cwd: ev.cwd,
                        started: Instant::now(),
                        redirects: ev.redirects,
                        pipeline: ev.pipeline,
                        announced: false,
                    },
                );
            }
            // Best-effort by design: a dropped nudge (full channel) costs at most a late
            // announcement, and a full channel means events are flowing anyway — which wakes the
            // flusher just the same.
            let _ = obs.tx.try_send(Msg::Wake);
            Some(token)
        }))
        .ok()
        .flatten()
        .unwrap_or(0);
        brush_core::gate::Verdict::Allow(token)
    }

    fn on_exit(&self, token: u64, ev: brush_core::gate::ExecEnd) {
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let Some(obs) = OBSERVER.get() else { return };
            let Some(pending) = obs.pending.lock().ok().and_then(|mut p| p.remove(&token))
            else {
                return;
            };
            let ts = now_millis();
            let cwd_str = pending.cwd.to_string_lossy().into_owned();

            // Derived, names-only state-change events (docs/NASH.md §5.3–5.4).
            match pending.argv.first().map(String::as_str) {
                Some("cd") if ev.exit_code == 0 => {
                    let to = pending
                        .argv
                        .get(1)
                        .map_or_else(|| "~".to_string(), Clone::clone);
                    obs.send(Event::Cd {
                        seq: obs.seq(),
                        ts,
                        from: cwd_str.clone(),
                        to,
                    });
                }
                Some(op @ ("export" | "declare" | "typeset" | "unset" | "readonly" | "local"))
                    if ev.exit_code == 0 =>
                {
                    let names: Vec<String> = pending
                        .argv
                        .iter()
                        .skip(1)
                        .filter(|a| !a.starts_with('-'))
                        .map(|a| a.split('=').next().unwrap_or(a).to_string())
                        .collect();
                    if !names.is_empty() {
                        obs.send(Event::Export {
                            seq: obs.seq(),
                            ts,
                            op: op.to_string(),
                            names,
                        });
                    }
                }
                _ => {}
            }

            let exec_seq = obs.seq();
            obs.send(Event::Exec {
                seq: exec_seq,
                ts,
                token,
                argv: pending.argv,
                cwd: cwd_str,
                exit_code: ev.exit_code,
                duration_ms: u64::try_from(pending.started.elapsed().as_millis()).unwrap_or(0),
                pipeline: pending.pipeline.map(PipelineRef::from),
            });

            // Data-flow read-back for this command's redirects (docs/NASH.md §5.2). Only the
            // *range* is settled here (one stat); the read and the encoding ride to the flusher.
            for record in &pending.redirects {
                // The host resolves a redirect target against the session's worktree to decide
                // which file was edited (and which lock that is), so report the path absolutely:
                // `> out.txt` is only meaningful next to the directory the command ran in, and the
                // redirect event — unlike `exec` — carries no cwd of its own.
                let target = record
                    .path
                    .as_ref()
                    .map(|p| absolute_path(p, &pending.cwd).to_string_lossy().into_owned());
                obs.send(Event::Redirect {
                    seq: obs.seq(),
                    ts,
                    cmd_seq: exec_seq,
                    op: record.op.clone(),
                    fd: record.fd,
                    target,
                    capture: measure_readback(record),
                });
            }
        }));
    }

    fn on_cmdsub(&self, output: &str) {
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let Some(obs) = OBSERVER.get() else { return };
            let full = output.as_bytes();
            // Copies the capped prefix and nothing else: this runs on the interpreter thread, in
            // the middle of the expansion the shell is waiting on, and `$(cat big-file)` is a
            // substitution whose *whole* result used to be hashed here (§P3.1).
            obs.send(Event::Cmdsub {
                seq: obs.seq(),
                ts: now_millis(),
                capture: Capture::from_bytes(full.len() as u64, full),
            });
        }));
    }

    fn on_pipe(&self, ev: brush_core::gate::PipeEvent) {
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let Some(obs) = OBSERVER.get() else { return };
            let capped = ev.captured.len().min(PREVIEW_CAP);
            let mut data = ev.captured;
            data.truncate(capped);
            obs.send(Event::Pipe {
                seq: obs.seq(),
                ts: now_millis(),
                pipeline_id: ev.pipeline_id,
                from_index: ev.from_index as u64,
                to_index: ev.to_index as u64,
                from_text: ev.from_text,
                to_text: ev.to_text,
                capture: Capture::Bytes {
                    bytes: ev.total_bytes,
                    truncated: ev.truncated || ev.total_bytes > capped as u64,
                    data,
                },
            });
        }));
    }
}

// ---------------------------------------------------------------------------
// Transport
// ---------------------------------------------------------------------------

/// Set once the host rejects our token; we stop posting and spool instead.
static UNAUTHORIZED: AtomicBool = AtomicBool::new(false);

fn http_request(path: &str, token: Option<&str>, body: &[u8]) -> Vec<u8> {
    // Keep-alive, deliberately (docs/NASH_STREAM_PERF_PLAN.md §P4.1): a shell posts a batch every
    // 500 ms for as long as it lives, and `Connection: close` made every one of them a fresh
    // connect + accept + teardown on both sides. The host's route already answers with
    // `Connection: keep-alive` and a `Content-Length`, which is what makes reuse framable.
    let mut req = format!(
        "POST {path} HTTP/1.1\r\nHost: nucleic\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: keep-alive\r\n",
        body.len()
    );
    if let Some(token) = token {
        req.push_str(&format!("Authorization: Bearer {token}\r\n"));
    }
    req.push_str("\r\n");
    let mut bytes = req.into_bytes();
    bytes.extend_from_slice(body);
    bytes
}

/// The outcome of one post on a live connection.
enum Posted {
    /// The host took the batch and the connection can carry another.
    Ok,
    /// The host took the batch but the connection cannot be reused (no `Content-Length` to frame
    /// the next response by, or the host asked to close).
    OkAndClose,
    /// The batch was not delivered; the connection is dropped. `retry` is false when a fresh
    /// connection would not help (an outright refusal), true for an I/O failure — most often a
    /// kept-open socket the host has since closed, which is nobody's fault and worth one retry.
    Failed { retry: bool },
}

/// Anything the transport can post over: the unix socket and the TCP fallback both qualify.
trait Wire: Read + Write {}
impl<T: Read + Write> Wire for T {}

/// A connection kept open across batches (docs/NASH_STREAM_PERF_PLAN.md §P4.1). Lives on the
/// flusher thread, which is the only thing that posts, so it needs no locking.
#[derive(Default)]
struct Connection {
    /// The live connection and the request path the transport it came from wants.
    open: Option<(Box<dyn Wire>, String)>,
}

impl Connection {
    /// Post one batch, reconnecting once if the kept-open connection turned out to be dead.
    fn post(&mut self, cfg: &Config, body: &[u8]) -> Result<(), ()> {
        for _ in 0..2 {
            if self.open.is_none() {
                self.open = connect(cfg);
            }
            let Some((stream, path)) = self.open.as_mut() else {
                return Err(()); // nothing to connect to; the caller spools
            };
            let request = http_request(path, cfg.token.as_deref(), body);
            match post_on(stream.as_mut(), &request) {
                Posted::Ok => return Ok(()),
                Posted::OkAndClose => {
                    self.open = None;
                    return Ok(());
                }
                Posted::Failed { retry } => {
                    self.open = None;
                    if !retry {
                        return Err(());
                    }
                }
            }
        }
        Err(())
    }
}

/// Open the configured transport: the unix socket first, then the TCP hook.
fn connect(cfg: &Config) -> Option<(Box<dyn Wire>, String)> {
    if let Some(socket) = &cfg.socket {
        if let Ok(stream) = std::os::unix::net::UnixStream::connect(socket) {
            let _ = stream.set_write_timeout(Some(IO_TIMEOUT));
            let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
            return Some((Box::new(stream), "/shell-event".to_string()));
        }
    }
    let url = cfg.hook_url.as_ref()?;
    let addr = std::net::ToSocketAddrs::to_socket_addrs(&(url.host.as_str(), url.port))
        .ok()?
        .next()?;
    let stream = std::net::TcpStream::connect_timeout(&addr, IO_TIMEOUT).ok()?;
    let _ = stream.set_write_timeout(Some(IO_TIMEOUT));
    let _ = stream.set_read_timeout(Some(IO_TIMEOUT));
    // Batches are one write each and latency is not what this path optimizes, but a delayed ACK
    // waiting on Nagle would hold the *response* — and with it the next batch — for milliseconds.
    let _ = stream.set_nodelay(true);
    Some((Box::new(stream), url.path.clone()))
}

/// Write one request and read its whole response, so the connection is left framed for the next.
fn post_on(stream: &mut dyn Wire, request: &[u8]) -> Posted {
    if stream.write_all(request).is_err() {
        return Posted::Failed { retry: true };
    }
    let Some(response) = read_response(stream) else {
        return Posted::Failed { retry: true };
    };
    match response.status {
        401 => {
            UNAUTHORIZED.store(true, Ordering::Relaxed);
            Posted::Failed { retry: false }
        }
        code if (200..300).contains(&code) => {
            if response.reusable {
                Posted::Ok
            } else {
                Posted::OkAndClose
            }
        }
        // The host 202s known tokens, so anything else is not retryable.
        _ => Posted::Failed { retry: false },
    }
}

struct Response {
    status: u16,
    /// Whether the body was fully consumed and the peer means to keep the connection.
    reusable: bool,
}

/// Read a response's head, then drain exactly its `Content-Length` body — the whole point being
/// that the next response starts where this one ends. A response we cannot frame (no length,
/// chunked, `Connection: close`) is still reported, but its connection is not reused.
fn read_response(stream: &mut dyn Read) -> Option<Response> {
    let mut buf = Vec::with_capacity(256);
    let mut chunk = [0u8; 512];
    let head_end = loop {
        if let Some(at) = find_headers_end(&buf) {
            break at;
        }
        // A response head this long is not one of ours; give up rather than read forever.
        if buf.len() > 8192 {
            return None;
        }
        match stream.read(&mut chunk) {
            Ok(0) => return None,
            Ok(n) => buf.extend_from_slice(&chunk[..n]),
            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => (),
            Err(_) => return None,
        }
    };
    let head = String::from_utf8_lossy(&buf[..head_end]);
    let mut lines = head.split("\r\n");
    let status: u16 = lines.next()?.split_whitespace().nth(1)?.parse().ok()?;
    let mut length: Option<usize> = None;
    let mut close = false;
    for line in lines {
        let Some((name, value)) = line.split_once(':') else {
            continue;
        };
        match name.trim().to_ascii_lowercase().as_str() {
            "content-length" => length = value.trim().parse().ok(),
            "connection" => close = value.trim().eq_ignore_ascii_case("close"),
            _ => (),
        }
    }
    let Some(length) = length else {
        return Some(Response {
            status,
            reusable: false,
        });
    };
    // Drain the body so the socket is positioned at the next response.
    let mut remaining = length.saturating_sub(buf.len() - (head_end + 4));
    while remaining > 0 {
        let want = remaining.min(chunk.len());
        match stream.read(&mut chunk[..want]) {
            Ok(0) => {
                return Some(Response {
                    status,
                    reusable: false,
                })
            }
            Ok(n) => remaining -= n,
            Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => (),
            Err(_) => {
                return Some(Response {
                    status,
                    reusable: false,
                })
            }
        }
    }
    Some(Response {
        status,
        reusable: !close,
    })
}

/// Offset of the `\r\n\r\n` that ends a response head.
fn find_headers_end(buf: &[u8]) -> Option<usize> {
    buf.windows(4).position(|w| w == b"\r\n\r\n")
}

fn spool(dir: &Path, shell_id: &str, body: &[u8]) {
    let _ = std::fs::create_dir_all(dir);
    let path = dir.join(format!("{shell_id}.jsonl"));
    if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(path) {
        let _ = f.write_all(body);
        let _ = f.write_all(b"\n");
    }
}

fn deliver(cfg: &Config, shell_id: &str, conn: &mut Connection, body: &[u8]) {
    if !UNAUTHORIZED.load(Ordering::Relaxed) && conn.post(cfg, body).is_ok() {
        return;
    }
    if let Some(dir) = &cfg.spool {
        spool(dir, shell_id, body);
    }
}

// ---------------------------------------------------------------------------
// Flusher thread
// ---------------------------------------------------------------------------

/// What one sweep of the pending map found: the commands to announce as still running, and
/// whether anything is still too young to announce (which is what keeps the flusher ticking).
struct Sweep {
    announce: Vec<Event>,
    waiting: bool,
}

/// Announce every command that has been in flight longer than [`ANNOUNCE_AFTER`] without
/// returning (docs/NASH.md §5.1).
///
/// Runs on the flusher thread, never on a command's own path, so a slow or contended sweep can
/// only delay an event — never a command. Events are returned rather than sent through the
/// channel the flusher is itself draining: pushing them straight into the outgoing buffer keeps
/// a start event ahead of the exit event it belongs to (a command whose `on_exit` has already
/// removed it from the map is no longer here to announce).
fn sweep_running() -> Sweep {
    let mut sweep = Sweep { announce: Vec::new(), waiting: false };
    let Some(obs) = OBSERVER.get() else { return sweep };
    let Ok(mut pending) = obs.pending.lock() else { return sweep };
    // Oldest first, so a batch that announces several commands at once still numbers them in the
    // order they started (a `HashMap` would otherwise hand them over in an arbitrary one).
    let mut ripe: Vec<u64> = pending
        .iter()
        .filter(|(_, exec)| !exec.announced)
        .filter_map(|(token, exec)| {
            if exec.started.elapsed() >= ANNOUNCE_AFTER {
                Some(*token)
            } else {
                sweep.waiting = true;
                None
            }
        })
        .collect();
    ripe.sort_unstable();
    for token in ripe {
        let Some(exec) = pending.get_mut(&token) else { continue };
        exec.announced = true;
        sweep.announce.push(Event::ExecStart {
            seq: obs.seq(),
            ts: now_millis(),
            token,
            elapsed_ms: u64::try_from(exec.started.elapsed().as_millis()).unwrap_or(0),
            argv: exec.argv.clone(),
            cwd: exec.cwd.to_string_lossy().into_owned(),
            pipeline: exec.pipeline.map(PipelineRef::from),
        });
    }
    sweep
}

/// The captured half of an event, for the flusher's two passes over a batch.
fn event_capture(event: &mut Event) -> Option<&mut Capture> {
    match event {
        Event::Redirect { capture, .. }
        | Event::Cmdsub { capture, .. }
        | Event::Pipe { capture, .. } => Some(capture),
        _ => None,
    }
}

/// Settle a batch's payloads before it is serialized — the flusher-thread half of §P3/§P4.2.
///
/// Two passes, in this order: read back every deferred file range, then spend the per-batch
/// preview budget over what that leaves. Events keep their place and their counts either way; only
/// previews are ever given up, oldest kept first, so a storm of huge pipes loses its tail rather
/// than the whole batch losing its head.
fn settle(events: &mut [Event]) {
    let mut spent = 0usize;
    for event in events {
        let Some(capture) = event_capture(event) else {
            continue;
        };
        capture.resolve();
        spent = spent.saturating_add(capture.preview_len());
        if spent > BATCH_PREVIEW_BUDGET {
            capture.count_only();
        }
    }
}

fn flusher(cfg: Config, shell_id: String, parent_shell: Option<String>, rx: mpsc::Receiver<Msg>) {
    let mut buf: Vec<Event> = Vec::new();
    let mut first_at: Option<Instant> = None;
    let mut conn = Connection::default();

    let flush = |buf: &mut Vec<Event>, conn: &mut Connection| {
        if buf.is_empty() {
            return;
        }
        settle(buf);
        let batch = Batch {
            batch_type: "shell-batch",
            source: "nash",
            session_id: cfg.session_id.as_deref(),
            environment_kind: cfg.environment_kind.as_deref(),
            environment_id: cfg.environment_id.as_deref(),
            environment_label: cfg.environment_label.as_deref(),
            shell_id: &shell_id,
            parent_shell: parent_shell.as_deref(),
            events: buf,
        };
        if let Ok(body) = serde_json::to_vec(&batch) {
            deliver(&cfg, &shell_id, conn, &body);
        }
        buf.clear();
    };

    loop {
        // Commands that have outlived `ANNOUNCE_AFTER` go out with (or ahead of) whatever else
        // is buffered, so a host learns about a long build while it is still building.
        let sweep = std::panic::catch_unwind(sweep_running)
            .unwrap_or(Sweep { announce: Vec::new(), waiting: false });
        if !sweep.announce.is_empty() {
            if buf.is_empty() {
                first_at = Some(Instant::now());
            }
            buf.extend(sweep.announce);
            if buf.len() >= FLUSH_MAX_EVENTS {
                flush(&mut buf, &mut conn);
                first_at = None;
            }
        }

        let flush_due = match first_at {
            Some(t) => FLUSH_MAX_AGE.saturating_sub(t.elapsed()),
            None => Duration::from_secs(3600),
        };
        // Tick only while something is in flight and not yet announced. Once every running
        // command has been announced — and whenever none is — the thread parks as it always did.
        // The tick shortens the *wait*, never the flush: waking early to re-sweep must not turn
        // the locked 500 ms batching cadence into a 250 ms one for every shell running a command.
        let ticking = sweep.waiting && ANNOUNCE_TICK < flush_due;
        match rx.recv_timeout(if ticking { ANNOUNCE_TICK } else { flush_due }) {
            Ok(Msg::Event(ev)) => {
                if buf.is_empty() {
                    first_at = Some(Instant::now());
                }
                buf.push(ev);
                if buf.len() >= FLUSH_MAX_EVENTS {
                    flush(&mut buf, &mut conn);
                    first_at = None;
                }
            }
            Ok(Msg::FlushSync(ack)) => {
                flush(&mut buf, &mut conn);
                first_at = None;
                let _ = ack.send(());
            }
            // Nothing to do but go round again: the sweep at the top of the loop is the work, and
            // the timeout it computes is the point of having been woken.
            Ok(Msg::Wake) => {}
            Err(mpsc::RecvTimeoutError::Timeout) => {
                // A tick is just "go re-sweep"; the batch is not due yet.
                if !ticking {
                    flush(&mut buf, &mut conn);
                    first_at = None;
                }
            }
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                flush(&mut buf, &mut conn);
                return;
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Public API
// ---------------------------------------------------------------------------

/// Reads `NUCLEIC_SHELL_*` config from the environment; when observation is
/// configured, installs the recording gate into brush-core and starts the
/// flusher thread. Returns whether observation is active.
///
/// `require_observation` / `fallback_spool` come from the trusted policy file
/// (docs/NASH.md §4.3): with it set, an environment that says "don't observe"
/// no longer wins, because that environment belongs to the observed process.
///
/// Also threads shell lineage: reads `NUCLEIC_SHELL_PARENT` as this shell's
/// parent and re-exports it as this shell's own id for child shells.
pub fn install_from_env(require_observation: bool, fallback_spool: &Path) -> bool {
    let Some(cfg) = Config::from_env(require_observation, fallback_spool) else {
        return false;
    };

    let shell_id = format!("{:x}-{:x}", std::process::id(), now_millis());
    let parent_shell = std::env::var("NUCLEIC_SHELL_PARENT").ok();
    // SAFETY: called from `main` before any other threads exist.
    unsafe { std::env::set_var("NUCLEIC_SHELL_PARENT", &shell_id) };

    let (tx, rx) = mpsc::sync_channel(CHANNEL_BOUND);
    let observer = Observer {
        tx,
        next_token: AtomicU64::new(1),
        next_seq: AtomicU64::new(1),
        dropped: AtomicU64::new(0),
        pending: Mutex::new(HashMap::new()),
    };
    if OBSERVER.set(observer).is_err() {
        return false;
    }

    std::thread::Builder::new()
        .name("nash-observe-flush".into())
        .spawn(move || flusher(cfg, shell_id, parent_shell, rx))
        .ok();

    brush_core::gate::set_gate(Box::new(RecordingGate));
    true
}

/// Flushes any buffered events, waiting up to a short deadline. Intended for
/// the process-exit hook; a no-op when observation is not installed.
pub fn flush_sync() {
    let Some(obs) = OBSERVER.get() else { return };
    let (ack_tx, ack_rx) = mpsc::sync_channel(1);
    if obs.tx.try_send(Msg::FlushSync(ack_tx)).is_ok() {
        let _ = ack_rx.recv_timeout(EXIT_FLUSH_TIMEOUT);
    }
}

/// Records an operator-lever event (docs/NASH.md §4.3): a lever nash refused to take
/// from the environment (`action: "ignored"`) or one the trusted policy set
/// (`action: "honored"`). Not flushed here — callers that are about to `exec` away
/// call [`flush_sync`] themselves, and the ordinary path flushes at exit.
pub fn report_policy(lever: &str, value: &str, action: &str, note: &str) {
    let Some(obs) = OBSERVER.get() else { return };
    obs.send(Event::Policy {
        seq: obs.seq(),
        ts: now_millis(),
        lever: lever.to_string(),
        // Bounded like any other agent-authored string on this path: the value is
        // whatever the command line put there.
        value: value.chars().take(512).collect(),
        action: action.to_string(),
        note: note.to_string(),
    });
}

/// Records a compat-fallback event (docs/NASH.md §4.1): nash is about to
/// re-exec the real bash because it could not handle `input`.
pub fn report_fallback(reason: &str, input: &str) {
    let Some(obs) = OBSERVER.get() else { return };
    let truncated: String = input.chars().take(4096).collect();
    obs.send(Event::Fallback {
        seq: obs.seq(),
        ts: now_millis(),
        reason: reason.to_string(),
        input: truncated,
    });
    flush_sync();
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;

    /// Serialize one event the way the flusher does, as JSON.
    fn json(event: &Event) -> serde_json::Value {
        serde_json::to_value(event).expect("event serializes")
    }

    fn pipe(bytes: u64, data: Vec<u8>) -> Event {
        Event::Pipe {
            seq: 1,
            ts: 1,
            pipeline_id: 1,
            from_index: 0,
            to_index: 1,
            from_text: "a".to_string(),
            to_text: "b".to_string(),
            capture: Capture::Bytes {
                bytes,
                truncated: bytes > data.len() as u64,
                data,
            },
        }
    }

    /// The four payload fields sit where they always have, and an empty hash still means "nothing
    /// was captured" rather than "the hash of nothing" — the host reads it that way.
    #[test]
    fn capture_serializes_the_wire_shape() {
        let event = json(&pipe(5, b"hello".to_vec()));
        assert_eq!(event["kind"], "pipe");
        assert_eq!(event["bytes"], 5);
        assert_eq!(event["truncated"], false);
        assert_eq!(event["previewB64"], b64(b"hello"));
        assert_eq!(event["hash"], fnv1a_hex(b"hello"));

        let empty = json(&Event::Cmdsub {
            seq: 1,
            ts: 1,
            capture: Capture::None,
        });
        assert_eq!(empty["bytes"], 0);
        assert_eq!(empty["hash"], "");
        assert_eq!(empty["previewB64"], "");
    }

    /// Redaction and base64 run at serialization time now, so they must still actually run —
    /// on the preview *and* on the stage labels a link carries.
    #[test]
    fn serialization_redacts_previews_and_stage_text() {
        let secret = "ghp_0123456789abcdefghij";
        let event = json(&Event::Pipe {
            seq: 1,
            ts: 1,
            pipeline_id: 1,
            from_index: 0,
            to_index: 1,
            from_text: format!("curl -H {secret}"),
            to_text: "sh".to_string(),
            capture: Capture::from_bytes(secret.len() as u64, secret.as_bytes()),
        });
        assert!(!event["fromText"].as_str().unwrap().contains(secret));
        assert_eq!(event["fromText"], "curl -H «redacted»");
        assert_eq!(event["previewB64"], b64("«redacted»".as_bytes()));
    }

    /// A capture is capped to the preview cap where it is taken, and says so.
    #[test]
    fn capture_caps_and_marks_truncation() {
        let big = vec![b'x'; PREVIEW_CAP + 10];
        let Capture::Bytes {
            bytes,
            truncated,
            data,
        } = Capture::from_bytes(big.len() as u64, &big)
        else {
            panic!("expected captured bytes");
        };
        assert_eq!(bytes, big.len() as u64);
        assert_eq!(data.len(), PREVIEW_CAP);
        assert!(truncated);
    }

    /// The per-batch budget (docs/NASH.md §5.4): once a batch has carried its allowance of
    /// payload, later events keep their counts and hashes and give up their previews.
    #[test]
    fn batch_preview_budget_drops_previews_not_events() {
        let payload = vec![b'z'; PREVIEW_CAP];
        let over = BATCH_PREVIEW_BUDGET / PREVIEW_CAP + 2;
        let mut events: Vec<Event> = (0..over)
            .map(|_| pipe(PREVIEW_CAP as u64, payload.clone()))
            .collect();
        settle(&mut events);

        let rendered: Vec<serde_json::Value> = events.iter().map(json).collect();
        // Every event survives — only previews are given up, and the earliest keep theirs.
        assert_eq!(rendered.len(), over);
        assert_ne!(rendered[0]["previewB64"], "");
        let last = rendered.last().unwrap();
        assert_eq!(last["previewB64"], "");
        assert_eq!(last["bytes"], PREVIEW_CAP);
        assert_eq!(last["hash"], fnv1a_hex(&payload));
        assert_eq!(last["truncated"], true);
        let kept = rendered
            .iter()
            .filter(|e| e["previewB64"] != "")
            .count();
        assert_eq!(kept, BATCH_PREVIEW_BUDGET / PREVIEW_CAP);
    }

    /// A redirect's bytes are read by the flusher, from the range measured when the command
    /// exited — an append that lands afterwards must not widen it.
    #[test]
    fn deferred_readback_reads_the_measured_range() {
        let dir = std::env::temp_dir().join(format!("nash-observe-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("out.txt");
        std::fs::write(&path, b"first\n").unwrap();

        let mut event = Event::Redirect {
            seq: 1,
            ts: 1,
            cmd_seq: 1,
            op: ">".to_string(),
            fd: 1,
            target: Some(path.to_string_lossy().into_owned()),
            capture: Capture::Deferred {
                path: path.clone(),
                offset: 0,
                total: 6,
            },
        };
        // The file grows before the batch flushes; the event still reports what its command wrote.
        std::fs::write(&path, b"first\nsecond\n").unwrap();
        settle(std::slice::from_mut(&mut event));

        let rendered = json(&event);
        assert_eq!(rendered["bytes"], 6);
        assert_eq!(rendered["previewB64"], b64(b"first\n"));
        assert_eq!(rendered["truncated"], false);

        // A target that has since vanished degrades to metadata only rather than failing.
        std::fs::remove_file(&path).unwrap();
        let mut gone = Event::Redirect {
            seq: 2,
            ts: 1,
            cmd_seq: 1,
            op: ">".to_string(),
            fd: 1,
            target: None,
            capture: Capture::Deferred {
                path,
                offset: 0,
                total: 6,
            },
        };
        settle(std::slice::from_mut(&mut gone));
        assert_eq!(json(&gone)["bytes"], 0);
        assert_eq!(json(&gone)["hash"], "");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Binary payloads are previewed as a hex head (now table-driven, §P3.3) — same bytes as the
    /// `format!`-per-byte encoding it replaced.
    #[test]
    fn binary_previews_as_hex() {
        let preview = preview_b64(&[0xff, 0x00, 0x0a]);
        assert_eq!(preview, b64(b"<binary> ff000a"));
    }
}
