openlatch-client 0.3.3

//! `openlatch evaluate` — the **only** way another process obtains a verdict.
//!
//! One process per job stage, fed NDJSON on stdin and answering NDJSON on
//! stdout. The tracked wire contract is `docs/evaluate-protocol.md`.
//! Consumers validate operational frames; engine identity and corpus metadata
//! are optional diagnostics, not release pins or startup requirements. Prefer
//! additive evolution; incompatible framing is a deliberate contract change.
//!
//! # Statelessness is the requirement that makes the rest checkable
//!
//! This subcommand must produce a verdict on a box holding **nothing but the
//! binary** — no enrolment, no credential, no daemon, no configuration file. So:
//! no audit line, no outbox spool, no dedup map, no filesystem access driven by
//! the event, no network, no clock (`now_ms` arrives in the frame), no
//! randomness, no config or keyring read, no global state.
//! `ci/check-engine-purity.py` scans this file for exactly that, and the
//! sentence that settles the grey cases is: *the process may cache what is
//! derived from its inputs; it may never retain what is derived from its
//! history.* A parsed bundle keyed by content digest is fine. A session counter
//! never is.
//!
//! Two consequences visible in the code below. The handshake's three baked
//! values come from `env!` — `build.rs` computed them — because reading
//! `schemas/conformance/` here would be a filesystem call. And the process
//! reaches this module **above** every `main()` init that resolves a data
//! directory or reads config, which is why `run` returns an exit code instead of
//! calling `process::exit`.
//!
//! # Two absolutes and one deadlock
//!
//! - **Errors are per row, never per stream.** A malformed event yields an error
//!   frame and the run continues, so a replay reports "all but four events
//!   evaluated" instead of dying midway.
//! - **Never write a partial line.** A response is serialised whole, then
//!   written.
//! - **Interleave.** Read a row, answer it, write the response — never buffer
//!   the run. Callers must feed stdin and drain stdout concurrently because
//!   this side interleaves; a caller that writes everything before reading
//!   deadlocks when the pipes fill.

use std::io::{BufRead, Write};
use std::panic::AssertUnwindSafe;

use clap::Args;
use serde::Serialize;
use serde_json::Value;

use crate::core::error::{
    ERR_EVAL_BUNDLE_REJECTED, ERR_EVAL_NO_BUNDLE, ERR_EVAL_PANICKED, ERR_EVAL_ROW_MALFORMED,
    ERR_EVAL_STATE_MISMATCH,
};
use crate::zone_eval::{self, Bundle, Decision, Event, SessionState, SkippedItem};

// ── Exit codes ───────────────────────────────────────────────────────────────
//
// The table in `docs/evaluate-protocol.md`, and it is different from the
// diagnostic one in `.claude/rules/error-handling.md`: this command answers a
// wire protocol, not "is this host healthy?". **Exit 1 is a RESULT, not a
// crash** — it means "answered, with at least one row error", which is the whole
// reason errors are per row.

/// Every row was answered with a decision.
const EXIT_ALL_ANSWERED: i32 = 0;
/// Every row was answered; at least one answer was an error frame.
const EXIT_ROW_ERRORS: i32 = 1;
/// The run never happened. Nothing was evaluated and nothing was answered.
const EXIT_COULD_NOT_START: i32 = 2;

// ── The handshake's baked provenance ─────────────────────────────────────────

/// The conformance corpus embedded at build time, from `build.rs`.
///
/// Optional, opaque provenance to the consumer, never an acceptance gate.
/// Publication readiness is verified separately by client CI. The corpus left
/// every published package (D5), so nothing on npm or PyPI can recompute it —
/// and this file may not read it off disk, so the value is baked at compile
/// time or it does not exist.
const CORPUS_DIGEST: &str = env!("OPENLATCH_CORPUS_DIGEST");

/// The wire schemas' release line, from `.release-please-manifest.schemas.json`.
const SCHEMAS_VERSION: &str = env!("OPENLATCH_SCHEMAS_VERSION");

