openlatch-client 0.6.19

//! The Bedrock Converse decoder — usage and model out of `converse` and `converse-stream`
//! answers (AGT-6 C-6, plan AGT-6-P2 D-6).
//!
//! **Capture-only.** The relay forwards every byte it receives, untouched, before this module
//! sees it; nothing here can change, delay or drop a byte of the response. A frame this decoder
//! cannot trust ends capture for that response (`decode_failed`) and the stream goes on to the
//! agent exactly as it came.
//!
//! ## The frame grammar (the AWS event-stream encoding)
//!
//! ```text
//! | total_len u32 BE | headers_len u32 BE | prelude_crc u32 BE |   12-byte prelude
//! | headers (headers_len bytes)                                |
//! | payload (total_len - headers_len - 16 bytes)               |
//! | message_crc u32 BE                                         |   CRC32 of everything before it
//! ```
//!
//! A header is `name_len u8 · name · type u8 · value`, the value shaped by its type: `0`/`1`
//! booleans (no bytes), `2` byte, `3` i16, `4` i32, `5` i64, `6` bytes and `7` string (a `u16`
//! length, then the bytes), `8` timestamp (8 bytes), `9` uuid (16 bytes).
//!
//! ## Checked before anything is trusted
//!
//! The prelude CRC is checked before its lengths are believed, and only then:
//! `16 <= total_len <= 24 MiB`, `headers_len <= total_len - 16`, and the header block under
//! [`MAX_HEADERS_LEN`] — so a hostile length never allocates. A header value over
//! [`MAX_HEADER_VALUE`] bytes, or an unknown header type, ends capture. Only the payload of a
//! `:event-type = metadata` event is buffered, under [`MAX_METADATA_PAYLOAD`]; every other
//! payload byte streams through the CRC and is dropped. The message CRC is checked before a
//! buffered payload is read.
//!
//! ## What is kept
//!
//! The `metadata` event's `usage` (token counts) and, when the response carries it,
//! `trace.promptRouter.invokedModelId` — reduced to its resource id, the AWS account id dropped.
//! `trace` itself is never retained, and no payload is logged. An exception frame
//! (`:message-type = exception`) ends capture cleanly: no usage, no record, and nothing logged
//! beyond its `:exception-type` name.
//!
//! No `aws-smithy-eventstream`: a capture-only reader needs the frame grammar alone, and that
//! crate buffers whole messages (plan D-6). `crc32fast` comes in through the `model-relay`
//! feature only, so the hook binary's dependencies do not change.

use crc32fast::Hasher;
use serde_json::Value;

use super::capture::Usage;

/// Smallest legal frame: the 12-byte prelude and the 4-byte message CRC.
pub const MIN_FRAME_LEN: u32 = 16;
/// Largest frame this decoder accepts (plan D-6).
pub const MAX_FRAME_LEN: u32 = 24 * 1024 * 1024;
/// Largest header block this decoder buffers. AWS caps a message's headers at 128 KiB.
pub const MAX_HEADERS_LEN: u32 = 128 * 1024;
/// Largest single header value — a bytes or string header — this decoder accepts (plan D-6).
pub const MAX_HEADER_VALUE: usize = 32_767;
/// Largest `metadata` payload this decoder buffers (plan D-6).
pub const MAX_METADATA_PAYLOAD: usize = 64 * 1024;
/// Longest model id this decoder records. Anything longer is not an id worth labelling a record
/// with, so the record keeps the path's model instead.
const MAX_MODEL_LEN: usize = 512;
/// Longest exception name this decoder logs.
const MAX_LOGGED_EXCEPTION: usize = 64;
const PRELUDE_LEN: usize = 12;

/// Why a frame ended capture (`decode_failed`).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Failure {
    /// The prelude CRC did not match its first eight bytes.
    PreludeCrc,
    /// `total_len` outside `16..=24 MiB`.
    FrameLength,
    /// `headers_len` larger than the frame leaves room for, or than [`MAX_HEADERS_LEN`].
    HeadersLength,
    /// A bytes or string header value longer than [`MAX_HEADER_VALUE`].
    HeaderValueTooLong,
    /// A header type outside `0..=9`.
    UnknownHeaderType,
    /// A header block that ends inside a header.
    MalformedHeader,
    /// A `metadata` payload larger than [`MAX_METADATA_PAYLOAD`].
    MetadataTooLarge,
    /// The message CRC did not match the frame.
    MessageCrc,
    /// The stream ended inside a frame — its prelude, headers, payload or message CRC.
    Truncated,
}

