harn_vm/testbench/

tape.rs

//! Unified event tape for the testbench.
//!
//! A tape is the canonical artifact behind `harn test-bench --emit-tape`.
//! Every non-deterministic input the script consumed — clock advances,
//! LLM responses, FS reads/writes, subprocess spawns — lands as a typed
//! [`TapeRecord`] with a logical sequence number, execution phase, and a
//! virtual-time stamp. The tape is what the [`fidelity`] oracle compares;
//! it is what `harn test-bench replay` reads to drive a deterministic
//! re-run.
//!
//! [`fidelity`]: super::fidelity
//!
//! ## File layout
//!
//! ```text
//! tape.tape       # NDJSON: one header line + one record line per event
//! tape.cas/       # content-addressed sidecar (BLAKE3 hex names)
//! ```
//!
//! Small payloads are serialized inline. Anything over [`MAX_INLINE_BYTES`]
//! lands in `tape.cas/<blake3>` and the record carries `{"cas": "<blake3>"}`.
//! That keeps the main stream diffable when the only thing that changes
//! is a multi-MB LLM response.
//!
//! ## Versioning
//!
//! Every tape carries a `version` integer in its header. The current
//! schema is [`TAPE_FORMAT_VERSION`]. Loaders accept anything `<=` the
//! current version and emit a structured error for newer tapes; this
//! gives us room to add record kinds (under `#[serde(other)]`) without
//! silently breaking older runners.
//!
//! ## Recording
//!
//! Recording is opt-in: the testbench installs a thread-local
//! [`TapeRecorder`] when `Testbench::tape = TapeConfig::Emit { path }`.
//! Every host-capability axis that already has a record path
//! ([`super::process_tape`], [`super::overlay_fs`], [`crate::llm::mock`],
//! [`crate::clock_mock`]) calls into this module to push a record. When
//! no recorder is installed, the helpers are no-ops — production code
//! pays nothing.

use std::cell::RefCell;
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
use std::sync::{Arc, Mutex};

use serde::{Deserialize, Serialize};

use crate::clock_mock;

/// Format version of the on-disk tape representation. Bump on any
/// breaking change. Loaders refuse tapes with a higher version.
pub const TAPE_FORMAT_VERSION: u32 = 1;

/// Records whose serialized payload exceeds this size are spilled into
/// the content-addressed sidecar. Picked to be larger than typical
/// stdout/file-read sizes but smaller than full LLM responses, so the
/// main NDJSON stream stays diffable.
pub const MAX_INLINE_BYTES: usize = 4 * 1024;

/// Header line written first in every tape file. Captures the metadata a
/// fidelity-checker needs to interpret the records that follow.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TapeHeader {
    pub version: u32,
    /// Crate version of the producer (`harn-vm` `CARGO_PKG_VERSION`).
    /// Surfaced so a fidelity report can attribute divergences across
    /// runtime upgrades.
    pub harn_version: String,
    /// UNIX-epoch milliseconds the script was launched at — i.e. the
    /// initial value of the testbench's paused clock. `null` when the
    /// run used the real clock.
    #[serde(default)]
    pub started_at_unix_ms: Option<i64>,
    /// Path passed to `harn test-bench run`. Informational only; replays
    /// resolve scripts via the CLI argument, not this field.
    #[serde(default)]
    pub script_path: Option<String>,
    /// Positional arguments forwarded to the script (post `--`). Captured
    /// so two re-runs that differ only in argv are distinguishable.
    #[serde(default)]
    pub argv: Vec<String>,
}

impl TapeHeader {
    pub fn current(
        started_at_unix_ms: Option<i64>,
        script_path: Option<String>,
        argv: Vec<String>,
    ) -> Self {
        Self {
            version: TAPE_FORMAT_VERSION,
            harn_version: env!("CARGO_PKG_VERSION").to_string(),
            started_at_unix_ms,
            script_path,
            argv,
        }
    }
}

/// One on-disk line of the tape. Wrapping the header and record kinds
/// behind a single tagged enum lets us write the whole file as
/// homogeneous NDJSON.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum TapeLine {
    Header(TapeHeader),
    Record(TapeRecord),
}