/// The client's own version, from `Cargo.toml`.
///
/// **Not** `OPENLATCH_VERSION` (the git-describe identity `--version` prints):
/// the contract says `client_version` comes from `Cargo.toml`, and a
/// git-describe string would make the handshake — and the golden transcript —
/// change on every commit. `ENGINE_VERSION` separately tracks evaluation
/// semantics; neither version obliges a consumer to compare identities.
const CLIENT_VERSION: &str = env!("CARGO_PKG_VERSION");

// ── Arguments ────────────────────────────────────────────────────────────────

/// `openlatch evaluate [--describe] [--listen <SOCKET>]`.
#[derive(Args, Clone, Debug, Default)]
pub struct EvaluateArgs {
    /// Print the engine identity — the handshake object, verbatim — and exit.
    ///
    /// Reads no input and has no side effects. It prints the SAME object the
    /// stream's first frame carries, `"type":"handshake"` included, so the
    /// consumer parses both with one parser.
    #[arg(long)]
    pub describe: bool,

    /// Reserved for the remote decision endpoint (D7). Not implemented.
    ///
    /// Named in the tracked contract and built by nobody. It is declared so the
    /// CLI shape matches what the contract describes and so the refusal is a
    /// clear message rather than an unknown-flag error, and it is refused at
    /// runtime. Batch has no listener, no port and no token: the platform never
    /// holds a credential whose blast radius is code execution.
    #[arg(long, value_name = "SOCKET", conflicts_with = "describe")]
    pub listen: Option<String>,
}

// ── Frames ───────────────────────────────────────────────────────────────────
//
// Field ORDER is checked by the producer's golden transcript; consumers parse
// objects and do not gate on byte order or diagnostic identity. `type` is open:
// unknown frames and fields are ignored, but a consumer must still account for
// every required acknowledgment and row response before accepting the run.

/// The first frame written, before any row is read; and the whole of
/// `--describe`.
#[derive(Serialize)]
struct Handshake {
    #[serde(rename = "type")]
    frame: &'static str,
    engine_version: &'static str,
    client_version: &'static str,
    schemas_version: &'static str,
    protocol_version: u32,
    corpus_digest: &'static str,
}

/// Stage 2 succeeded: the bundle IS loaded, and `skipped` names what is not.
#[derive(Serialize)]
struct BundleAck<'a> {
    #[serde(rename = "type")]
    frame: &'static str,
    bundle_digest: &'a str,
    /// **Always present, `[]` when nothing was skipped** — absent and empty are
    /// not different. Without it an authoring UI shows a policy author a green
    /// light on a rule that never loaded.
    skipped: &'a [SkippedItem],
}

/// Stage 1 failed: the envelope did not deserialise, so **nothing loaded**.
#[derive(Serialize)]
struct BundleErrorFrame {
    #[serde(rename = "type")]
    frame: &'static str,
    error: FrameError,
}

/// One answer for one row. Provenance rides **around** the frozen `Decision`.
///
/// That is what lets the corpus compare expected against actual byte for byte
/// with no stripping step. A field moved inside `Decision` gives the frozen
/// contract a second definition, and the comparison then needs a stripping step
/// that both implementations have to agree about.
#[derive(Serialize)]
struct Response<'a> {
    #[serde(rename = "type")]
    frame: &'static str,
    /// Echoed verbatim, whatever the caller chose — `null` when it sent none.
    id: &'a Value,
    engine_version: &'static str,
    /// `null` when no bundle was loaded for this row. Present either way:
    /// provenance rides the envelope, so a consumer never has to ask whether the
    /// key exists.
    bundle_digest: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    decision: Option<&'a Decision>,
    /// **Absent on an errored row.** A row that failed must not advance session
    /// state, or the platform's 48-hour replay silently diverges: the next row's
    /// `state_in` is the last SUCCESSFUL `state_out`.
    #[serde(skip_serializing_if = "Option::is_none")]
    state_out: Option<&'a SessionState>,
    #[serde(skip_serializing_if = "Option::is_none")]
    error: Option<FrameError>,
}