impl Failure {
    /// A stable label for the local log.
    pub fn as_str(self) -> &'static str {
        match self {
            Failure::PreludeCrc => "prelude_crc",
            Failure::FrameLength => "frame_length",
            Failure::HeadersLength => "headers_length",
            Failure::HeaderValueTooLong => "header_value_too_long",
            Failure::UnknownHeaderType => "unknown_header_type",
            Failure::MalformedHeader => "malformed_header",
            Failure::MetadataTooLarge => "metadata_too_large",
            Failure::MessageCrc => "message_crc",
            Failure::Truncated => "truncated",
        }
    }
}

/// Where the decoder stands.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Outcome {
    /// Still reading frames.
    Running,
    /// A frame failed a check: capture has ended (`decode_failed`). The bytes still flow.
    DecodeFailed(Failure),
    /// An exception frame arrived: capture has ended cleanly, with no usage and no record.
    Exception,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Phase {
    Prelude,
    Headers,
    Payload,
    Crc,
}

/// What the current frame is, from its headers.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum FrameKind {
    /// `:message-type = event`, `:event-type = metadata` — the one payload kept.
    Metadata,
    /// `:message-type = exception`.
    Exception,
    /// Anything else: its payload streams through the CRC and is dropped.
    Other,
}

/// An incremental decoder over one `converse-stream` response, fed the response chunks in order,
/// split anywhere.
#[derive(Clone)]
pub struct Decoder {
    phase: Phase,
    /// The prelude, the header block or the message CRC being assembled. Never a payload.
    buf: Vec<u8>,
    hasher: Hasher,
    headers_len: usize,
    payload_left: usize,
    kind: FrameKind,
    /// The `metadata` payload being assembled, under [`MAX_METADATA_PAYLOAD`].
    payload: Vec<u8>,
    outcome: Outcome,
    usage: Option<Usage>,
    model: Option<String>,
}

impl Default for Decoder {
    fn default() -> Self {
        Decoder {
            phase: Phase::Prelude,
            buf: Vec::new(),
            hasher: Hasher::new(),
            headers_len: 0,
            payload_left: 0,
            kind: FrameKind::Other,
            payload: Vec::new(),
            outcome: Outcome::Running,
            usage: None,
            model: None,
        }
    }
}

impl std::fmt::Debug for Decoder {
    /// Phase and outcome only: the decoder holds no payload worth showing, and holds a metadata
    /// payload only between its bytes and its CRC.
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Decoder")
            .field("phase", &self.phase)
            .field("outcome", &self.outcome)
            .field("usage", &self.usage)
            .finish_non_exhaustive()
    }
}

/// Move bytes from `input` into `buf` until it holds `target` bytes. True once it does.
fn fill(buf: &mut Vec<u8>, input: &mut &[u8], target: usize) -> bool {
    let need = target.saturating_sub(buf.len());
    let take = need.min(input.len());
    buf.extend_from_slice(&input[..take]);
    *input = &input[take..];
    buf.len() >= target
}

fn be_u32(b: &[u8]) -> u32 {
    let mut a = [0u8; 4];
    a.copy_from_slice(&b[..4]);
    u32::from_be_bytes(a)
}