/// One captured non-deterministic event. The variant carries the record
/// payload; the wrapping [`TapeRecord`] adds the metadata every variant
/// shares.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TapeRecord {
    /// Monotonic logical sequence number assigned at record time.
    pub seq: u64,
    /// Execution phase that produced the record. The fidelity oracle
    /// uses this to keep script-visible boundaries strict while letting
    /// runtime finalization evolve without regenerating user fixtures.
    #[serde(default)]
    pub phase: TapePhase,
    /// Wall-clock value (UNIX-epoch ms) observed at record time. Reads
    /// from the unified mock clock when one is installed.
    pub virtual_time_ms: i64,
    /// Monotonic ms since the testbench was activated. Independent of
    /// `virtual_time_ms` so a paused clock that never advances still
    /// produces an ordered stream.
    pub monotonic_ms: i64,
    /// The actual event.
    pub kind: TapeRecordKind,
}

/// Coarse execution phase for host-boundary tape records.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TapePhase {
    /// Records produced while evaluating the user-authored script body.
    #[default]
    UserScript,
    /// Records produced while the runtime drains finish/resume/finalizer
    /// lifecycle work after the script body has yielded its result.
    RuntimeFinalize,
}

impl TapePhase {
    fn as_u8(self) -> u8 {
        match self {
            Self::UserScript => 0,
            Self::RuntimeFinalize => 1,
        }
    }

    fn from_u8(value: u8) -> Self {
        match value {
            1 => Self::RuntimeFinalize,
            _ => Self::UserScript,
        }
    }

    pub fn label(self) -> &'static str {
        match self {
            Self::UserScript => "user_script",
            Self::RuntimeFinalize => "runtime_finalize",
        }
    }
}

/// Discriminated union of every record kind the v1 tape captures. New
/// kinds can be added without breaking older readers (`serde(other)`
/// support is intentional — unknown variants surface as
/// [`TapeRecordKind::Unknown`] so a fidelity check still runs).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TapeRecordKind {
    /// Script read the wall-clock or monotonic clock. The captured value
    /// is what the script actually saw, so a re-run that drifts (e.g.
    /// because the operator forgot `--clock paused`) produces a
    /// different content hash and the fidelity oracle flags it.
    ClockRead { source: ClockSource, value_ms: i64 },
    /// Script slept (or otherwise advanced the unified mock clock) by
    /// `duration_ms`. The recorded virtual time is post-advance.
    ClockSleep { duration_ms: u64 },
    /// LLM call. `request_digest` is a deterministic hash of the call's
    /// matchable surface (messages + system + tools + tool_choice +
    /// thinking). `response` is the recorded mock — inline JSON for
    /// small payloads, a CAS reference for large ones.
    LlmCall {
        request_digest: String,
        response: TapePayload,
    },
    /// Filesystem read against the testbench overlay. The content hash
    /// lets fidelity checks reason about read consistency without
    /// inlining every byte.
    FileRead {
        path: String,
        content_hash: String,
        len_bytes: u64,
    },
    /// Filesystem write into the testbench overlay.
    FileWrite {
        path: String,
        content_hash: String,
        len_bytes: u64,
    },
    /// Filesystem delete in the overlay layer.
    FileDelete { path: String },
    /// Subprocess spawn captured by [`super::process_tape`]. Stdout and
    /// stderr are stored under `stdout_payload`/`stderr_payload` so the
    /// large blobs land in CAS rather than the NDJSON line.
    ProcessSpawn {
        program: String,
        args: Vec<String>,
        cwd: Option<String>,
        exit_code: i32,
        duration_ms: u64,
        stdout_payload: TapePayload,
        stderr_payload: TapePayload,
    },
    /// MCP JSON-RPC exchange observed by Harn's MCP client. The request
    /// and response payloads are redacted before they are written so
    /// cassettes and unified tapes share the same privacy boundary.
    McpJsonRpc {
        server: String,
        method: String,
        request_digest: String,
        response_digest: String,
        latency_ms: u64,
        request_payload: TapePayload,
        response_payload: TapePayload,
    },
    /// Asynchronous model-job lifecycle event (`harn.model_job_event.v1`)
    /// observed through the host event bridge. Asset digests named in the
    /// event payload let fidelity checks reason about media lineage without
    /// calling a live model.
    ModelJob {
        job_id: String,
        request_id: String,
        backend: String,
        state: String,
        event_kind: String,
        event: TapePayload,
        asset_digests: Vec<String>,
    },
    /// Catch-all for record kinds emitted by a newer producer. Lets
    /// older fidelity checkers compare what they understand and flag
    /// the rest as `Unknown` divergence rather than refusing to load.
    #[serde(other)]
    Unknown,
}