/// `OL-`-prefixed open string, plus prose for a human.
///
/// The consumer never switches exhaustively on the code and never parses the
/// message.
#[derive(Serialize)]
struct FrameError {
    code: &'static str,
    message: String,
}

// ── Entry points ─────────────────────────────────────────────────────────────

/// Run the subcommand against the real standard streams.
///
/// Returns the exit code rather than taking it: `process::exit` here would be a
/// purity violation, and the caller in `main.rs` reaches this function *above*
/// every init that resolves a data directory, so there is nothing to unwind.
pub fn run(args: &EvaluateArgs) -> i32 {
    let stdin = std::io::stdin();
    let stdout = std::io::stdout();
    let stderr = std::io::stderr();
    run_with(args, stdin.lock(), stdout.lock(), stderr.lock())
}

/// [`run`], with the three streams supplied — the seam the tests drive.
pub fn run_with<R: BufRead, W: Write, E: Write>(
    args: &EvaluateArgs,
    input: R,
    mut out: W,
    mut err: E,
) -> i32 {
    if let Some(socket) = args.listen.as_deref() {
        let _ = writeln!(
            err,
            "openlatch evaluate --listen is not implemented: {socket} would be the remote \
             decision endpoint, and nothing builds one. Feed NDJSON rows on stdin instead."
        );
        return EXIT_COULD_NOT_START;
    }

    let handshake = match line_of(&Handshake {
        frame: "handshake",
        engine_version: zone_eval::ENGINE_VERSION,
        client_version: CLIENT_VERSION,
        schemas_version: SCHEMAS_VERSION,
        protocol_version: zone_eval::PROTOCOL_VERSION,
        corpus_digest: CORPUS_DIGEST,
    }) {
        Some(line) => line,
        // Unreachable for a struct of `&'static str`s, and still not a panic:
        // the one thing this command may never do is write half a line.
        None => return EXIT_COULD_NOT_START,
    };

    // Written once, BEFORE any row is read — and for `--describe`, it is the
    // whole output. One object, one parser.
    if write_line(&mut out, &handshake).is_err() {
        return EXIT_COULD_NOT_START;
    }
    if args.describe {
        return EXIT_ALL_ANSWERED;
    }

    stream(input, &mut out, &mut err)
}

// ── The stream ───────────────────────────────────────────────────────────────

/// The bundle the following rows evaluate against, and the digest that names it.
struct Active {
    digest: String,
    bundle: Bundle,
}

/// Read a frame, answer it, write the answer. Never buffer the run.
fn stream<R: BufRead, W: Write, E: Write>(input: R, out: &mut W, err: &mut E) -> i32 {
    let mut header: Option<Active> = None;
    let mut row_errors = 0usize;

    for line in input.lines() {
        let line = match line {
            Ok(line) => line,
            Err(e) => {
                // Invalid UTF-8, or the pipe broke. Neither is a frame, so
                // neither can be answered without desynchronising a consumer
                // that matches replies to frames positionally.
                let _ = writeln!(err, "openlatch evaluate: cannot read stdin: {e}");
                break;
            }
        };
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let Ok(Value::Object(frame)) = serde_json::from_str::<Value>(line) else {
            // A line that is not a JSON object carries no `type`, so there is no
            // way to know whether it wanted a reply. Reported and skipped: both
            // the platform's adapter and `tools/zone-eval-ref/drt.py` pair
            // replies with frames by position, and inventing a reply here would
            // shift every answer after it by one.
            let _ = writeln!(
                err,
                "openlatch evaluate: skipping a line that is not a JSON object"
            );
            continue;
        };

        match frame.get("type").and_then(Value::as_str) {
            Some("bundle") => {
                header = bundle_frame(&frame, out);
            }
            Some("row") => {
                let errored = row_frame(&frame, header.as_ref(), out);
                row_errors += usize::from(errored);
            }
            // An unknown `type` is skipped, never fatal, and consumes no
            // position in the reply stream. `type` is an OPEN string (R14).
            _ => continue,
        }
    }

    if row_errors > 0 {
        EXIT_ROW_ERRORS
    } else {
        EXIT_ALL_ANSWERED
    }
}