impl Decoder {
    /// Feed the next response chunk. Never fails and never panics: a frame that does not check
    /// out ends capture ([`Outcome::DecodeFailed`]), and every later chunk is ignored.
    pub fn feed(&mut self, mut input: &[u8]) {
        while self.outcome == Outcome::Running {
            match self.phase {
                Phase::Prelude => {
                    if !fill(&mut self.buf, &mut input, PRELUDE_LEN) {
                        return;
                    }
                    if let Err(f) = self.start_frame() {
                        self.fail(f);
                    }
                }
                Phase::Headers => {
                    if !fill(&mut self.buf, &mut input, self.headers_len) {
                        return;
                    }
                    self.hasher.update(&self.buf);
                    match parse_headers(&self.buf) {
                        Ok(kind) => {
                            self.kind = kind;
                            self.buf.clear();
                            self.phase = Phase::Payload;
                        }
                        Err(f) => self.fail(f),
                    }
                }
                Phase::Payload => {
                    if self.payload_left > 0 {
                        if input.is_empty() {
                            return;
                        }
                        let n = self.payload_left.min(input.len());
                        let (chunk, rest) = input.split_at(n);
                        input = rest;
                        self.hasher.update(chunk);
                        self.payload_left -= n;
                        if self.kind == FrameKind::Metadata {
                            if self.payload.len() + n > MAX_METADATA_PAYLOAD {
                                self.fail(Failure::MetadataTooLarge);
                                continue;
                            }
                            self.payload.extend_from_slice(chunk);
                        }
                    }
                    if self.payload_left == 0 {
                        self.phase = Phase::Crc;
                    }
                }
                Phase::Crc => {
                    if !fill(&mut self.buf, &mut input, 4) {
                        return;
                    }
                    let expected = be_u32(&self.buf);
                    let actual = std::mem::take(&mut self.hasher).finalize();
                    if expected != actual {
                        self.fail(Failure::MessageCrc);
                        continue;
                    }
                    self.end_frame();
                }
            }
        }
    }

    /// The prelude is complete in `buf`: check it, then believe its lengths.
    fn start_frame(&mut self) -> Result<(), Failure> {
        let total = be_u32(&self.buf[0..4]);
        let headers = be_u32(&self.buf[4..8]);
        let prelude_crc = be_u32(&self.buf[8..12]);
        // The CRC first: an unchecked length is never compared, let alone allocated for.
        if crc32fast::hash(&self.buf[0..8]) != prelude_crc {
            return Err(Failure::PreludeCrc);
        }
        if !(MIN_FRAME_LEN..=MAX_FRAME_LEN).contains(&total) {
            return Err(Failure::FrameLength);
        }
        if headers > total - MIN_FRAME_LEN || headers > MAX_HEADERS_LEN {
            return Err(Failure::HeadersLength);
        }
        self.hasher = Hasher::new();
        self.hasher.update(&self.buf);
        self.buf.clear();
        self.headers_len = headers as usize;
        self.payload_left = (total - MIN_FRAME_LEN - headers) as usize;
        self.kind = FrameKind::Other;
        self.payload.clear();
        self.phase = Phase::Headers;
        Ok(())
    }

    /// A frame passed its message CRC: act on it, then expect the next prelude.
    fn end_frame(&mut self) {
        match self.kind {
            FrameKind::Metadata => {
                if let Ok(v) = serde_json::from_slice::<Value>(&self.payload) {
                    if let Some(u) = v.get("usage").and_then(converse_usage) {
                        self.usage = Some(u);
                    }
                    if let Some(m) = invoked_model(&v) {
                        self.model = Some(m);
                    }
                }
            }
            FrameKind::Exception => {
                self.outcome = Outcome::Exception;
            }
            FrameKind::Other => {}
        }
        self.payload = Vec::new();
        self.buf.clear();
        self.phase = Phase::Prelude;
    }

    fn fail(&mut self, failure: Failure) {
        tracing::debug!(
            reason = failure.as_str(),
            "bedrock converse stream: frame failed its checks — capture ends, bytes forwarded unchanged"
        );
        self.outcome = Outcome::DecodeFailed(failure);
        self.buf = Vec::new();
        self.payload = Vec::new();
    }

    /// The response stream has ended — completed, errored or dropped. A stream cut inside a
    /// frame never passed that frame's message CRC, so capture ends `decode_failed`
    /// ([`Failure::Truncated`]): whatever usage an earlier frame carried is not the provider's
    /// last word. A stream that ended between frames is left as it is.
    pub fn finish(&mut self) {
        if self.outcome == Outcome::Running
            && (self.phase != Phase::Prelude || !self.buf.is_empty())
        {
            self.fail(Failure::Truncated);
        }
    }