impl TapeRecordKind {
    /// Stable, snake_case label for this kind. Mirrors the `kind` tag
    /// `serde` writes to disk so display-side code (CLI summaries,
    /// report headers, error messages) is consistent with the wire
    /// format without re-deriving the string each call site.
    pub fn label(&self) -> &'static str {
        match self {
            Self::ClockRead { .. } => "clock_read",
            Self::ClockSleep { .. } => "clock_sleep",
            Self::LlmCall { .. } => "llm_call",
            Self::FileRead { .. } => "file_read",
            Self::FileWrite { .. } => "file_write",
            Self::FileDelete { .. } => "file_delete",
            Self::ProcessSpawn { .. } => "process_spawn",
            Self::McpJsonRpc { .. } => "mcp_json_rpc",
            Self::ModelJob { .. } => "model_job",
            Self::Unknown => "unknown",
        }
    }
}

/// Which face of the unified clock the script read. Captured so a
/// fidelity report can attribute drift back to the right axis.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ClockSource {
    Wall,
    Monotonic,
}

/// On-disk representation of a record payload. Inline for small values,
/// CAS-by-hash for anything over [`MAX_INLINE_BYTES`].
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum TapePayload {
    /// Inline UTF-8 text payload. Carries a content hash so a fidelity
    /// check can compare without re-hashing.
    Inline { content_hash: String, text: String },
    /// CAS-stored payload. The bytes live at `<tape>.cas/<content_hash>`.
    Cas {
        content_hash: String,
        len_bytes: u64,
    },
}

impl TapePayload {
    pub fn content_hash(&self) -> &str {
        match self {
            Self::Inline { content_hash, .. } | Self::Cas { content_hash, .. } => content_hash,
        }
    }

    pub fn len_bytes(&self) -> u64 {
        match self {
            Self::Inline { text, .. } => text.len() as u64,
            Self::Cas { len_bytes, .. } => *len_bytes,
        }
    }
}

/// Compute a stable BLAKE3 hex digest for a byte slice. Centralized so
/// every record path keys CAS lookups identically.
pub fn content_hash(bytes: &[u8]) -> String {
    blake3::hash(bytes).to_hex().to_string()
}

/// Build a [`TapePayload`] from raw bytes, spilling to the sidecar map
/// when the body is large enough to clutter the NDJSON.
fn build_payload(bytes: Vec<u8>, cas: &mut BTreeMap<String, Vec<u8>>) -> TapePayload {
    let hash = content_hash(&bytes);
    if bytes.len() > MAX_INLINE_BYTES {
        let len_bytes = bytes.len() as u64;
        cas.entry(hash.clone()).or_insert(bytes);
        TapePayload::Cas {
            content_hash: hash,
            len_bytes,
        }
    } else {
        let text = match String::from_utf8(bytes) {
            Ok(text) => text,
            Err(error) => {
                // Non-utf8 bytes still need to round-trip. Stash the raw
                // bytes in CAS and inline a sentinel so a fidelity diff
                // is still meaningful.
                let bytes = error.into_bytes();
                let len_bytes = bytes.len() as u64;
                cas.entry(hash.clone()).or_insert(bytes);
                return TapePayload::Cas {
                    content_hash: hash,
                    len_bytes,
                };
            }
        };
        TapePayload::Inline {
            content_hash: hash,
            text,
        }
    }
}

/// In-memory tape: header + ordered record list + sidecar bytes pending
/// flush to disk. Built by [`TapeRecorder`] during a record run; loaded
/// by [`EventTape::load`] for replay or fidelity comparison.
#[derive(Debug, Clone)]
pub struct EventTape {
    pub header: TapeHeader,
    pub records: Vec<TapeRecord>,
    /// Content-addressed bodies. Populated either by the recorder (in
    /// memory until [`EventTape::persist`]) or by [`EventTape::load`]
    /// (read back from `<tape>.cas/`).
    cas: BTreeMap<String, Vec<u8>>,
}