/// A bundle header frame: load it, answer with the ack or the error.
///
/// The reply IS the load-check result — `load_check` is not a separate verb, it
/// is this frame followed by zero rows.
///
/// A `bundle_error` does **not** make the run exit 1: exit 1 means *at least one
/// ROW error*, and a consumer that reconciles the two would call a load-check
/// run with no rows in it a contradiction.
fn bundle_frame<W: Write>(frame: &serde_json::Map<String, Value>, out: &mut W) -> Option<Active> {
    let document = frame.get("bundle").cloned().unwrap_or(Value::Null);
    let digest = digest_of(&document);

    match zone_eval::load(document) {
        Ok(bundle) => {
            emit(
                out,
                &BundleAck {
                    frame: "bundle_ack",
                    bundle_digest: &digest,
                    skipped: &bundle.skipped,
                },
            );
            Some(Active { digest, bundle })
        }
        Err(e) => {
            emit(
                out,
                &BundleErrorFrame {
                    frame: "bundle_error",
                    error: FrameError {
                        code: ERR_EVAL_BUNDLE_REJECTED,
                        message: e.to_string(),
                    },
                },
            );
            // Stage 1 failed, so NOTHING loaded — and a header that failed must
            // not leave the previous one armed. Rows that follow are answered
            // with OL-1242 until the next header frame succeeds.
            None
        }
    }
}

/// One request row, answered. Returns whether the answer was an error frame.
fn row_frame<W: Write>(
    frame: &serde_json::Map<String, Value>,
    header: Option<&Active>,
    out: &mut W,
) -> bool {
    let id = frame.get("id").unwrap_or(&Value::Null);

    // A row's own `bundle` overrides the header for that row only, and it never
    // touches disk: it is parsed straight out of the frame — no temporary file,
    // no cleanup.
    let inline = match frame.get("bundle") {
        Some(Value::Null) | None => None,
        Some(document) => {
            let digest = digest_of(document);
            match zone_eval::load(document.clone()) {
                Ok(bundle) => Some(Active { digest, bundle }),
                Err(e) => {
                    // Per ROW: this row's inline bundle is unusable, and the
                    // header (if any) is untouched for the rows that follow.
                    return error_response(
                        out,
                        id,
                        None,
                        ERR_EVAL_BUNDLE_REJECTED,
                        format!("inline bundle: {e}"),
                    );
                }
            }
        }
    };

    let Some(active) = inline.as_ref().or(header) else {
        return error_response(
            out,
            id,
            None,
            ERR_EVAL_NO_BUNDLE,
            "no bundle is loaded: send a `bundle` header frame, or give the row its own `bundle`"
                .to_string(),
        );
    };
    let digest = Some(active.digest.as_str());

    let event: Event = match frame.get("event") {
        Some(raw) => match serde_json::from_value(raw.clone()) {
            Ok(event) => event,
            Err(e) => {
                return error_response(
                    out,
                    id,
                    digest,
                    ERR_EVAL_ROW_MALFORMED,
                    format!("`event` did not parse: {e}"),
                )
            }
        },
        None => {
            return error_response(
                out,
                id,
                digest,
                ERR_EVAL_ROW_MALFORMED,
                "the row carries no `event`".to_string(),
            )
        }
    };

    // Injected, never read from a clock: a call to "now" in here makes every
    // replay of the same row non-reproducible.
    let Some(now_ms) = frame.get("now_ms").and_then(Value::as_i64) else {
        return error_response(
            out,
            id,
            digest,
            ERR_EVAL_ROW_MALFORMED,
            "`now_ms` is absent or not an integer".to_string(),
        );
    };

    // `null` is NOT a blank state. It means the session's registers are gone,
    // and a Tier 2 artifact answers that through its own `on_evict` — `reinit`,
    // `unknown` or `fail_static`. Collapsing the two makes `fail_static` a
    // silent allow, which is a fail-open in the one place AGENTS.md says there
    // must never be one.
    let state_in: Option<SessionState> = match frame.get("state_in") {
        Some(Value::Null) | None => None,
        Some(raw) => match serde_json::from_value::<SessionState>(raw.clone()) {
            Ok(state) => Some(state),
            Err(e) => {
                return error_response(
                    out,
                    id,
                    digest,
                    ERR_EVAL_ROW_MALFORMED,
                    format!("`state_in` did not parse: {e}"),
                )
            }
        },
    };
    if let Some(state) = state_in.as_ref() {
        // Malformed, never padded: two implementations cannot agree about which
        // counter is which if one of them silently invents a slot.
        if !state.matches(&active.bundle.state_layout) {
            return error_response(
                out,
                id,
                digest,
                ERR_EVAL_STATE_MISMATCH,
                format!(
                    "`state_in` does not match the bundle's state_layout {:?}",
                    active.bundle.state_layout
                ),
            );
        }
    }

    // The engine must never panic on input it did not author, and if it ever
    // does, that bug costs ONE row. A replay that dies at event 40 000 of 90 000
    // reports nothing about the 50 000 after it.
    let evaluated = std::panic::catch_unwind(AssertUnwindSafe(|| {
        zone_eval::evaluate(&active.bundle, &event, state_in.as_ref(), now_ms)
    }));
    let (decision, state_out) = match evaluated {
        Ok(answer) => answer,
        Err(_) => {
            return error_response(
                out,
                id,
                digest,
                ERR_EVAL_PANICKED,
                "the engine panicked evaluating this row".to_string(),
            )
        }
    };

    emit(
        out,
        &Response {
            frame: "response",
            id,
            engine_version: zone_eval::ENGINE_VERSION,
            bundle_digest: digest,
            decision: Some(&decision),
            state_out: Some(&state_out),
            error: None,
        },
    );
    false
}