    /// Where the decoder stands.
    pub fn outcome(&self) -> Outcome {
        self.outcome
    }

    /// The token counts of the `metadata` event, once one passed its CRC.
    pub fn usage(&self) -> Option<Usage> {
        self.usage
    }

    /// The model the prompt router invoked, when the `metadata` event named one.
    pub fn model(&self) -> Option<&str> {
        self.model.as_deref()
    }
}

/// The next `n` bytes of a header block, or [`Failure::MalformedHeader`] when it ends first.
fn take<'a>(b: &mut &'a [u8], n: usize) -> Result<&'a [u8], Failure> {
    if b.len() < n {
        return Err(Failure::MalformedHeader);
    }
    let (head, rest) = b.split_at(n);
    *b = rest;
    Ok(head)
}

/// Parse a complete header block into the frame's kind. Only `:message-type`, `:event-type` and
/// `:exception-type` are read; every other header is walked and skipped.
fn parse_headers(mut b: &[u8]) -> Result<FrameKind, Failure> {
    let mut message_type: Option<&[u8]> = None;
    let mut event_type: Option<&[u8]> = None;
    let mut exception_type: Option<&[u8]> = None;
    while !b.is_empty() {
        let name_len = take(&mut b, 1)?[0] as usize;
        let name = take(&mut b, name_len)?;
        let value_type = take(&mut b, 1)?[0];
        let value: Option<&[u8]> = match value_type {
            0 | 1 => None,
            2 => Some(take(&mut b, 1)?),
            3 => Some(take(&mut b, 2)?),
            4 => Some(take(&mut b, 4)?),
            5 | 8 => Some(take(&mut b, 8)?),
            6 | 7 => {
                let len_bytes = take(&mut b, 2)?;
                let len = u16::from_be_bytes([len_bytes[0], len_bytes[1]]) as usize;
                if len > MAX_HEADER_VALUE {
                    return Err(Failure::HeaderValueTooLong);
                }
                Some(take(&mut b, len)?)
            }
            9 => Some(take(&mut b, 16)?),
            _ => return Err(Failure::UnknownHeaderType),
        };
        if value_type == 7 {
            match name {
                b":message-type" => message_type = value,
                b":event-type" => event_type = value,
                b":exception-type" => exception_type = value,
                _ => {}
            }
        }
    }
    Ok(match message_type {
        Some(b"exception") => {
            let name = exception_type
                .map(|n| String::from_utf8_lossy(&n[..n.len().min(MAX_LOGGED_EXCEPTION)]))
                .unwrap_or_default();
            // The exception's name, and nothing else: its payload is never read.
            tracing::debug!(
                exception_type = %name,
                "bedrock converse stream: exception frame — capture ends, no record"
            );
            FrameKind::Exception
        }
        Some(b"event") if event_type == Some(b"metadata".as_slice()) => FrameKind::Metadata,
        _ => FrameKind::Other,
    })
}

/// The six token counts out of a Converse `usage` object — the `metadata` event's, or the
/// non-streaming answer's top-level one. `None` unless it is an object.
///
/// `inputTokens` is Bedrock's uncached input: cache reads and writes are reported beside it
/// (`cacheReadInputTokens`, `cacheWriteInputTokens`) and are not subtracted, the same shape as
/// Anthropic's own `usage`.
pub fn converse_usage(u: &Value) -> Option<Usage> {
    if !u.is_object() {
        return None;
    }
    let n = |k: &str| u.get(k).and_then(Value::as_u64).unwrap_or(0);
    Some(Usage {
        input_tokens: n("inputTokens"),
        cache_read: n("cacheReadInputTokens"),
        cache_write: n("cacheWriteInputTokens"),
        eph_5m: 0,
        eph_1h: 0,
        output_tokens: n("outputTokens"),
    })
}

/// `trace.promptRouter.invokedModelId`, reduced to its resource id. `trace` is read for this one
/// string and never retained.
pub fn invoked_model(v: &Value) -> Option<String> {
    v.get("trace")?
        .get("promptRouter")?
        .get("invokedModelId")?
        .as_str()
        .and_then(model_id)
}