impl EventTape {
    pub fn new(header: TapeHeader) -> Self {
        Self {
            header,
            records: Vec::new(),
            cas: BTreeMap::new(),
        }
    }

    /// Resolve a payload to its full bytes. Inline payloads return their
    /// text; CAS payloads look up the sidecar.
    pub fn resolve_payload(&self, payload: &TapePayload) -> Result<Vec<u8>, String> {
        match payload {
            TapePayload::Inline { text, .. } => Ok(text.as_bytes().to_vec()),
            TapePayload::Cas { content_hash, .. } => self
                .cas
                .get(content_hash)
                .cloned()
                .ok_or_else(|| format!("tape CAS missing entry for {content_hash}")),
        }
    }

    /// Total CAS payload count. Useful for diagnostics and tests.
    pub fn cas_len(&self) -> usize {
        self.cas.len()
    }

    /// Persist the tape (NDJSON + sidecar) to `path`. The sidecar lives
    /// at `<path>.cas/`; the parent directory is created if needed.
    pub fn persist(&self, path: &Path) -> Result<(), String> {
        if let Some(parent) = path.parent() {
            if !parent.as_os_str().is_empty() {
                std::fs::create_dir_all(parent)
                    .map_err(|err| format!("mkdir {}: {err}", parent.display()))?;
            }
        }

        let mut body = String::new();
        let header_line = serde_json::to_string(&TapeLine::Header(self.header.clone()))
            .map_err(|err| format!("serialize tape header: {err}"))?;
        body.push_str(&header_line);
        body.push('\n');
        for record in &self.records {
            let line = serde_json::to_string(&TapeLine::Record(record.clone()))
                .map_err(|err| format!("serialize tape record: {err}"))?;
            body.push_str(&line);
            body.push('\n');
        }
        std::fs::write(path, body).map_err(|err| format!("write {}: {err}", path.display()))?;

        if !self.cas.is_empty() {
            let cas_dir = cas_dir_for(path);
            std::fs::create_dir_all(&cas_dir)
                .map_err(|err| format!("mkdir {}: {err}", cas_dir.display()))?;
            for (hash, bytes) in &self.cas {
                let entry = cas_dir.join(hash);
                std::fs::write(&entry, bytes)
                    .map_err(|err| format!("write {}: {err}", entry.display()))?;
            }
        }
        Ok(())
    }

    /// Load a tape from `path`. Reads the NDJSON body and lazily fetches
    /// any referenced CAS entries from `<path>.cas/`.
    pub fn load(path: &Path) -> Result<Self, String> {
        let body = std::fs::read_to_string(path)
            .map_err(|err| format!("read {}: {err}", path.display()))?;
        let mut lines = body.lines();
        let first_line = lines
            .next()
            .ok_or_else(|| format!("empty tape file: {}", path.display()))?;
        let header_line: TapeLine = serde_json::from_str(first_line)
            .map_err(|err| format!("parse tape header in {}: {err}", path.display()))?;
        let header = match header_line {
            TapeLine::Header(header) => header,
            TapeLine::Record(_) => {
                return Err(format!(
                    "tape {} is missing its header (first line is a record)",
                    path.display()
                ))
            }
        };
        if header.version > TAPE_FORMAT_VERSION {
            return Err(format!(
                "tape {} declares version {} but this runtime supports up to {TAPE_FORMAT_VERSION}",
                path.display(),
                header.version
            ));
        }
        let mut records = Vec::new();
        for (idx, line) in lines.enumerate() {
            let trimmed = line.trim();
            if trimmed.is_empty() {
                continue;
            }
            let parsed: TapeLine = serde_json::from_str(trimmed).map_err(|err| {
                format!(
                    "parse tape record at line {} in {}: {err}",
                    idx + 2,
                    path.display()
                )
            })?;
            match parsed {
                TapeLine::Record(record) => records.push(record),
                TapeLine::Header(_) => {
                    return Err(format!(
                        "tape {} contains a second header at line {}",
                        path.display(),
                        idx + 2
                    ))
                }
            }
        }

        let mut cas = BTreeMap::new();
        let cas_dir = cas_dir_for(path);
        if cas_dir.is_dir() {
            for record in &records {
                visit_payloads(&record.kind, |payload| {
                    if let TapePayload::Cas { content_hash, .. } = payload {
                        if cas.contains_key(content_hash) {
                            return;
                        }
                        let entry = cas_dir.join(content_hash);
                        if let Ok(bytes) = std::fs::read(&entry) {
                            cas.insert(content_hash.clone(), bytes);
                        }
                    }
                });
            }
        }
        Ok(Self {
            header,
            records,
            cas,
        })
    }
}