/// Answer one row with an error frame. **No `state_out`.** Always returns `true`
/// so the caller counts the row error that decides the exit code.
fn error_response<W: Write>(
    out: &mut W,
    id: &Value,
    bundle_digest: Option<&str>,
    code: &'static str,
    message: String,
) -> bool {
    emit(
        out,
        &Response {
            frame: "response",
            id,
            engine_version: zone_eval::ENGINE_VERSION,
            bundle_digest,
            decision: None,
            state_out: None,
            error: Some(FrameError { code, message }),
        },
    );
    true
}

// ── Writing ──────────────────────────────────────────────────────────────────

/// Serialise a frame whole, then write it. A partial line is never written.
fn emit<W: Write, T: Serialize>(out: &mut W, frame: &T) {
    let Some(line) = line_of(frame) else {
        return;
    };
    let _ = write_line(out, &line);
}

/// One frame, one line, `\n`-terminated — or `None`, having written nothing.
fn line_of<T: Serialize>(frame: &T) -> Option<String> {
    let mut line = serde_json::to_string(frame).ok()?;
    line.push('\n');
    Some(line)
}

/// Write a whole line and flush it.
///
/// The flush is the interleave: the consumer feeds stdin and drains stdout
/// concurrently, and an answer sitting in a buffer until the run ends is a run
/// that deadlocks the moment the pipe fills.
fn write_line<W: Write>(out: &mut W, line: &str) -> std::io::Result<()> {
    out.write_all(line.as_bytes())?;
    out.flush()
}