/// The model of a Converse route's still-percent-encoded `<id>` segment, percent-decoded and
/// reduced to its resource id.
pub fn model_from_segment(segment: &str) -> Option<String> {
    model_id(&percent_decode(segment))
}

/// A model id as recorded: an ARN loses everything but its resource id — the partition, region
/// and AWS account id never reach a record — and an id over [`MAX_MODEL_LEN`] is not recorded.
///
/// `arn:aws:bedrock:us-east-1:123456789012:inference-profile/us.anthropic.claude-…-v1:0` →
/// `us.anthropic.claude-…-v1:0`. A plain id (`anthropic.claude-…-v1:0`, `us.amazon.nova-pro-v1:0`)
/// passes as it is, its inference-profile prefix included.
fn model_id(id: &str) -> Option<String> {
    let id = id.trim();
    let id = match id.strip_prefix("arn:") {
        // partition:service:region:account:resource — the resource may itself hold ':'.
        Some(rest) => {
            let resource = rest.splitn(5, ':').nth(4)?;
            resource.rsplit_once('/').map_or(resource, |(_, tail)| tail)
        }
        None => id,
    };
    (!id.is_empty() && id.len() <= MAX_MODEL_LEN).then(|| id.to_string())
}

/// Decode `%XX` escapes. A `%` not followed by two hex digits is kept as it is.
fn percent_decode(s: &str) -> String {
    let b = s.as_bytes();
    let mut out = Vec::with_capacity(b.len());
    let mut i = 0;
    let hex = |c: u8| (c as char).to_digit(16).map(|d| d as u8);
    while i < b.len() {
        if b[i] == b'%' && i + 2 < b.len() {
            if let (Some(h), Some(l)) = (hex(b[i + 1]), hex(b[i + 2])) {
                out.push(h << 4 | l);
                i += 3;
                continue;
            }
        }
        out.push(b[i]);
        i += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}

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

    fn fixture(name: &str) -> Vec<u8> {
        let p = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("tests/fixtures/model_relay/opencode")
            .join(name);
        std::fs::read(&p).unwrap_or_else(|e| panic!("{}: {e}", p.display()))
    }

    /// Encode one frame — the grammar above, string headers only.
    fn frame(headers: &[(&str, &str)], payload: &[u8]) -> Vec<u8> {
        let mut h = Vec::new();
        for (name, value) in headers {
            h.push(name.len() as u8);
            h.extend_from_slice(name.as_bytes());
            h.push(7);
            h.extend_from_slice(&(value.len() as u16).to_be_bytes());
            h.extend_from_slice(value.as_bytes());
        }
        raw_frame(&h, payload)
    }

    fn raw_frame(headers: &[u8], payload: &[u8]) -> Vec<u8> {
        let total = (16 + headers.len() + payload.len()) as u32;
        let mut f = Vec::new();
        f.extend_from_slice(&total.to_be_bytes());
        f.extend_from_slice(&(headers.len() as u32).to_be_bytes());
        let crc = crc32fast::hash(&f);
        f.extend_from_slice(&crc.to_be_bytes());
        f.extend_from_slice(headers);
        f.extend_from_slice(payload);
        let crc = crc32fast::hash(&f);
        f.extend_from_slice(&crc.to_be_bytes());
        f
    }

    fn event(kind: &str, payload: &str) -> Vec<u8> {
        frame(
            &[
                (":event-type", kind),
                (":content-type", "application/json"),
                (":message-type", "event"),
            ],
            payload.as_bytes(),
        )
    }

    fn decode(bytes: &[u8]) -> Decoder {
        let mut d = Decoder::default();
        d.feed(bytes);
        d
    }

    const STREAM_USAGE: Usage = Usage {
        input_tokens: 25,
        cache_read: 0,
        cache_write: 0,
        eph_5m: 0,
        eph_1h: 0,
        output_tokens: 31,
    };

    /// The recorded stream decodes whole: the `metadata` event's usage, nothing else kept.
    #[test]
    fn bedrock_decode_recorded_stream_yields_metadata_usage() {
        let d = decode(&fixture("bedrock-converse-stream.bin"));
        assert_eq!(d.outcome(), Outcome::Running);
        assert_eq!(d.usage(), Some(STREAM_USAGE));
        assert_eq!(d.model(), None);
    }

    /// AC-14, plan D-7: every recorded stream, split at every byte offset, decodes to the same
    /// usage and outcome as the whole.
    #[test]
    fn bedrock_decode_every_single_split_offset() {
        for name in [
            "bedrock-converse-stream.bin",
            "bedrock-converse-stream-router.bin",
            "bedrock-converse-stream-minimal.bin",
            "bedrock-converse-stream-bad-crc.bin",
            "bedrock-converse-stream-exception.bin",
        ] {
            let bytes = fixture(name);
            let whole = decode(&bytes);
            for cut in 0..=bytes.len() {
                let mut d = Decoder::default();
                d.feed(&bytes[..cut]);
                d.feed(&bytes[cut..]);
                assert_eq!(d.outcome(), whole.outcome(), "{name} @ {cut}");
                assert_eq!(d.usage(), whole.usage(), "{name} @ {cut}");
                assert_eq!(d.model(), whole.model(), "{name} @ {cut}");
            }
        }
    }

    /// Plan D-7: the smallest recorded stream at every PAIR of offsets — three pieces.
    #[test]
    fn bedrock_decode_every_pair_of_split_offsets_on_the_smallest_fixture() {
        let bytes = fixture("bedrock-converse-stream-minimal.bin");
        let whole = decode(&bytes);
        assert!(whole.usage().is_some());
        for a in 0..=bytes.len() {
            for b in a..=bytes.len() {
                let mut d = Decoder::default();
                d.feed(&bytes[..a]);
                d.feed(&bytes[a..b]);
                d.feed(&bytes[b..]);
                assert_eq!(d.usage(), whole.usage(), "@ {a},{b}");
                assert_eq!(d.outcome(), Outcome::Running, "@ {a},{b}");
            }
        }
    }

    /// Byte by byte, too: the degenerate split.
    #[test]
    fn bedrock_decode_byte_at_a_time() {
        let bytes = fixture("bedrock-converse-stream.bin");
        let mut d = Decoder::default();
        for b in &bytes {
            d.feed(std::slice::from_ref(b));
        }
        assert_eq!(d.usage(), Some(STREAM_USAGE));
    }

    /// AC-14: a frame whose message CRC is wrong ends capture `decode_failed`. The fixture's
    /// `metadata` frame is the corrupted one, so no usage is taken from it.
    #[test]
    fn bedrock_decode_corrupted_crc_ends_capture() {
        let d = decode(&fixture("bedrock-converse-stream-bad-crc.bin"));
        assert_eq!(d.outcome(), Outcome::DecodeFailed(Failure::MessageCrc));
        assert_eq!(d.usage(), None);
    }

    /// AC-14: a stream cut inside a frame ends `decode_failed` once it ends — in the prelude, the
    /// headers, the payload or the message CRC — even after an earlier frame carried usage. A
    /// stream that ended between frames is untouched.
    #[test]
    fn bedrock_decode_a_stream_cut_inside_a_frame_finishes_decode_failed() {
        let usage = event(
            "metadata",
            r#"{"usage":{"inputTokens":25,"outputTokens":31,"totalTokens":56}}"#,
        );
        let next = event("contentBlockDelta", r#"{"delta":{"text":"more"}}"#);
        let headers_len = be_u32(&next[4..8]) as usize;
        let payload_end = next.len() - 4;
        // One cut inside each phase of the frame that follows the usage.
        for (phase, cut) in [
            ("prelude", 5),
            ("headers", PRELUDE_LEN + headers_len / 2),
            ("payload", PRELUDE_LEN + headers_len + 1),
            ("message crc", payload_end + 2),
        ] {
            let mut bytes = usage.clone();
            bytes.extend_from_slice(&next[..cut]);
            let mut d = decode(&bytes);
            assert_eq!(d.outcome(), Outcome::Running, "{phase}: still reading");
            d.finish();
            assert_eq!(
                d.outcome(),
                Outcome::DecodeFailed(Failure::Truncated),
                "{phase}"
            );
        }
        // Ended between frames: a clean stream stays clean.
        let mut d = decode(&usage);
        d.finish();
        assert_eq!(d.outcome(), Outcome::Running);
        assert!(d.usage().is_some());
    }

    /// A corrupted prelude CRC fails before any length is believed — even a length that would
    /// otherwise ask for gigabytes.
    #[test]
    fn bedrock_decode_prelude_crc_is_checked_before_the_lengths() {
        let mut f = event("metadata", r#"{"usage":{"inputTokens":1}}"#);
        f[0] = 0xFF; // total_len now ~4 GiB, and the prelude CRC no longer matches
        let d = decode(&f);
        assert_eq!(d.outcome(), Outcome::DecodeFailed(Failure::PreludeCrc));
    }

    /// Prelude with valid CRCs but lengths out of bounds.
    fn prelude(total: u32, headers: u32) -> Vec<u8> {
        let mut p = Vec::new();
        p.extend_from_slice(&total.to_be_bytes());
        p.extend_from_slice(&headers.to_be_bytes());
        let crc = crc32fast::hash(&p);
        p.extend_from_slice(&crc.to_be_bytes());
        p
    }

    #[test]
    fn bedrock_decode_size_checks() {
        for (total, headers, failure) in [
            (15, 0, Failure::FrameLength),
            (MAX_FRAME_LEN + 1, 0, Failure::FrameLength),
            (u32::MAX, 0, Failure::FrameLength),
            (16, 1, Failure::HeadersLength),
            (100, 85, Failure::HeadersLength),
            (MAX_FRAME_LEN, MAX_HEADERS_LEN + 1, Failure::HeadersLength),
        ] {
            let d = decode(&prelude(total, headers));
            assert_eq!(
                d.outcome(),
                Outcome::DecodeFailed(failure),
                "{total}/{headers}"
            );
        }
        // At the bounds: accepted, waiting for the rest.
        for (total, headers) in [(16, 0), (MAX_FRAME_LEN, MAX_HEADERS_LEN), (100, 84)] {
            let d = decode(&prelude(total, headers));
            assert_eq!(d.outcome(), Outcome::Running, "{total}/{headers}");
        }
    }

    #[test]
    fn bedrock_decode_header_checks() {
        // An unknown header type.
        let mut h = vec![1, b'x', 10];
        let d = decode(&raw_frame(&h, b""));
        assert_eq!(
            d.outcome(),
            Outcome::DecodeFailed(Failure::UnknownHeaderType)
        );
        // A string value over 32 767 bytes.
        h = vec![1, b'x', 7];
        h.extend_from_slice(&(MAX_HEADER_VALUE as u16 + 1).to_be_bytes());
        h.extend(std::iter::repeat_n(b'a', MAX_HEADER_VALUE + 1));
        let d = decode(&raw_frame(&h, b""));
        assert_eq!(
            d.outcome(),
            Outcome::DecodeFailed(Failure::HeaderValueTooLong)
        );
        // A header block that ends inside a header.
        h = vec![5, b'x'];
        let d = decode(&raw_frame(&h, b""));
        assert_eq!(d.outcome(), Outcome::DecodeFailed(Failure::MalformedHeader));
        // Every known type, walked and skipped.
        h = Vec::new();
        for (t, len) in [
            (0u8, 0usize),
            (1, 0),
            (2, 1),
            (3, 2),
            (4, 4),
            (5, 8),
            (8, 8),
            (9, 16),
        ] {
            h.extend_from_slice(&[1, b'k', t]);
            h.extend(std::iter::repeat_n(0u8, len));
        }
        h.extend_from_slice(&[1, b'b', 6, 0, 2, 9, 9]);
        let d = decode(&raw_frame(&h, b"ignored"));
        assert_eq!(d.outcome(), Outcome::Running);
        assert_eq!(d.usage(), None);
    }

    /// A `metadata` payload over the 64 KiB cap ends capture rather than growing the buffer.
    #[test]
    fn bedrock_decode_metadata_is_capped() {
        let big = format!(
            r#"{{"usage":{{"inputTokens":1}},"p":"{}"}}"#,
            "a".repeat(MAX_METADATA_PAYLOAD)
        );
        let d = decode(&event("metadata", &big));
        assert_eq!(
            d.outcome(),
            Outcome::DecodeFailed(Failure::MetadataTooLarge)
        );
        // The same size on any other event streams through.
        let d = decode(&event("contentBlockDelta", &big));
        assert_eq!(d.outcome(), Outcome::Running);
    }

    /// AC-14: an exception frame ends capture cleanly — no usage, even from a metadata frame that
    /// comes after it.
    #[test]
    fn bedrock_decode_exception_frame_ends_cleanly() {
        let d = decode(&fixture("bedrock-converse-stream-exception.bin"));
        assert_eq!(d.outcome(), Outcome::Exception);
        assert_eq!(d.usage(), None);
        let mut bytes = fixture("bedrock-converse-stream-exception.bin");
        bytes.extend(event("metadata", r#"{"usage":{"inputTokens":9}}"#));
        let d = decode(&bytes);
        assert_eq!(d.outcome(), Outcome::Exception);
        assert_eq!(d.usage(), None);
    }

    /// The prompt router's invoked model, as a resource id: the account id never survives.
    #[test]
    fn bedrock_decode_router_stream_names_the_invoked_model() {
        let d = decode(&fixture("bedrock-converse-stream-router.bin"));
        assert_eq!(
            d.model(),
            Some("us.anthropic.claude-3-5-sonnet-20240620-v1:0")
        );
        assert!(d.usage().is_some());
    }

    #[test]
    fn bedrock_decode_model_ids() {
        for (segment, model) in [
            (
                "anthropic.claude-3-5-sonnet-20240620-v1%3A0",
                "anthropic.claude-3-5-sonnet-20240620-v1:0",
            ),
            ("us.amazon.nova-pro-v1:0", "us.amazon.nova-pro-v1:0"),
            (
                "arn%3Aaws%3Abedrock%3Aus-east-1%3A123456789012%3Ainference-profile%2Fus.anthropic.claude-3-5-sonnet-20240620-v1%3A0",
                "us.anthropic.claude-3-5-sonnet-20240620-v1:0",
            ),
            (
                "arn%3Aaws%3Abedrock%3Aus-east-1%3A%3Afoundation-model%2Fanthropic.claude-v2",
                "anthropic.claude-v2",
            ),
            (
                "arn%3Aaws-cn%3Abedrock%3Acn-north-1%3A123456789012%3Aapplication-inference-profile%2Fabc123",
                "abc123",
            ),
            ("bad%zzescape", "bad%zzescape"),
            ("trailing%4", "trailing%4"),
        ] {
            let got = model_from_segment(segment);
            assert_eq!(got.as_deref(), Some(model), "{segment}");
            assert!(!got.unwrap_or_default().contains("123456789012"));
        }
        assert_eq!(model_from_segment(""), None);
        assert_eq!(model_from_segment("arn%3Aaws%3Abedrock"), None);
        assert_eq!(model_from_segment(&"m".repeat(MAX_MODEL_LEN + 1)), None);
    }

    #[test]
    fn bedrock_decode_usage_reads_every_count() {
        let u = converse_usage(&serde_json::json!({
            "inputTokens": 10, "outputTokens": 20, "totalTokens": 130,
            "cacheReadInputTokens": 60, "cacheWriteInputTokens": 40
        }));
        assert_eq!(
            u,
            Some(Usage {
                input_tokens: 10,
                cache_read: 60,
                cache_write: 40,
                eph_5m: 0,
                eph_1h: 0,
                output_tokens: 20,
            })
        );
        assert_eq!(converse_usage(&Value::Null), None);
    }

    /// Never panics on noise, at any length.
    #[test]
    fn bedrock_decode_survives_noise() {
        let mut seed: u32 = 0x9E37_79B9;
        for len in 0..512 {
            let bytes: Vec<u8> = (0..len)
                .map(|_| {
                    seed ^= seed << 13;
                    seed ^= seed >> 17;
                    seed ^= seed << 5;
                    seed as u8
                })
                .collect();
            let d = decode(&bytes);
            assert_eq!(d.usage(), None);
        }
    }
}