fn cas_dir_for(tape_path: &Path) -> PathBuf {
    let mut buf = tape_path.as_os_str().to_owned();
    buf.push(".cas");
    PathBuf::from(buf)
}

fn visit_payloads(kind: &TapeRecordKind, mut visit: impl FnMut(&TapePayload)) {
    match kind {
        TapeRecordKind::LlmCall { response, .. } => visit(response),
        TapeRecordKind::ProcessSpawn {
            stdout_payload,
            stderr_payload,
            ..
        } => {
            visit(stdout_payload);
            visit(stderr_payload);
        }
        TapeRecordKind::McpJsonRpc {
            request_payload,
            response_payload,
            ..
        } => {
            visit(request_payload);
            visit(response_payload);
        }
        TapeRecordKind::ModelJob { event, .. } => visit(event),
        TapeRecordKind::ClockRead { .. }
        | TapeRecordKind::ClockSleep { .. }
        | TapeRecordKind::FileRead { .. }
        | TapeRecordKind::FileWrite { .. }
        | TapeRecordKind::FileDelete { .. }
        | TapeRecordKind::Unknown => {}
    }
}

/// Recorder consulted by every host-capability axis. When installed as
/// the [`active_recorder`], each axis's record path also pushes a
/// [`TapeRecord`] here so the unified tape stays in sync without
/// re-routing every capability through this module.
#[derive(Debug)]
pub struct TapeRecorder {
    next_seq: AtomicU64,
    phase: AtomicU8,
    started_at: clock_mock::ClockInstant,
    inner: Mutex<RecorderInner>,
}

#[derive(Debug, Default)]
struct RecorderInner {
    records: Vec<TapeRecord>,
    cas: BTreeMap<String, Vec<u8>>,
}

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

impl TapeRecorder {
    pub fn new() -> Self {
        Self {
            next_seq: AtomicU64::new(0),
            phase: AtomicU8::new(TapePhase::UserScript.as_u8()),
            started_at: clock_mock::instant_now(),
            inner: Mutex::new(RecorderInner::default()),
        }
    }

    /// Append a record built from `kind`. The recorder stamps the seq
    /// number and timing metadata; callers only worry about the payload.
    pub fn record(&self, kind: TapeRecordKind) {
        let virtual_time_ms = clock_mock::now_ms();
        let monotonic_ms = clock_mock::instant_now()
            .duration_since(self.started_at)
            .as_millis()
            .min(i64::MAX as u128) as i64;
        self.record_at(kind, virtual_time_ms, monotonic_ms);
    }

    fn record_at(&self, kind: TapeRecordKind, virtual_time_ms: i64, monotonic_ms: i64) {
        let seq = self.next_seq.fetch_add(1, Ordering::SeqCst);
        let record = TapeRecord {
            seq,
            phase: TapePhase::from_u8(self.phase.load(Ordering::SeqCst)),
            virtual_time_ms,
            monotonic_ms,
            kind,
        };
        self.inner
            .lock()
            .expect("tape recorder mutex poisoned")
            .records
            .push(record);
    }

    fn swap_phase(&self, phase: TapePhase) -> TapePhase {
        TapePhase::from_u8(self.phase.swap(phase.as_u8(), Ordering::SeqCst))
    }

    /// Convenience wrapper: build a [`TapePayload`] from `bytes` (spilling
    /// to CAS as needed) and register the bytes for persistence. Used by
    /// axes that have raw bodies on hand (subprocess stdout, LLM
    /// response JSON, file content).
    pub fn payload_from_bytes(&self, bytes: Vec<u8>) -> TapePayload {
        let mut inner = self.inner.lock().expect("tape recorder mutex poisoned");
        build_payload(bytes, &mut inner.cas)
    }