/// `sha256:` over the bundle document in RFC 8785 canonical form.
///
/// JCS, not `to_string`: the digest names a *document*, so two callers that
/// serialised the same bundle with different key order must produce the same
/// name for it. It is the same canonicaliser `ground_key` hashes with.
fn digest_of(document: &Value) -> String {
    use sha2::{Digest as _, Sha256};
    use std::fmt::Write as _;

    let canonical = serde_json_canonicalizer::to_string(document).unwrap_or_default();
    let mut hasher = Sha256::new();
    hasher.update(canonical.as_bytes());

    let mut out = String::with_capacity(71);
    out.push_str("sha256:");
    for byte in hasher.finalize() {
        let _ = write!(out, "{byte:02x}");
    }
    out
}

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

    /// A bundle with one Tier 1 artifact that blocks on `curl`, and one artifact
    /// whose body cannot be parsed — so `skipped` is real output.
    fn bundle_document() -> Value {
        serde_json::json!({
            "schema_version": 2,
            "organization_id": "org-test",
            "revision": 1,
            "built_at": "2026-09-01T00:00:00Z",
            "enforcement_enabled": true,
            "signature": null,
            "artifacts": [
                {
                    "artifact_id": "block-curl",
                    "atom_id": "atom-block-curl",
                    "mode": "enforce",
                    "tier": 1,
                    "kind": "t1_predicate_tree",
                    "body": {
                        "node": {"op": "leaf", "leaf": {
                            "pred": "keyword", "field": "input.strings", "value": ["curl"]
                        }},
                        "verdict": "block",
                        "reason": "no ad-hoc network egress",
                    },
                },
                {
                    "artifact_id": "malformed",
                    "atom_id": "atom-malformed",
                    "mode": "enforce",
                    "tier": 1,
                    "kind": "t1_predicate_tree",
                    "body": {"node": "not a node", "verdict": "block", "reason": "unloadable"},
                },
            ],
        })
    }

    fn frames(lines: &[Value]) -> String {
        lines
            .iter()
            .map(|line| serde_json::to_string(line).expect("a test frame serialises"))
            .collect::<Vec<_>>()
            .join("\n")
    }

    fn drive(input: &str) -> (i32, Vec<Value>, String) {
        let mut out: Vec<u8> = Vec::new();
        let mut err: Vec<u8> = Vec::new();
        let code = run_with(
            &EvaluateArgs::default(),
            input.as_bytes(),
            &mut out,
            &mut err,
        );
        let out = String::from_utf8(out).expect("the producer writes UTF-8");
        let frames = out
            .lines()
            .map(|line| serde_json::from_str(line).expect("every line is one JSON object"))
            .collect();
        (code, frames, String::from_utf8_lossy(&err).into_owned())
    }

    #[test]
    fn describe_prints_the_handshake_verbatim_and_reads_nothing() {
        let mut out: Vec<u8> = Vec::new();
        let mut err: Vec<u8> = Vec::new();
        let args = EvaluateArgs {
            describe: true,
            listen: None,
        };
        // Stdin holds a row. `--describe` must not read it.
        let code = run_with(
            &args,
            b"{\"type\":\"row\"}\n".as_slice(),
            &mut out,
            &mut err,
        );
        assert_eq!(code, EXIT_ALL_ANSWERED);
        let out = String::from_utf8(out).expect("UTF-8");
        assert_eq!(out.lines().count(), 1, "one line, exactly");
        let frame: Value = serde_json::from_str(out.trim()).expect("JSON");
        assert_eq!(frame["type"], "handshake", "not a shape of its own");
        assert_eq!(frame["engine_version"], zone_eval::ENGINE_VERSION);
        assert_eq!(frame["protocol_version"], zone_eval::PROTOCOL_VERSION);
        assert_ne!(
            frame["engine_version"], frame["client_version"],
            "engine semantics and the client release have independent version lifecycles"
        );
        assert!(err.is_empty(), "no side effects, nothing on stderr");
    }

    #[test]
    fn the_stream_writes_its_handshake_before_any_row_is_read() {
        let (_, frames, _) = drive("");
        assert_eq!(frames.len(), 1);
        assert_eq!(frames[0]["type"], "handshake");
    }

    #[test]
    fn a_bundle_frame_acks_with_a_skipped_entry_and_the_rest_armed() {
        let input = frames(&[serde_json::json!({
            "type": "bundle", "bundle": bundle_document()
        })]);
        let (code, frames, _) = drive(&input);
        assert_eq!(code, EXIT_ALL_ANSWERED, "a load check answered, no rows");
        let ack = &frames[1];
        assert_eq!(ack["type"], "bundle_ack");
        assert!(ack["bundle_digest"]
            .as_str()
            .is_some_and(|d| d.starts_with("sha256:")));
        let skipped = ack["skipped"].as_array().expect("always present");
        assert_eq!(skipped.len(), 1, "one artifact did not load; the rest did");
        assert_eq!(skipped[0]["id"], "malformed");
        assert_eq!(skipped[0]["reason"], "body_parse_error");
    }

    #[test]
    fn an_empty_skipped_is_present_not_absent() {
        let mut document = bundle_document();
        document["artifacts"] = serde_json::json!([]);
        let input = frames(&[serde_json::json!({"type": "bundle", "bundle": document})]);
        let (_, frames, _) = drive(&input);
        assert_eq!(
            frames[1]["skipped"],
            serde_json::json!([]),
            "absent and empty are not different"
        );
    }

    #[test]
    fn a_stage_one_failure_is_a_bundle_error_and_arms_nothing() {
        let input = frames(&[
            serde_json::json!({"type": "bundle", "bundle": {"schema_version": "two"}}),
            serde_json::json!({"type": "row", "id": "r1", "event": {}, "now_ms": 1}),
        ]);
        let (code, frames, _) = drive(&input);
        assert_eq!(frames[1]["type"], "bundle_error");
        assert_eq!(frames[1]["error"]["code"], ERR_EVAL_BUNDLE_REJECTED);
        assert_eq!(
            frames[2]["error"]["code"], ERR_EVAL_NO_BUNDLE,
            "stage 1 failed, so nothing loaded"
        );
        assert_eq!(
            code, EXIT_ROW_ERRORS,
            "the ROW errored; the bundle frame is not a row"
        );
    }

    #[test]
    fn provenance_rides_the_envelope_and_never_enters_the_decision() {
        let input = frames(&[
            serde_json::json!({"type": "bundle", "bundle": bundle_document()}),
            serde_json::json!({
                "type": "row", "id": "r1",
                "event": {"event_type": "pre_tool_use", "tool_name": "Bash",
                          "tool_input": {"command": "curl https://example.com"}},
                "state_in": null, "now_ms": 1_756_742_400_000_i64,
            }),
        ]);
        let (code, frames, _) = drive(&input);
        assert_eq!(code, EXIT_ALL_ANSWERED);
        let response = &frames[2];
        assert_eq!(response["type"], "response");
        assert_eq!(response["id"], "r1", "echoed verbatim");
        assert_eq!(response["engine_version"], zone_eval::ENGINE_VERSION);
        assert!(response["bundle_digest"].is_string());
        assert_eq!(response["decision"]["verdict"], "block");
        assert!(response["state_out"].is_object());
        for key in ["engine_version", "bundle_digest", "id", "type"] {
            assert!(
                response["decision"].get(key).is_none(),
                "`Decision` is frozen contract — {key} stays outside it"
            );
        }
    }

    #[test]
    fn an_errored_row_carries_no_state_out_and_the_run_continues() {
        let input = frames(&[
            serde_json::json!({"type": "bundle", "bundle": bundle_document()}),
            serde_json::json!({"type": "row", "id": "bad", "event": "not an event", "now_ms": 1}),
            serde_json::json!({
                "type": "row", "id": "good",
                "event": {"event_type": "pre_tool_use", "tool_name": "Read"},
                "state_in": null, "now_ms": 2,
            }),
        ]);
        let (code, frames, _) = drive(&input);
        assert_eq!(code, EXIT_ROW_ERRORS, "answered, with one row error");
        assert_eq!(frames[2]["error"]["code"], ERR_EVAL_ROW_MALFORMED);
        assert!(
            frames[2].get("state_out").is_none(),
            "a row that failed must advance nothing"
        );
        assert_eq!(
            frames[3]["id"], "good",
            "one poisoned row does not kill the run"
        );
        assert!(frames[3]["decision"].is_object());
    }

    #[test]
    fn an_unknown_frame_type_is_skipped_and_consumes_no_reply_position() {
        let input = frames(&[
            serde_json::json!({"type": "bundle", "bundle": bundle_document()}),
            serde_json::json!({"type": "from_a_newer_platform", "whatever": 1}),
            serde_json::json!({
                "type": "row", "id": "r1",
                "event": {"event_type": "pre_tool_use"}, "state_in": null, "now_ms": 1,
            }),
        ]);
        let (code, frames, _) = drive(&input);
        assert_eq!(code, EXIT_ALL_ANSWERED);
        assert_eq!(
            frames.len(),
            3,
            "handshake, ack, one response — and no more"
        );
        assert_eq!(frames[2]["id"], "r1");
    }

    #[test]
    fn an_inline_bundle_overrides_the_header_for_that_row_only() {
        let mut permissive = bundle_document();
        permissive["artifacts"] = serde_json::json!([]);
        let event = serde_json::json!({
            "event_type": "pre_tool_use", "tool_name": "Bash",
            "tool_input": {"command": "curl https://example.com"}
        });
        let input = frames(&[
            serde_json::json!({"type": "bundle", "bundle": bundle_document()}),
            serde_json::json!({"type": "row", "id": "inline", "event": event,
                               "state_in": null, "now_ms": 1, "bundle": permissive}),
            serde_json::json!({"type": "row", "id": "header", "event": event,
                               "state_in": null, "now_ms": 1}),
        ]);
        let (code, frames, _) = drive(&input);
        assert_eq!(code, EXIT_ALL_ANSWERED);
        assert_eq!(
            frames[2]["decision"]["verdict"], "allow",
            "the row's own bundle"
        );
        assert_eq!(
            frames[3]["decision"]["verdict"], "block",
            "the header, still armed"
        );
        assert_ne!(
            frames[2]["bundle_digest"], frames[3]["bundle_digest"],
            "each response names the bundle that actually answered it"
        );
    }

    #[test]
    fn a_state_in_that_disagrees_with_the_layout_is_malformed_never_padded() {
        let input = frames(&[
            serde_json::json!({"type": "bundle", "bundle": bundle_document()}),
            serde_json::json!({
                "type": "row", "id": "r1", "event": {"event_type": "pre_tool_use"},
                "state_in": {"c": [1, 2, 3], "f": [], "t": [], "a": [], "run": null},
                "now_ms": 1,
            }),
        ]);
        let (code, frames, _) = drive(&input);
        assert_eq!(code, EXIT_ROW_ERRORS);
        assert_eq!(frames[2]["error"]["code"], ERR_EVAL_STATE_MISMATCH);
    }

    #[test]
    fn a_line_that_is_not_json_is_reported_and_shifts_no_answer() {
        let input = format!(
            "{}\nnot json at all\n{}\n",
            serde_json::json!({"type": "bundle", "bundle": bundle_document()}),
            serde_json::json!({
                "type": "row", "id": "r1", "event": {"event_type": "pre_tool_use"},
                "state_in": null, "now_ms": 1,
            }),
        );
        let (code, frames, err) = drive(&input);
        assert_eq!(code, EXIT_ALL_ANSWERED);
        assert_eq!(frames.len(), 3);
        assert_eq!(frames[2]["id"], "r1");
        assert!(err.contains("not a JSON object"));
    }

    #[test]
    fn listen_is_refused_rather_than_implemented() {
        let mut out: Vec<u8> = Vec::new();
        let mut err: Vec<u8> = Vec::new();
        let args = EvaluateArgs {
            describe: false,
            listen: Some("/tmp/nope.sock".to_string()),
        };
        let code = run_with(&args, b"".as_slice(), &mut out, &mut err);
        assert_eq!(code, EXIT_COULD_NOT_START);
        assert!(
            out.is_empty(),
            "not even a handshake: the run never started"
        );
        assert!(String::from_utf8_lossy(&err).contains("not implemented"));
    }

    #[test]
    fn the_digest_names_the_document_not_its_key_order() {
        let a = serde_json::json!({"a": 1, "b": [2, 3]});
        let b = serde_json::json!({"b": [2, 3], "a": 1});
        assert_eq!(digest_of(&a), digest_of(&b));
        assert_ne!(digest_of(&a), digest_of(&serde_json::json!({"a": 1})));
    }
}