    /// Snapshot the tape into a self-contained [`EventTape`]. Consumes
    /// the recorder's CAS by `clone()` so a recorder can be sampled
    /// mid-run for diagnostics — production callers usually move into
    /// `into_tape` instead.
    pub fn snapshot(&self, header: TapeHeader) -> EventTape {
        let inner = self.inner.lock().expect("tape recorder mutex poisoned");
        EventTape {
            header,
            records: inner.records.clone(),
            cas: inner.cas.clone(),
        }
    }
}

thread_local! {
    static ACTIVE_RECORDER: RefCell<Option<Arc<TapeRecorder>>> = const { RefCell::new(None) };
}

/// RAII guard returned by [`install_recorder`]. Restores the previous
/// recorder (if any) on drop so nested testbench sessions stay sane.
pub struct TapeRecorderGuard {
    previous: Option<Arc<TapeRecorder>>,
}

impl Drop for TapeRecorderGuard {
    fn drop(&mut self) {
        let prev = self.previous.take();
        ACTIVE_RECORDER.with(|slot| {
            *slot.borrow_mut() = prev;
        });
    }
}

pub fn install_recorder(recorder: Arc<TapeRecorder>) -> TapeRecorderGuard {
    let previous = ACTIVE_RECORDER.with(|slot| slot.replace(Some(recorder)));
    TapeRecorderGuard { previous }
}

/// Currently installed recorder, if any. Production callers stay
/// untouched because nothing installs a recorder outside testbench mode.
pub fn active_recorder() -> Option<Arc<TapeRecorder>> {
    ACTIVE_RECORDER.with(|slot| slot.borrow().clone())
}

/// Record a model-job lifecycle event in the active unified tape.
///
/// Called from the host event bridge so fake, replay, and live backends share
/// one cassette shape. Missing recorders are a no-op.
pub fn record_model_job_event(payload: &serde_json::Value) {
    let Some(recorder) = active_recorder() else {
        return;
    };
    let policy = crate::redact::current_policy();
    let redacted = policy.redact_json(payload);
    let event_bytes = serde_json::to_vec(&redacted).unwrap_or_default();
    let event = recorder.payload_from_bytes(event_bytes);
    let job_id = redacted
        .get("job_id")
        .and_then(|value| value.as_str())
        .unwrap_or("")
        .to_string();
    let request_id = redacted
        .get("request_id")
        .and_then(|value| value.as_str())
        .unwrap_or("")
        .to_string();
    let backend = redacted
        .get("backend")
        .and_then(|value| value.as_str())
        .unwrap_or("")
        .to_string();
    let state = redacted
        .get("state")
        .and_then(|value| value.as_str())
        .unwrap_or("")
        .to_string();
    let event_kind = redacted
        .get("kind")
        .and_then(|value| value.as_str())
        .unwrap_or("")
        .to_string();
    let mut asset_digests = Vec::new();
    if let Some(assets) = redacted.get("assets").and_then(|value| value.as_array()) {
        for asset in assets {
            if let Some(digest) = asset
                .get("sha256")
                .and_then(|value| value.as_str())
                .filter(|value| !value.is_empty())
            {
                asset_digests.push(digest.to_string());
            } else if let Some(uri) = asset.get("uri").and_then(|value| value.as_str()) {
                if let Some(digest) = uri.strip_prefix("asset://sha256/") {
                    if !digest.is_empty() {
                        asset_digests.push(digest.to_string());
                    }
                }
            }
        }
    }
    if let Some(digest) = redacted
        .get("asset_digest")
        .and_then(|value| value.as_str())
        .filter(|value| !value.is_empty())
    {
        asset_digests.push(digest.to_string());
    }
    asset_digests.sort();
    asset_digests.dedup();
    recorder.record(TapeRecordKind::ModelJob {
        job_id,
        request_id,
        backend,
        state,
        event_kind,
        event,
        asset_digests,
    });
}

/// Record an MCP JSON-RPC exchange in the active unified tape, if one
/// is installed. Payloads are redacted here so every caller gets the
/// same privacy behavior as MCP cassettes.
pub fn record_mcp_json_rpc(
    server: &str,
    method: &str,
    request: &serde_json::Value,
    response: &serde_json::Value,
    latency_ms: u64,
) {
    let Some(recorder) = active_recorder() else {
        return;
    };
    let policy = crate::redact::current_policy();
    let request = policy.redact_json(request);
    let response = policy.redact_json(response);
    let request_bytes = serde_json::to_vec(&request).unwrap_or_default();
    let response_bytes = serde_json::to_vec(&response).unwrap_or_default();
    let request_digest = content_hash(&request_bytes);
    let response_digest = content_hash(&response_bytes);
    let request_payload = recorder.payload_from_bytes(request_bytes);
    let response_payload = recorder.payload_from_bytes(response_bytes);
    recorder.record(TapeRecordKind::McpJsonRpc {
        server: server.to_string(),
        method: method.to_string(),
        request_digest,
        response_digest,
        latency_ms,
        request_payload,
        response_payload,
    });
}

/// RAII guard that temporarily changes the phase stamped onto records
/// from the active recorder.
pub struct TapePhaseGuard {
    recorder: Arc<TapeRecorder>,
    previous: TapePhase,
}

impl Drop for TapePhaseGuard {
    fn drop(&mut self) {
        self.recorder.swap_phase(self.previous);
    }
}

/// Enter `phase` for subsequent records emitted by the active recorder.
/// Returns `None` when tape recording is off.
pub fn enter_phase(phase: TapePhase) -> Option<TapePhaseGuard> {
    let recorder = active_recorder()?;
    let previous = recorder.swap_phase(phase);
    Some(TapePhaseGuard { recorder, previous })
}

/// Push a record if a recorder is active. The closure is only evaluated
/// when recording is on, so the per-axis hooks pay nothing in production.
pub fn with_active_recorder<F>(build: F)
where
    F: FnOnce(&Arc<TapeRecorder>) -> Option<TapeRecordKind>,
{
    let Some(recorder) = active_recorder() else {
        return;
    };
    if let Some(kind) = build(&recorder) {
        recorder.record(kind);
    }
}

/// Push a record stamped by an exact capability-owned clock.
///
/// The unified testbench clock remains the default for ambient runtime axes,
/// while nominal Harness clocks use this path so an attenuated test clock does
/// not leak into unrelated VMs or lose fidelity metadata.
pub fn with_active_recorder_clock<F>(clock: &dyn harn_clock::Clock, build: F)
where
    F: FnOnce(&Arc<TapeRecorder>) -> Option<TapeRecordKind>,
{
    let Some(recorder) = active_recorder() else {
        return;
    };
    if let Some(kind) = build(&recorder) {
        recorder.record_at(kind, harn_clock::now_wall_ms(clock), clock.monotonic_ms());
    }
}

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

    fn small_record(seq: u64, dur: u64) -> TapeRecord {
        TapeRecord {
            seq,
            phase: TapePhase::UserScript,
            virtual_time_ms: seq as i64 * 1000,
            monotonic_ms: seq as i64 * 1000,
            kind: TapeRecordKind::ClockSleep { duration_ms: dur },
        }
    }

    #[test]
    fn round_trip_inline_records() {
        let temp = TempDir::new().unwrap();
        let path = temp.path().join("run.tape");
        let mut tape = EventTape::new(TapeHeader::current(
            Some(1_700_000_000_000),
            Some("script.harn".to_string()),
            vec!["a".into()],
        ));
        tape.records.push(small_record(0, 250));
        tape.records.push(small_record(1, 750));
        tape.persist(&path).unwrap();

        let loaded = EventTape::load(&path).unwrap();
        assert_eq!(loaded.header.version, TAPE_FORMAT_VERSION);
        assert_eq!(loaded.header.argv, vec!["a".to_string()]);
        assert_eq!(loaded.records.len(), 2);
        match &loaded.records[0].kind {
            TapeRecordKind::ClockSleep { duration_ms } => assert_eq!(*duration_ms, 250),
            other => panic!("unexpected: {other:?}"),
        }
    }

    #[test]
    fn recorder_phase_guard_stamps_and_restores() {
        let recorder = Arc::new(TapeRecorder::new());
        let _recorder_guard = install_recorder(Arc::clone(&recorder));

        with_active_recorder(|_| Some(TapeRecordKind::ClockSleep { duration_ms: 1 }));
        {
            let _phase_guard = enter_phase(TapePhase::RuntimeFinalize).unwrap();
            with_active_recorder(|_| Some(TapeRecordKind::ClockSleep { duration_ms: 2 }));
        }
        with_active_recorder(|_| Some(TapeRecordKind::ClockSleep { duration_ms: 3 }));

        let tape = recorder.snapshot(TapeHeader::current(None, None, Vec::new()));
        let phases = tape
            .records
            .iter()
            .map(|record| record.phase)
            .collect::<Vec<_>>();
        assert_eq!(
            phases,
            vec![
                TapePhase::UserScript,
                TapePhase::RuntimeFinalize,
                TapePhase::UserScript
            ]
        );
    }

    #[test]
    fn large_payloads_spill_to_cas_and_round_trip() {
        let temp = TempDir::new().unwrap();
        let path = temp.path().join("run.tape");
        let mut tape = EventTape::new(TapeHeader::current(None, None, Vec::new()));
        let big = vec![b'x'; MAX_INLINE_BYTES + 32];
        let payload = build_payload(big.clone(), &mut tape.cas);
        let hash = payload.content_hash().to_string();
        let kind = TapeRecordKind::ProcessSpawn {
            program: "/bin/echo".to_string(),
            args: vec!["x".to_string()],
            cwd: None,
            exit_code: 0,
            duration_ms: 1,
            stdout_payload: payload,
            stderr_payload: build_payload(Vec::new(), &mut tape.cas),
        };
        tape.records.push(TapeRecord {
            seq: 0,
            phase: TapePhase::UserScript,
            virtual_time_ms: 0,
            monotonic_ms: 0,
            kind,
        });
        tape.persist(&path).unwrap();

        // CAS sidecar exists.
        assert!(path.with_extension("tape.cas").exists() || cas_dir_for(&path).exists());
        let cas_dir = cas_dir_for(&path);
        assert!(cas_dir.join(&hash).exists());

        let loaded = EventTape::load(&path).unwrap();
        let resolved = match &loaded.records[0].kind {
            TapeRecordKind::ProcessSpawn { stdout_payload, .. } => {
                loaded.resolve_payload(stdout_payload).unwrap()
            }
            other => panic!("unexpected: {other:?}"),
        };
        assert_eq!(resolved.len(), big.len());
    }

    #[test]
    fn rejects_newer_version() {
        let temp = TempDir::new().unwrap();
        let path = temp.path().join("future.tape");
        std::fs::write(
            &path,
            r#"{"type":"header","version":99,"harn_version":"x","started_at_unix_ms":null,"script_path":null,"argv":[]}
"#,
        )
        .unwrap();
        let err = EventTape::load(&path).unwrap_err();
        assert!(err.contains("version 99"), "{err}");
    }

    #[test]
    fn recorder_assigns_monotonic_seq() {
        let recorder = Arc::new(TapeRecorder::new());
        recorder.record(TapeRecordKind::ClockSleep { duration_ms: 1 });
        recorder.record(TapeRecordKind::ClockSleep { duration_ms: 2 });
        let snapshot = recorder.snapshot(TapeHeader::current(None, None, Vec::new()));
        assert_eq!(snapshot.records[0].seq, 0);
        assert_eq!(snapshot.records[1].seq, 1);
    }

    #[test]
    fn records_model_job_events_with_asset_digests() {
        let recorder = Arc::new(TapeRecorder::new());
        let _guard = install_recorder(Arc::clone(&recorder));
        record_model_job_event(&serde_json::json!({
            "schema": "harn.model_job_event.v1",
            "kind": "output",
            "job_id": "job-1",
            "request_id": "req-1",
            "backend": "fixture",
            "state": "succeeded",
            "assets": [
                {
                    "uri": "asset://sha256/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                    "sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                }
            ]
        }));
        let snapshot = recorder.snapshot(TapeHeader::current(None, None, Vec::new()));
        assert_eq!(snapshot.records.len(), 1);
        match &snapshot.records[0].kind {
            TapeRecordKind::ModelJob {
                job_id,
                event_kind,
                asset_digests,
                ..
            } => {
                assert_eq!(job_id, "job-1");
                assert_eq!(event_kind, "output");
                assert_eq!(
                    asset_digests,
                    &vec![
                        "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                            .to_string()
                    ]
                );
            }
            other => panic!("unexpected tape kind: {other:?}"),
        }
    }
}