autumn-web 0.9.0

//! The request-scoped capture buffer and the Tower layer that establishes it.
//!
//! [`CaptureLayer`] sits outer to
//! [`ReportingLayer`](crate::reporting::ReportingLayer) — the failure is not
//! known yet when the request arrives, so every request gets a
//! [`CaptureScope`] and the reporting layer decides at the end whether it is
//! worth writing. A scope is reachable two ways while the handler runs:
//!
//! * through the [`CAPSULE_SCOPE`] task-local, for effect sources deep in the
//!   stack that have no handle to thread (the clock);
//! * through a [`CaptureHandle`] in the request extensions, for the reporting
//!   layer, which must keep the scope alive across a panic unwind.
//!
//! Database recording cannot use the task-local (the pooled connection's I/O
//! runs on its own task), so scopes are additionally published in a
//! weak-reference registry keyed by capsule id; the connection recorder looks
//! its scope up by the id it read off the `SET autumn.capsule_request` marker.

// autumn-panic-gate: request-path module — production code path must be panic-free.
// See CONTRIBUTING.md "Request-path panic gate". Justify exceptions with
// #[allow(clippy::<lint>, reason = "…")] at the narrowest scope.
#![cfg_attr(
    not(test),
    deny(
        clippy::unwrap_used,
        clippy::expect_used,
        clippy::panic,
        clippy::unreachable,
        clippy::todo,
        clippy::unimplemented,
        clippy::indexing_slicing,
        clippy::string_slice,
        clippy::arithmetic_side_effects,
    )
)]

use std::collections::{BTreeMap, HashMap};
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, LazyLock, Mutex, OnceLock, Weak};
use std::task::{Context, Poll};

use axum::body::Body;
use axum::extract::MatchedPath;
use axum::http::{Request, Response};
use chrono::{DateTime, Utc};
use tower::{Layer, Service};

use crate::capsule::redact::{CapturedBody, RawRequest};
use crate::capsule::schema::{
    CacheEffect, CapsuleBody, CapsuleDb, CapsuleEffects, CapsuleJob, ConnectionTape, HttpEffect,
    JobEffect, MailEffect, RandomEffect, TenantEffect,
};
use crate::log::filter::ParameterFilter;

tokio::task_local! {
    /// The capture scope of the request currently being served on this task.
    pub(crate) static CAPSULE_SCOPE: Arc<CaptureScope>;
}

/// The capture scope of the request being served on this task, if any.
#[must_use]
pub fn current_scope() -> Option<Arc<CaptureScope>> {
    CAPSULE_SCOPE.try_with(Arc::clone).ok()
}

/// Run `future` with `scope` established as the current capture scope.
///
/// The task-local itself is crate-private (a request's scope is the framework's
/// bookkeeping, not an extension point); this is the seam integration tests use
/// to drive effect sources that read [`current_scope`].
#[cfg(any(test, feature = "test-support"))]
pub async fn with_capture_scope<F: Future>(scope: Arc<CaptureScope>, future: F) -> F::Output {
    CAPSULE_SCOPE.scope(scope, future).await
}

/// The client identity the trusted-proxies resolver settled on for a request.
///
/// All three fields are recorded so replay can restore the whole
/// `ResolvedClientIdentity` — the address alone is not enough: behind trusted
/// proxies the resolved *public* client IP is itself untrusted, so a replayed
/// resolver run would ignore the recorded forwarded headers and settle on a
/// different host and scheme than the failing request saw.
#[derive(Debug, Clone, Default)]
pub struct CapturedClientIdentity {
    /// Resolved client IP, when one was resolved.
    pub addr: Option<std::net::IpAddr>,
    /// Resolved external host.
    pub host: Option<String>,
    /// Resolved external scheme (`"http"`/`"https"`).
    pub scheme: Option<String>,
}

/// Immutable knobs a scope needs to bound and place its capsule.
#[derive(Debug, Clone)]
pub struct CaptureSettings {
    /// Directory capsules are written to.
    pub dir: String,
    /// Largest request body copied into a capsule.
    pub max_body_bytes: usize,
    /// Size ceiling for recorded effects before a capsule is marked truncated.
    pub max_capsule_bytes: usize,
    /// How many capsules to retain before pruning oldest-first.
    pub max_capsules: usize,
    /// Recording application's name, for cross-build mismatch warnings.
    pub app_name: Option<String>,
    /// Recording application's active profile.
    pub profile: Option<String>,
    /// Database roles the application has configured (`primary`, `replica`),
    /// recorded so a replay can rebuild the same shape even for a request
    /// that never touched the database.
    pub db_roles: Vec<String>,
}

impl Default for CaptureSettings {
    fn default() -> Self {
        Self {
            dir: "tmp/autumn-capsules".to_owned(),
            max_body_bytes: 65_536,
            max_capsule_bytes: 1_048_576,
            max_capsules: 50,
            app_name: None,
            profile: None,
            db_roles: Vec::new(),
        }
    }
}

/// Recorded database traffic for one request, keyed by connection.
///
/// The connection recorder owns the contents; this type only provides the
/// per-request accumulation and the byte budget that stops an unbounded query
/// result from filling memory.
///
/// Tapes are kept in the order the request **first used** each connection, not
/// by connection id. Ids are process-wide birth order and say nothing about
/// this request: a long-lived pooled connection can carry a much lower id than
/// one minted moments ago, so a request that used the fresh connection first
/// would have its tapes listed backwards. Replay hands tape *i* to the *i*-th
/// connection its pool opens (F12), so a reordering there swaps the tapes and
/// makes both connections diverge against traffic that was recorded perfectly.
#[derive(Debug, Default)]
pub struct DbBuffer {
    tapes: BTreeMap<u64, ConnectionTape>,
    /// Connection ids in first-use order — the order [`snapshot`](Self::snapshot)
    /// writes them, and therefore the order replay claims them in.
    order: Vec<u64>,
    bytes: usize,
}

impl DbBuffer {
    /// The tape for a connection, created on first use — and remembered in
    /// `order` at that moment, which is what makes the snapshot first-use
    /// ordered.
    pub fn tape_mut(&mut self, connection_id: u64) -> &mut ConnectionTape {
        match self.tapes.entry(connection_id) {
            std::collections::btree_map::Entry::Occupied(tape) => tape.into_mut(),
            std::collections::btree_map::Entry::Vacant(slot) => {
                self.order.push(connection_id);
                slot.insert(ConnectionTape {
                    id: connection_id,
                    ..ConnectionTape::default()
                })
            }
        }
    }

    /// Charge `bytes` against the capsule budget.
    ///
    /// Returns `false` once the budget is exhausted, at which point the caller
    /// must stop recording and mark the capsule truncated.
    pub const fn charge(&mut self, bytes: usize, budget: usize) -> bool {
        self.bytes = self.bytes.saturating_add(bytes);
        self.bytes <= budget
    }

    /// Bytes charged so far.
    #[must_use]
    pub const fn charged_bytes(&self) -> usize {
        self.bytes
    }

    /// Whether any traffic was recorded.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.tapes.is_empty()
    }

    /// Snapshot the tapes for serialization, in first-use order.
    #[must_use]
    pub fn snapshot(&self) -> Option<CapsuleDb> {
        if self.tapes.is_empty() {
            return None;
        }
        Some(CapsuleDb {
            connections: self
                .order
                .iter()
                .filter_map(|id| self.tapes.get(id))
                .cloned()
                .collect(),
        })
    }
}

/// Most entries one capsule holds for a single effect seam.
///
/// The same reasoning as [`MAX_CLOCK_READINGS`]: a pathological loop calling a
/// third party (or drawing UUIDs) must not grow the buffer without limit, and
/// a capsule that long is not replayable anyway. Crossing the cap marks the
/// capsule truncated, which is already a refusal.
const MAX_EFFECTS_PER_SEAM: usize = 2_000;

/// The framework effects one in-flight run has produced, in call order per
/// seam (#1634).
///
/// Held behind the scope's mutex and only ever touched in short critical
/// sections — never across an `.await` — because every seam that writes here
/// is called from inside async framework code.
#[derive(Debug, Default)]
pub struct EffectBuffer {
    effects: CapsuleEffects,
    /// Bytes charged against `max_capsule_bytes` so far.
    bytes: usize,
    /// Set once any seam hit [`MAX_EFFECTS_PER_SEAM`] or the byte budget, so
    /// the scope can mark the capsule truncated outside the lock.
    overflowed: bool,
    /// Slots handed out by [`reserve`](Self::reserve) that no
    /// [`fill`](Self::fill) has completed yet.
    ///
    /// An effect whose future is *cancelled* between the two — the losing
    /// branch of a `tokio::select!`, a client that hung up — drops its
    /// recorder without filling the slot, leaving the placeholder behind. The
    /// placeholder's error text would then replay as a recorded backend
    /// failure the run never had, which could flip which branch wins on
    /// replay. Counting outstanding reservations lets the snapshot say "this
    /// recording is incomplete" instead.
    outstanding: usize,
}

impl EffectBuffer {
    /// The accumulated effects.
    #[must_use]
    pub const fn effects(&self) -> &CapsuleEffects {
        &self.effects
    }

    /// Whether a seam ran past its cap.
    #[must_use]
    pub const fn overflowed(&self) -> bool {
        self.overflowed
    }

    /// Whether any reserved slot was never completed.
    #[must_use]
    pub const fn has_unfinished(&self) -> bool {
        self.outstanding > 0
    }

    /// Bytes charged so far.
    #[must_use]
    pub const fn charged_bytes(&self) -> usize {
        self.bytes
    }

    /// Reserve a slot on a seam and return its index, so a *concurrent* effect
    /// takes its tape position when it **starts** rather than when it
    /// finishes.
    ///
    /// This is the whole reason the ordered seams are two-phase. A handler that
    /// `join!`s two outbound calls records them in completion order if the
    /// entry is only appended at the end — but replay consumes the tape when
    /// each call *starts*, so the moment the second response beats the first,
    /// an unchanged handler is compared against the wrong tape entry and
    /// reports a divergence that is entirely an artefact of recording.
    ///
    /// `None` when the seam is full or the budget is spent; the caller then has
    /// nothing to fill in and the capsule is already marked truncated.
    fn reserve<T>(
        &mut self,
        seam: fn(&mut CapsuleEffects) -> &mut Vec<T>,
        placeholder: T,
        budget: usize,
    ) -> Option<usize> {
        let list = seam(&mut self.effects);
        if list.len() >= MAX_EFFECTS_PER_SEAM {
            self.overflowed = true;
            return None;
        }
        if self.bytes > budget {
            self.overflowed = true;
            return None;
        }
        let index = list.len();
        list.push(placeholder);
        self.outstanding = self.outstanding.saturating_add(1);
        Some(index)
    }

    /// Fill in a slot [`reserve`](Self::reserve) handed out, charging its
    /// weight now that the effect's real size is known.
    fn fill<T>(
        &mut self,
        seam: fn(&mut CapsuleEffects) -> &mut Vec<T>,
        index: usize,
        entry: T,
        weight: usize,
        budget: usize,
    ) {
        // Answered either way: a slot the budget refuses is already covered by
        // `overflowed`, and counting it as unfinished as well would report one
        // incomplete recording as two separate faults.
        self.outstanding = self.outstanding.saturating_sub(1);
        let charged = self.bytes.saturating_add(weight);
        if charged > budget {
            self.overflowed = true;
            return;
        }
        self.bytes = charged;
        if let Some(slot) = seam(&mut self.effects).get_mut(index) {
            *slot = entry;
        }
    }

    /// Push onto a seam, refusing (and flagging) once it is full.
    ///
    /// `weight` is the entry's approximate serialized size, charged against
    /// the capsule's `max_capsule_bytes` budget. Two bounds rather than one
    /// because they catch different pathologies: the count cap stops a loop
    /// making ten thousand tiny calls, and the byte budget stops a single
    /// handler streaming a hundred megabytes through the framework's HTTP
    /// client into the capsule buffer.
    fn push<T>(
        &mut self,
        seam: fn(&mut CapsuleEffects) -> &mut Vec<T>,
        entry: T,
        weight: usize,
        budget: usize,
    ) {
        let list = seam(&mut self.effects);
        if list.len() >= MAX_EFFECTS_PER_SEAM {
            self.overflowed = true;
            return;
        }
        // Charged only for an entry that is actually kept, so `charged_bytes`
        // means what it says.
        let charged = self.bytes.saturating_add(weight);
        if charged > budget {
            self.overflowed = true;
            return;
        }
        self.bytes = charged;
        let list = seam(&mut self.effects);
        list.push(entry);
    }
}

/// The approximate serialized size of a JSON payload, for the effect budget.
///
/// Serializing it just to measure would double the cost of every enqueue on a
/// capture-enabled request; the string length of a compact rendering is close
/// enough for a budget whose job is to stop unbounded growth.
fn json_weight(payload: &serde_json::Value) -> usize {
    /// A sink that counts bytes and keeps none of them.
    struct Counter(usize);
    impl std::io::Write for Counter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0 = self.0.saturating_add(buf.len());
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    let mut counter = Counter(0);
    let _ = serde_json::to_writer(&mut counter, payload);
    counter.0
}

/// The approximate serialized size of a recorded body, for the effect budget.
const fn body_weight(body: &CapsuleBody) -> usize {
    match body {
        CapsuleBody::Absent | CapsuleBody::Skipped { .. } => 0,
        CapsuleBody::Text(text) => text.len(),
        CapsuleBody::Base64(encoded) => encoded.len(),
    }
}

/// Clamp a recorded effect body to `max_body_bytes`.
///
/// A body over the cap becomes a [`CapsuleBody::Skipped`] carrying its real
/// length, rather than a silent truncation: half a JSON document replayed as
/// though it were whole would drive the handler with input the failing run
/// never had, which is the same falsification the *request* body cap already
/// refuses to commit.
fn clamp_body(body: CapsuleBody, max_body_bytes: usize) -> CapsuleBody {
    let len = body_weight(&body);
    if len <= max_body_bytes {
        return body;
    }
    CapsuleBody::Skipped {
        declared_len: Some(len),
    }
}

/// Most clock readings one capsule will hold.
const MAX_CLOCK_READINGS: usize = 10_000;

/// How the capture layer is treating one request's body.
///
/// The layer never reads the body itself (see [`TeeBody`]), so this is the
/// running state of a copy the *handler* is driving.
#[derive(Debug, Default)]
enum BodyTap {
    /// Nothing to copy: no body was declared, or it was declared empty.
    #[default]
    Absent,
    /// Deliberately not copied — the declared length was over the cap, so the
    /// body streams to the handler untouched.
    Skipped {
        /// Length the client declared, when it declared one.
        declared_len: Option<usize>,
    },
    /// Being copied frame by frame as the handler reads it.
    Teeing {
        /// Length the client declared, when it declared one.
        declared_len: Option<usize>,
        /// Bytes copied so far.
        buf: Vec<u8>,
        /// Whether the handler read the body all the way to its end.
        end_stream: bool,
        /// Whether the copy was abandoned for exceeding `max_body_bytes`.
        overflowed: bool,
    },
}

/// Note recorded when a streaming body grew past `max_body_bytes`.
const BODY_OVERFLOW_NOTE: &str =
    "request body exceeded max_body_bytes while streaming; it was not captured";

/// The approximate serialized size of a recorded header list.
fn headers_weight(headers: &[(String, String)]) -> usize {
    headers.iter().fold(0, |total, (name, value)| {
        total.saturating_add(name.len()).saturating_add(value.len())
    })
}

/// Note recorded when an outbound body was too large to keep.
///
/// The capsule is marked truncated alongside it: replay serves a skipped body
/// as empty bytes, and a handler that parses the response it recorded would
/// then be judged on input the failing run never had — the same falsification
/// an unrecorded *request* body already refuses to commit.
const HTTP_BODY_SKIPPED_NOTE: &str = "an outbound HTTP body exceeded `[failure_capture] max_body_bytes` and was not captured; \
     replaying would drive the handler with an empty body";

/// Note recorded when a mail body was too large to keep.
///
/// Same reasoning as [`HTTP_BODY_SKIPPED_NOTE`]: the replay comparison treats a
/// skipped body as matching anything, so a capsule carrying one can only be
/// honest by declaring itself incomplete.
const MAIL_BODY_SKIPPED_NOTE: &str = "a mail body exceeded `[failure_capture] max_body_bytes` and was not captured; replay \
     cannot tell whether the message contents still match";

/// Note recorded when an effect was reserved but never completed.
///
/// A seam reserves its tape position when the effect *starts* and fills it in
/// when the effect finishes. Between the two the future can be dropped — the
/// losing branch of a `tokio::select!`, a timeout, a client that hung up — and
/// then nothing ever fills the slot. Persisting the placeholder as if it were
/// a recorded outcome would hand replay a backend failure the run never had,
/// and for a `select!` that is enough to change which branch wins. So the
/// capsule says it is incomplete instead.
const UNFINISHED_EFFECT_NOTE: &str = "an effect was still in flight when the recording ended (a cancelled future — a losing \
     `tokio::select!` branch, a timeout), so its outcome was never recorded";

/// Note recorded when a run resolved two different tenants.
///
/// The capsule holds one tenant context, so a run that resolved two cannot be
/// replayed faithfully: replay would install one of them and a handler that
/// switched tenants mid-run would read the wrong one.
const TENANT_CONFLICT_NOTE: &str = "the run resolved more than one tenant; a capsule records a single tenant context, so this \
     one cannot reproduce the tenant the failing run saw";

/// Note recorded when the handler stopped reading the body before its end.
const BODY_PARTIAL_NOTE: &str =
    "request body was not read to its end before the failure; the captured body is incomplete";

/// Everything one in-flight request has offered up for its capsule.
#[derive(Debug)]
pub struct CaptureScope {
    id: String,
    settings: Arc<CaptureSettings>,
    filter: Arc<ParameterFilter>,
    request: OnceLock<RawRequest>,
    body: Mutex<BodyTap>,
    clock: Mutex<Vec<DateTime<Utc>>>,
    /// Monotonic readings, as offsets from the recording clock's origin.
    monotonic: Mutex<Vec<std::time::Duration>>,
    client_identity: OnceLock<CapturedClientIdentity>,
    /// Set when this scope wraps a *job execution* rather than a request.
    job: OnceLock<CapsuleJob>,
    /// The raw peer socket (`ConnectInfo`), before trusted-proxy resolution.
    peer_addr: OnceLock<std::net::SocketAddr>,
    db: Mutex<DbBuffer>,
    effects: Mutex<EffectBuffer>,
    notes: Mutex<Vec<String>>,
    truncated: AtomicBool,
    closed: AtomicBool,
    /// Whether the app state had a cache when the run started.
    state_cache: AtomicBool,
    /// `register_after_commit` callbacks that a rollback has not dropped.
    after_commit: std::sync::atomic::AtomicUsize,
}

impl CaptureScope {
    /// Create a scope for a request.
    #[must_use]
    pub fn new(id: String, settings: Arc<CaptureSettings>, filter: Arc<ParameterFilter>) -> Self {
        Self {
            id,
            settings,
            filter,
            request: OnceLock::new(),
            body: Mutex::new(BodyTap::Absent),
            clock: Mutex::new(Vec::new()),
            monotonic: Mutex::new(Vec::new()),
            client_identity: OnceLock::new(),
            job: OnceLock::new(),
            peer_addr: OnceLock::new(),
            db: Mutex::new(DbBuffer::default()),
            effects: Mutex::new(EffectBuffer::default()),
            notes: Mutex::new(Vec::new()),
            truncated: AtomicBool::new(false),
            closed: AtomicBool::new(false),
            state_cache: AtomicBool::new(false),
            after_commit: std::sync::atomic::AtomicUsize::new(0),
        }
    }

    /// Note whether the app state has a cache. Any `true` is kept.
    pub(crate) fn note_state_cache(&self, present: bool) {
        if present {
            self.state_cache.store(true, Ordering::Relaxed);
        }
    }

    /// Whether the app state had a cache when the run started.
    #[must_use]
    pub(crate) fn had_state_cache(&self) -> bool {
        self.state_cache.load(Ordering::Relaxed)
    }

    /// The capsule id (the request id, when one was available).
    #[must_use]
    pub fn id(&self) -> &str {
        &self.id
    }

    /// The knobs this scope was built with.
    #[must_use]
    pub fn settings(&self) -> &CaptureSettings {
        &self.settings
    }

    /// The redaction filter this scope's capsule must be written through.
    #[must_use]
    pub fn filter(&self) -> &ParameterFilter {
        &self.filter
    }

    /// Record the unredacted request snapshot. Only the first call takes.
    pub fn set_request(&self, request: RawRequest) {
        let _ = self.request.set(request);
    }

    /// The unredacted request snapshot, if the layer recorded one.
    ///
    /// The snapshot is the request *head* only; the body arrives separately
    /// through [`captured_body`](Self::captured_body), because it is copied
    /// while the handler reads it rather than up front.
    #[must_use]
    pub fn raw_request(&self) -> Option<&RawRequest> {
        self.request.get()
    }

    /// Decide how this request's body will be treated, before the handler runs.
    fn arm_body(&self, tap: BodyTap) {
        if let Ok(mut current) = self.body.lock() {
            *current = tap;
        }
    }

    /// Copy a data frame the handler has just read.
    ///
    /// Bounded by `max_body_bytes`: the frame that would cross the cap ends
    /// the copy and releases what was collected, so an unexpectedly large
    /// streamed upload cannot be buffered in memory by the mere presence of
    /// capture.
    fn tee_body_chunk(&self, chunk: &[u8]) {
        let limit = self.settings.max_body_bytes;
        if let Ok(mut tap) = self.body.lock()
            && let BodyTap::Teeing {
                buf, overflowed, ..
            } = &mut *tap
            && !*overflowed
        {
            if buf.len().saturating_add(chunk.len()) > limit {
                *overflowed = true;
                *buf = Vec::new();
            } else {
                buf.extend_from_slice(chunk);
            }
        }
    }

    /// Record that the handler read the body all the way to its end.
    fn mark_body_end(&self) {
        if let Ok(mut tap) = self.body.lock()
            && let BodyTap::Teeing { end_stream, .. } = &mut *tap
        {
            *end_stream = true;
        }
    }

    /// The request body, as far as the handler read it before the failure.
    #[must_use]
    pub fn captured_body(&self) -> CapturedBody {
        let Ok(tap) = self.body.lock() else {
            // A poisoned lock means a body copy was interrupted mid-write.
            // Reporting "no body" as though the request had none would be a
            // falsification, so the capsule is marked incomplete instead.
            self.mark_truncated();
            return CapturedBody::Absent;
        };
        match &*tap {
            BodyTap::Absent => CapturedBody::Absent,
            // Declared over the cap up front, or discovered to be over it
            // mid-stream: either way the capsule records a skip, not a body.
            BodyTap::Skipped { declared_len }
            | BodyTap::Teeing {
                declared_len,
                overflowed: true,
                ..
            } => CapturedBody::Skipped {
                declared_len: *declared_len,
            },
            BodyTap::Teeing { buf, .. } if buf.is_empty() => CapturedBody::Absent,
            BodyTap::Teeing { buf, .. } => CapturedBody::Buffered(bytes::Bytes::from(buf.clone())),
        }
    }

    /// A note explaining a body the capsule reader should not trust as
    /// complete, if this request produced one.
    #[must_use]
    pub fn body_note(&self) -> Option<&'static str> {
        let tap = self.body.lock().ok()?;
        match &*tap {
            BodyTap::Teeing {
                overflowed: true, ..
            } => Some(BODY_OVERFLOW_NOTE),
            // A handler that read exactly `Content-Length` bytes and stopped
            // has the whole body — it simply never polled once more for the
            // `None` that sets `end_stream`. The captured length settles it,
            // and treating that as partial would refuse a capsule whose body
            // is complete (replay refuses partial bodies, so a false positive
            // here costs a perfectly good reproduction).
            BodyTap::Teeing {
                declared_len: Some(declared),
                buf,
                end_stream: false,
                ..
            } if buf.len() >= *declared => None,
            BodyTap::Teeing {
                end_stream: false, ..
            } => Some(BODY_PARTIAL_NOTE),
            _ => None,
        }
    }

    /// Append a clock reading.
    ///
    /// Record the client identity the trusted-proxies resolver settled on.
    /// Only the first call takes — one resolution per request.
    pub fn set_client_identity(&self, identity: CapturedClientIdentity) {
        let _ = self.client_identity.set(identity);
    }

    /// The resolved client identity, when the resolver ran under this scope.
    #[must_use]
    pub fn client_identity(&self) -> Option<&CapturedClientIdentity> {
        self.client_identity.get()
    }

    /// Record the raw peer socket the request arrived on.
    pub fn set_peer_addr(&self, peer: std::net::SocketAddr) {
        let _ = self.peer_addr.set(peer);
    }

    /// The raw peer socket, when the server had one to give.
    #[must_use]
    pub fn peer_addr(&self) -> Option<std::net::SocketAddr> {
        self.peer_addr.get().copied()
    }

    /// Bounded: a pathological loop reading `now()` must not grow the buffer
    /// without limit, and a capsule that long is not replayable anyway.
    pub fn record_clock(&self, reading: DateTime<Utc>) {
        if let Ok(mut readings) = self.clock.lock() {
            if readings.len() >= MAX_CLOCK_READINGS {
                self.truncated.store(true, Ordering::Relaxed);
                return;
            }
            readings.push(reading);
        }
    }

    /// The clock readings taken during the request, in order.
    #[must_use]
    pub fn clock_readings(&self) -> Vec<DateTime<Utc>> {
        self.clock
            .lock()
            .map(|readings| readings.clone())
            .unwrap_or_default()
    }

    /// Record a [`ClockSource::monotonic`](crate::time::ClockSource::monotonic)
    /// reading, as its offset from the recording clock's origin. Bounded like
    /// [`record_clock`](Self::record_clock), for the same reason.
    pub fn record_monotonic(&self, since_origin: std::time::Duration) {
        if let Ok(mut readings) = self.monotonic.lock() {
            if readings.len() >= MAX_CLOCK_READINGS {
                self.truncated.store(true, Ordering::Relaxed);
                return;
            }
            readings.push(since_origin);
        }
    }

    /// The monotonic readings taken during the request, in order.
    #[must_use]
    pub fn monotonic_readings(&self) -> Vec<std::time::Duration> {
        self.monotonic
            .lock()
            .map(|readings| readings.clone())
            .unwrap_or_default()
    }

    /// Operate on the recorded database traffic.
    pub fn with_db<R>(&self, f: impl FnOnce(&mut DbBuffer) -> R) -> Option<R> {
        self.db.lock().ok().map(|mut db| f(&mut db))
    }

    /// Snapshot the recorded database traffic for serialization.
    ///
    /// A poisoned buffer lock yields no tape *and* marks the capsule
    /// truncated: "this request did no database work" and "the recorded
    /// database work is unreachable" must not look the same to replay.
    #[must_use]
    pub fn db_snapshot(&self) -> Option<CapsuleDb> {
        self.db.lock().map_or_else(
            |_| {
                self.mark_truncated();
                None
            },
            |db| db.snapshot(),
        )
    }

    /// Mark this scope as recording a *job execution* rather than a request.
    ///
    /// Only the first call takes: one entry point per capsule.
    pub fn set_job_entry(&self, job: CapsuleJob) {
        let _ = self.job.set(job);
    }

    /// The job this scope is recording, when it is a job-scoped capsule.
    #[must_use]
    pub fn job_entry(&self) -> Option<CapsuleJob> {
        self.job.get().cloned()
    }

    // ── Effect seams (#1634) ────────────────────────────────────────────

    /// Run `f` against the effect buffer, marking the capsule truncated if a
    /// seam has overflowed.
    ///
    /// Every seam recorder goes through here so the overflow flag is checked
    /// in exactly one place, and so the truncation store happens *after* the
    /// lock is released rather than inside the critical section.
    fn with_effects<R>(&self, f: impl FnOnce(&mut EffectBuffer) -> R) -> Option<R> {
        let (result, overflowed) = {
            let mut buffer = self.effects.lock().ok()?;
            let result = f(&mut buffer);
            (result, buffer.overflowed())
        };
        if overflowed {
            self.mark_truncated();
        }
        Some(result)
    }

    /// Reserve this run's next outbound-HTTP tape position, before the call is
    /// made.
    ///
    /// Returns the index [`fill_http`](Self::fill_http) completes. See
    /// [`EffectBuffer::reserve`] for why the position is taken at initiation.
    #[must_use]
    pub fn reserve_http(&self) -> Option<usize> {
        let budget = self.settings.max_capsule_bytes;
        self.with_effects(|buffer| {
            buffer.reserve(|effects| &mut effects.http, HttpEffect::pending(), budget)
        })
        .flatten()
    }

    /// Complete a reserved outbound-HTTP slot.
    ///
    /// Bodies over `max_body_bytes` are recorded as skipped rather than
    /// copied: a capsule must not become the place a large download is
    /// buffered. A skipped body also **marks the capsule truncated**, because
    /// a replayed handler that parses that response would be driven with an
    /// empty body the recording never gave it.
    pub fn fill_http(&self, index: usize, mut effect: HttpEffect) {
        let cap = self.settings.max_body_bytes;
        // An already-`Skipped` body counts. `http_client::encode_body` drops an
        // oversized body *before* it reaches here — deliberately, so a 50 MB
        // download is never copied just to be thrown away — and
        // `body_weight(Skipped)` is zero, so a weight comparison alone would
        // never see the one case this check exists for.
        let skipped = |body: &CapsuleBody| {
            matches!(body, CapsuleBody::Skipped { .. }) || body_weight(body) > cap
        };
        let request_over = skipped(&effect.request_body);
        let response_over = skipped(&effect.response_body);
        effect.request_body = clamp_body(effect.request_body, cap);
        effect.response_body = clamp_body(effect.response_body, cap);
        if request_over || response_over {
            self.note(HTTP_BODY_SKIPPED_NOTE);
            self.mark_truncated();
        }
        // Every field the effect *retains*, not just the obvious two: a handler
        // sending or receiving large headers would otherwise grow the buffer
        // and the persisted capsule past `max_capsule_bytes` without ever
        // tripping the bound that exists to stop exactly that.
        let weight = effect
            .url
            .len()
            .saturating_add(effect.final_url.as_ref().map_or(0, String::len))
            .saturating_add(headers_weight(&effect.request_headers))
            .saturating_add(headers_weight(&effect.response_headers))
            .saturating_add(body_weight(&effect.request_body))
            .saturating_add(body_weight(&effect.response_body));
        let budget = self.settings.max_capsule_bytes;
        let _ = self.with_effects(|buffer| {
            buffer.fill(|effects| &mut effects.http, index, effect, weight, budget);
        });
    }

    /// Reserve this run's next job-enqueue tape position, before the backend
    /// is asked.
    #[must_use]
    pub fn reserve_job_enqueue(&self) -> Option<usize> {
        let budget = self.settings.max_capsule_bytes;
        self.with_effects(|buffer| {
            buffer.reserve(|effects| &mut effects.jobs, JobEffect::pending(), budget)
        })
        .flatten()
    }

    /// Complete a reserved job-enqueue slot with the backend's outcome.
    pub fn fill_job_enqueue(&self, index: usize, effect: JobEffect) {
        // Every retained field is charged, the error text included: replay
        // hands it back verbatim, so it is kept whole (#2351 item 19).
        let weight = effect
            .name
            .len()
            .saturating_add(json_weight(&effect.payload))
            .saturating_add(effect.error.as_ref().map_or(0, String::len));
        let budget = self.settings.max_capsule_bytes;
        let _ = self.with_effects(|buffer| {
            buffer.fill(|effects| &mut effects.jobs, index, effect, weight, budget);
        });
    }

    /// Reserve this run's next mail tape position, before the send is made.
    #[must_use]
    pub fn reserve_mail(&self) -> Option<usize> {
        let budget = self.settings.max_capsule_bytes;
        self.with_effects(|buffer| {
            buffer.reserve(|effects| &mut effects.mail, MailEffect::pending(), budget)
        })
        .flatten()
    }

    /// Complete a reserved mail slot.
    pub fn fill_mail(&self, index: usize, mut effect: MailEffect) {
        let cap = self.settings.max_body_bytes;
        // Both halves of a multipart message have the same cap (#2351 item
        // 10).
        let over = body_weight(&effect.body) > cap || body_weight(&effect.alternate_body) > cap;
        effect.body = clamp_body(effect.body, cap);
        effect.alternate_body = clamp_body(effect.alternate_body, cap);
        // A skipped body is a wildcard to the replay comparison — it has to be,
        // there being nothing recorded to compare against — so a capsule
        // holding one must not present as complete. Otherwise the message
        // contents could change freely and still replay `reproduced`, which is
        // the outcome comparing the body at all was meant to prevent.
        if over {
            self.note(MAIL_BODY_SKIPPED_NOTE);
            self.mark_truncated();
        }
        // Every retained field is charged against the capsule budget.
        let weight = effect
            .subject
            .len()
            .saturating_add(body_weight(&effect.body))
            .saturating_add(body_weight(&effect.alternate_body))
            .saturating_add(headers_weight(&effect.extra_headers))
            .saturating_add(
                effect
                    .to
                    .iter()
                    .fold(0, |total: usize, to| total.saturating_add(to.len())),
            )
            .saturating_add(effect.from.as_ref().map_or(0, String::len))
            .saturating_add(effect.reply_to.as_ref().map_or(0, String::len))
            .saturating_add(effect.list_unsubscribe.as_ref().map_or(0, String::len))
            .saturating_add(effect.error.as_ref().map_or(0, String::len))
            .saturating_add(
                effect
                    .attachments
                    .iter()
                    .fold(0, |total: usize, attachment| {
                        total
                            .saturating_add(attachment.filename.len())
                            .saturating_add(attachment.content_type.len())
                            .saturating_add(attachment.sha256.len())
                    }),
            );
        let budget = self.settings.max_capsule_bytes;
        let _ = self.with_effects(|buffer| {
            buffer.fill(|effects| &mut effects.mail, index, effect, weight, budget);
        });
    }

    /// Reserve this run's next cache tape position, before an async removal
    /// starts.
    #[must_use]
    pub fn reserve_cache(&self) -> Option<usize> {
        let budget = self.settings.max_capsule_bytes;
        self.with_effects(|buffer| {
            buffer.reserve(|effects| &mut effects.cache, CacheEffect::pending(), budget)
        })
        .flatten()
    }

    /// Complete a reserved cache slot.
    pub fn fill_cache(&self, index: usize, effect: CacheEffect) {
        let weight = cache_weight(&effect);
        let budget = self.settings.max_capsule_bytes;
        let _ = self.with_effects(|buffer| {
            buffer.fill(|effects| &mut effects.cache, index, effect, weight, budget);
        });
    }

    /// Record one cache read or write.
    pub fn record_cache(&self, effect: CacheEffect) {
        let weight = cache_weight(&effect);
        let budget = self.settings.max_capsule_bytes;
        let _ = self.with_effects(|buffer| {
            buffer.push(|effects| &mut effects.cache, effect, weight, budget);
        });
    }

    /// Record the tenant context the run resolved. Later resolutions of the
    /// *same* tenant are idempotent; a different one is a second resolution
    /// the capsule cannot represent, so the capsule is marked truncated rather
    /// than silently recording only the first.
    pub fn record_tenant(&self, effect: TenantEffect) {
        let conflicting = self
            .with_effects(|buffer| match &buffer.effects.tenant {
                Some(existing) if *existing == effect => false,
                Some(_) => true,
                None => {
                    buffer.effects.tenant = Some(effect);
                    false
                }
            })
            .unwrap_or(false);
        if conflicting {
            self.note(TENANT_CONFLICT_NOTE);
            self.mark_truncated();
        }
    }

    /// Record one draw from the entropy source.
    pub fn record_random(&self, bytes: Vec<u8>) {
        let weight = bytes.len();
        let budget = self.settings.max_capsule_bytes;
        let _ = self.with_effects(|buffer| {
            buffer.push(
                |effects| &mut effects.random,
                RandomEffect { bytes },
                weight,
                budget,
            );
        });
    }

    /// Snapshot the recorded effects for serialization.
    ///
    /// A poisoned buffer marks the capsule truncated for the same reason
    /// [`db_snapshot`](Self::db_snapshot) does: "this run produced no effects"
    /// and "the recorded effects are unreachable" must not look the same to
    /// replay.
    #[must_use]
    pub fn effects_snapshot(&self) -> CapsuleEffects {
        let (effects, unfinished) = self.effects.lock().map_or_else(
            |_| {
                self.mark_truncated();
                (CapsuleEffects::default(), false)
            },
            |buffer| (buffer.effects().clone(), buffer.has_unfinished()),
        );
        if unfinished {
            self.note(UNFINISHED_EFFECT_NOTE);
            self.mark_truncated();
        }
        effects
    }

    /// Note a degraded-capture condition for the capsule reader.
    pub fn note(&self, note: impl Into<String>) {
        let note = note.into();
        if let Ok(mut notes) = self.notes.lock()
            && !notes.contains(&note)
        {
            notes.push(note);
        }
    }

    /// The accumulated notes.
    #[must_use]
    pub fn notes(&self) -> Vec<String> {
        self.notes
            .lock()
            .map(|notes| notes.clone())
            .unwrap_or_default()
    }

    /// Stop accepting effects: the request this scope belongs to is over.
    ///
    /// Called when the capture layer's future resolves — normally or through a
    /// panic unwind. The scope itself lives on until the capsule is written,
    /// so this is what stops *late* effects from joining it. Chiefly the
    /// connection pool's liveness check: `pool.get()` pings the connection
    /// before [`Db::checkout`](crate::db::Db::checkout) sends the next
    /// request's attribution marker, so without a close the ping would be
    /// recorded against whoever held that connection last, and replay of that
    /// capsule would then expect a query its handler never issued (F2).
    /// Release/Acquire rather than Relaxed: closing publishes everything the
    /// request recorded, and a connection recorder on another thread that
    /// observes the close must also observe those writes.
    pub fn close(&self) {
        self.closed.store(true, Ordering::Release);
    }

    /// Whether the request is over and the capsule is no longer accepting
    /// effects.
    #[must_use]
    pub fn is_closed(&self) -> bool {
        self.closed.load(Ordering::Acquire)
    }

    /// Count a `register_after_commit` callback (#2351 item 3).
    pub(crate) fn after_commit_registered(&self) {
        self.after_commit.fetch_add(1, Ordering::SeqCst);
    }

    /// A rollback dropped a counted callback. Each drop follows its own
    /// registration, so the count does not go below zero.
    pub(crate) fn after_commit_dropped(&self) {
        self.after_commit.fetch_sub(1, Ordering::SeqCst);
    }

    /// Whether a counted callback is still pending or ran.
    pub(crate) fn after_commit_pending(&self) -> bool {
        self.after_commit.load(Ordering::SeqCst) > 0
    }

    /// Mark the capsule as incomplete; replay must refuse it.
    pub fn mark_truncated(&self) {
        self.truncated.store(true, Ordering::Relaxed);
    }

    /// Whether a size cap stopped recording partway through.
    #[must_use]
    pub fn is_truncated(&self) -> bool {
        self.truncated.load(Ordering::Relaxed)
    }
}

/// The approximate serialized size of a cache effect.
fn cache_weight(effect: &CacheEffect) -> usize {
    effect.key().len().saturating_add(match effect {
        CacheEffect::Get { value, .. } => value.as_ref().map_or(0, String::len),
        CacheEffect::Insert { value, .. } => value.len(),
        CacheEffect::Invalidate { error, .. } | CacheEffect::InvalidateNamespace { error, .. } => {
            error.as_ref().map_or(0, |error| error.reason.len())
        }
        CacheEffect::Clear => 0,
    })
}

/// A cloneable handle to a request's [`CaptureScope`], carried in the request
/// extensions so the reporting layer can reach it after an unwind.
#[derive(Clone, Debug)]
pub struct CaptureHandle(Arc<CaptureScope>);

impl CaptureHandle {
    /// The scope this handle keeps alive.
    #[must_use]
    pub const fn scope(&self) -> &Arc<CaptureScope> {
        &self.0
    }
}

// ── Job-scoped capsules (#1634) ─────────────────────────────────────────────

/// Run a job execution under its own capture scope, writing a capsule when it
/// fails.
///
/// A job is a second entry point into the application, and failures in one are
/// exactly as hard to reproduce as failures in a request — often harder,
/// because nobody watched it happen. So a job gets the same treatment: its own
/// scope, so the clock, entropy, database and effect seams all record against
/// it, and a capsule on failure that `autumn replay` and a generated
/// regression test can drive.
///
/// The synthetic request record (`JOB /jobs/<name>`) is what lets every field
/// that names "the recorded entry point" keep resolving; the `job` field is
/// what tells replay to dispatch the job rather than the router.
///
/// `outcome` maps the job's result to a capsule outcome, returning `None` for
/// a success — there is nothing to capture then, and the scope is dropped.
#[cfg(feature = "reporting")]
pub async fn capture_job<T>(
    name: &str,
    payload: &serde_json::Value,
    settings: Arc<CaptureSettings>,
    filter: Arc<ParameterFilter>,
    run: impl Future<Output = T>,
    outcome: impl FnOnce(&T) -> Option<crate::capsule::schema::CapsuleOutcome>,
) -> T {
    let id = job_scope_id();
    let scope = Arc::new(CaptureScope::new(id, settings, filter));
    scope.set_job_entry(CapsuleJob {
        name: name.to_owned(),
        payload: payload.clone(),
    });
    // A synthetic head, so the capsule's `request` describes the entry point
    // the way a request capsule's does. `JOB` is deliberately not a real HTTP
    // method: a reader (and `autumn replay`) must not mistake this for a
    // request that can be driven through the router.
    scope.set_request(RawRequest {
        method: "JOB".to_owned(),
        uri: format!("/jobs/{name}")
            .parse()
            .unwrap_or_else(|_| axum::http::Uri::from_static("/jobs")),
        version: axum::http::Version::HTTP_11,
        headers: axum::http::HeaderMap::new(),
        // Braces are the route-template syntax, not a format placeholder.
        #[allow(
            clippy::literal_string_with_formatting_args,
            reason = "`{name}` is Autumn's route-template syntax, not an interpolation"
        )]
        route: Some("/jobs/{name}".to_owned()),
    });
    register(&scope);
    let guard = RegistryGuard(Arc::clone(&scope));
    let result = CAPSULE_SCOPE.scope(Arc::clone(&scope), run).await;
    drop(guard);

    if let Some(outcome) = outcome(&result) {
        // Persisting is blocking (a directory scan and a file write), and this
        // runs on a worker thread serving other jobs.
        let scope = Arc::clone(&scope);
        let _ = crate::time::spawn_blocking(move || {
            let _ = crate::capsule::persist(&scope, outcome);
        })
        .await;
    }
    result
}

/// A capsule id for a job execution.
///
/// Jobs carry no request id, so one is minted here — through the same
/// character set [`is_valid_scope_id`] accepts, because it is interpolated
/// into the database attribution marker exactly as a request id is.
#[cfg(feature = "reporting")]
fn job_scope_id() -> String {
    // Drawn through the framework's entropy seam rather than `Uuid::new_v4`,
    // so a simulation with a seeded source gets reproducible capsule ids too.
    use crate::entropy::Entropy as _;
    format!("job-{}", crate::entropy::OsEntropy.uuid_v4().simple())
}

// ── Registry ────────────────────────────────────────────────────────────────

/// Live scopes by capsule id, weakly held so a finished request's scope is
/// freed even if deregistration is skipped.
static REGISTRY: LazyLock<Mutex<HashMap<String, Weak<CaptureScope>>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));

/// Look a live scope up by the capsule id a connection marker carried.
#[must_use]
pub fn scope_by_id(id: &str) -> Option<Arc<CaptureScope>> {
    REGISTRY
        .lock()
        .ok()
        .and_then(|registry| registry.get(id).and_then(Weak::upgrade))
}

pub(crate) fn register(scope: &Arc<CaptureScope>) {
    if let Ok(mut registry) = REGISTRY.lock() {
        registry.insert(scope.id().to_owned(), Arc::downgrade(scope));
    }
}

fn deregister(id: &str) {
    if let Ok(mut registry) = REGISTRY.lock() {
        registry.remove(id);
    }
}

/// Closes a scope and removes it from the registry when the request's future
/// is dropped, including when it is dropped by a panic unwind.
struct RegistryGuard(Arc<CaptureScope>);

impl Drop for RegistryGuard {
    fn drop(&mut self) {
        // Order matters: closing first means a connection recorder that is
        // mid-append when the request ends cannot slip an effect in between
        // the two steps.
        self.0.close();
        deregister(self.0.id());
    }
}

// ── Scope id ────────────────────────────────────────────────────────────────

/// Longest capsule id accepted; the id is interpolated into the `SET
/// autumn.capsule_request` marker, so it is length- and charset-bounded.
const MAX_SCOPE_ID_LEN: usize = 64;

/// Whether an id is safe to interpolate into the connection marker SQL.
#[must_use]
pub fn is_valid_scope_id(id: &str) -> bool {
    !id.is_empty()
        && id.len() <= MAX_SCOPE_ID_LEN
        && id
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
}

// ── Tower layer ─────────────────────────────────────────────────────────────

/// Tower [`Layer`] that establishes a [`CaptureScope`] for every request.
///
/// Installed only when `[failure_capture] enabled = true`, immediately outer to
/// [`ReportingLayer`](crate::reporting::ReportingLayer) so a scope exists
/// before the reporting layer snapshots its request context.
#[derive(Clone)]
pub struct CaptureLayer {
    settings: Arc<CaptureSettings>,
    filter: Arc<ParameterFilter>,
    state_cache: Option<StateCacheProbe>,
}

/// Tells whether the app state has a cache.
type StateCacheProbe = Arc<dyn Fn() -> bool + Send + Sync>;

impl CaptureLayer {
    /// Build the layer from resolved settings and the shared redaction filter.
    #[must_use]
    pub fn new(settings: CaptureSettings, filter: Arc<ParameterFilter>) -> Self {
        Self {
            settings: Arc::new(settings),
            filter,
            state_cache: None,
        }
    }

    /// Read at each request start whether the app builder installed a cache.
    /// Code can resolve the cache when the app is built, before a request
    /// scope exists, so the request alone cannot tell (#2351).
    #[must_use]
    pub(crate) fn with_state_cache_probe(mut self, probe: StateCacheProbe) -> Self {
        self.state_cache = Some(probe);
        self
    }
}

impl<S> Layer<S> for CaptureLayer {
    type Service = CaptureService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        CaptureService {
            inner,
            settings: Arc::clone(&self.settings),
            filter: Arc::clone(&self.filter),
            state_cache: self.state_cache.clone(),
        }
    }
}

/// Tower [`Service`] produced by [`CaptureLayer`].
#[derive(Clone)]
pub struct CaptureService<S> {
    inner: S,
    settings: Arc<CaptureSettings>,
    filter: Arc<ParameterFilter>,
    state_cache: Option<StateCacheProbe>,
}

impl<S> Service<Request<Body>> for CaptureService<S>
where
    S: Service<Request<Body>, Response = Response<Body>> + Clone + Send + 'static,
    S::Future: Send + 'static,
    S::Error: Send + 'static,
{
    type Response = Response<Body>;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Request<Body>) -> Self::Future {
        // Clone-and-replace so the polled-ready service moves into the future.
        let cloned = self.inner.clone();
        let mut inner = std::mem::replace(&mut self.inner, cloned);
        let settings = Arc::clone(&self.settings);
        let filter = Arc::clone(&self.filter);
        let state_cache = self.state_cache.clone();

        Box::pin(async move {
            let id = scope_id(&req);
            let route = req
                .extensions()
                .get::<MatchedPath>()
                .map(|matched| matched.as_str().to_owned());
            let scope = Arc::new(CaptureScope::new(id, settings, filter));
            if let Some(probe) = state_cache {
                scope.note_state_cache(probe());
            }
            // The raw peer socket, before any trusted-proxy resolution: a
            // replay restores it verbatim so middleware and handlers that
            // inspect the peer directly (address *and* port) see what the
            // failing request's server saw.
            if let Some(peer) = req
                .extensions()
                .get::<axum::extract::ConnectInfo<std::net::SocketAddr>>()
            {
                scope.set_peer_addr(peer.0);
            }
            scope.set_request(RawRequest {
                method: req.method().as_str().to_owned(),
                uri: req.uri().clone(),
                version: req.version(),
                headers: req.headers().clone(),
                route,
            });
            // Note the body is *teed*, never pre-read: see `arm_body_capture`.
            let mut req = arm_body_capture(req, &scope);
            register(&scope);
            let _guard = RegistryGuard(Arc::clone(&scope));
            req.extensions_mut()
                .insert(CaptureHandle(Arc::clone(&scope)));

            // The scope ends when the response future resolves, so effects a
            // streaming body produces afterwards are not captured. The
            // reporting layer marks a failing response whose body is still
            // streaming at that point as truncated (with a note), so such a
            // capsule is refused by replay rather than presented as complete.
            //
            // `inner.call(req)` is deliberately made *inside* the scoped
            // future rather than passed to `scope` as an argument: arguments
            // are evaluated first, so an inner service that does its work in
            // `call` itself — as a hand-written Tower middleware does, and as
            // `TrustedProxiesService` does when it stamps the client identity
            // — would run before the task-local existed and find no scope.
            CAPSULE_SCOPE
                .scope(scope, async move { inner.call(req).await })
                .await
        })
    }
}

/// The capsule id for a request: its request id when
/// [`RequestIdLayer`](crate::middleware::RequestIdLayer) (installed outer to
/// this one) has already minted one, else a fresh id.
///
/// An inbound id from a proxy is not unique: a proxy retry sends it again.
/// Two scopes with one id collide in the registry, so it gets a fresh id.
fn scope_id(req: &Request<Body>) -> String {
    req.extensions()
        .get::<crate::middleware::RequestId>()
        .filter(|id| !id.is_inbound())
        .map(std::string::ToString::to_string)
        .filter(|id| is_valid_scope_id(id))
        .unwrap_or_else(|| uuid::Uuid::new_v4().simple().to_string())
}

/// Arrange for the request body to be copied into the scope *as the handler
/// reads it*, rather than buffered here.
///
/// This layer is installed outer to the request-timeout layer (see the layer
/// order in `router.rs`), so anything it reads off the socket is read before
/// the deadline starts: pre-buffering even a small body would let a client
/// drip-feed it forever and hold a worker open — a slow-loris vector that
/// would exist only when capture is enabled. Teeing leaves the read where it
/// belongs, inside the handler, where the timeout already bounds it.
///
/// A body declared larger than `max_body_bytes` is not wrapped at all, so an
/// upload streams to the handler exactly as it would without capture.
fn arm_body_capture(req: Request<Body>, scope: &Arc<CaptureScope>) -> Request<Body> {
    let max_body_bytes = scope.settings().max_body_bytes;
    let declared_len = body_length(&req);

    match declared_len {
        Some(0) => {
            scope.arm_body(BodyTap::Absent);
            req
        }
        None if !has_undeclared_body(&req) => {
            scope.arm_body(BodyTap::Absent);
            req
        }
        Some(len) if len > max_body_bytes => {
            scope.arm_body(BodyTap::Skipped { declared_len });
            req
        }
        _ => {
            scope.arm_body(BodyTap::Teeing {
                declared_len,
                buf: Vec::new(),
                end_stream: false,
                overflowed: false,
            });
            let (parts, body) = req.into_parts();
            let teed = Body::new(TeeBody {
                inner: body,
                scope: Arc::clone(scope),
            });
            Request::from_parts(parts, teed)
        }
    }
}

/// Request body that copies each data frame into the capture scope on its way
/// through to the handler.
///
/// Every method delegates, so the handler sees the body it would have seen
/// without capture — same frames, same order, same end-of-stream, same size
/// hint. The copy is bounded by `max_body_bytes` and stops there.
struct TeeBody {
    inner: Body,
    scope: Arc<CaptureScope>,
}

impl http_body::Body for TeeBody {
    type Data = bytes::Bytes;
    type Error = axum::Error;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
        let this = self.get_mut();
        let polled = Pin::new(&mut this.inner).poll_frame(cx);
        match &polled {
            Poll::Ready(Some(Ok(frame))) => {
                if let Some(data) = frame.data_ref() {
                    this.scope.tee_body_chunk(data);
                }
                // A body that announces its end *with* its last frame is
                // finished here, and a handler is entitled to stop rather than
                // poll once more for `Ready(None)`. Waiting for that extra
                // poll called such a body partial and had replay refuse a
                // capsule that was in fact complete — the failure mode that
                // throws away faithful recordings rather than the one that
                // over-trusts them, but a failure mode either way.
                if http_body::Body::is_end_stream(&this.inner) {
                    this.scope.mark_body_end();
                }
            }
            // The handler read the body to its end: what was copied is whole.
            Poll::Ready(None) => this.scope.mark_body_end(),
            Poll::Ready(Some(Err(_))) | Poll::Pending => {}
        }
        polled
    }

    fn is_end_stream(&self) -> bool {
        self.inner.is_end_stream()
    }

    fn size_hint(&self) -> http_body::SizeHint {
        http_body::Body::size_hint(&self.inner)
    }
}

/// The request body's length, from `Content-Length` or — for a body already in
/// memory, as an in-process test client or a body-buffering outer layer
/// produces — the body's own exact size hint.
fn body_length(req: &Request<Body>) -> Option<usize> {
    req.headers()
        .get(axum::http::header::CONTENT_LENGTH)
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.parse::<usize>().ok())
        .or_else(|| usize::try_from(http_body::Body::size_hint(req.body()).exact()?).ok())
}

/// Whether a request whose length [`body_length`] could not determine still has
/// a body worth teeing.
///
/// Asked of the **body**, not the headers. `Transfer-Encoding: chunked` is the
/// HTTP/1.1 way of announcing a body of unknown length, but HTTP/2 and HTTP/3
/// have no such header: a streamed h2 request arrives with no `Content-Length`,
/// no exact size hint and nothing in the headers to go on. Requiring the header
/// would classify every one of those as having no body at all, and the capsule
/// would replay the request empty — a silent falsification, not a limitation.
///
/// Only a body already at end-of-stream is treated as absent. Anything else is
/// teed: `max_body_bytes` still bounds what is kept, and a body that turns out
/// to be empty snapshots as [`CapturedBody::Absent`] anyway, so guessing "yes"
/// costs a wrapper and never a wrong capsule.
fn has_undeclared_body(req: &Request<Body>) -> bool {
    !http_body::Body::is_end_stream(req.body())
}

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

    use bytes::Bytes;
    use http_body::Frame;

    /// Ordered log of who touched what, shared between a test body and the
    /// service it is sent through.
    #[derive(Clone, Default)]
    struct Trace(Arc<Mutex<Vec<&'static str>>>);

    impl Trace {
        fn record(&self, what: &'static str) {
            if let Ok(mut entries) = self.0.lock() {
                entries.push(what);
            }
        }

        fn entries(&self) -> Vec<&'static str> {
            self.0
                .lock()
                .map(|entries| entries.clone())
                .unwrap_or_default()
        }
    }

    /// A request body that logs every poll, so a test can see exactly when the
    /// bytes were read relative to the handler running.
    struct WatchedBody {
        trace: Trace,
        chunks: Vec<&'static [u8]>,
    }

    impl http_body::Body for WatchedBody {
        type Data = Bytes;
        type Error = axum::Error;

        fn poll_frame(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
            let this = self.get_mut();
            this.trace.record("body-polled");
            if this.chunks.is_empty() {
                Poll::Ready(None)
            } else {
                let chunk = this.chunks.remove(0);
                Poll::Ready(Some(Ok(Frame::data(Bytes::from_static(chunk)))))
            }
        }
    }

    fn test_layer(settings: CaptureSettings) -> CaptureLayer {
        CaptureLayer::new(settings, Arc::new(ParameterFilter::new(&[], &[])))
    }

    /// Run a request through the capture layer with an inner service that
    /// records its entry, optionally drains the body, and hands back the
    /// capture handle it found in the extensions.
    async fn run_capture(
        settings: CaptureSettings,
        request: Request<Body>,
        trace: Trace,
        read_body: bool,
    ) -> Arc<CaptureScope> {
        let seen: Arc<Mutex<Option<CaptureHandle>>> = Arc::new(Mutex::new(None));
        let inner_seen = Arc::clone(&seen);
        let inner_trace = trace.clone();
        let inner = tower::service_fn(move |req: Request<Body>| {
            let seen = Arc::clone(&inner_seen);
            let trace = inner_trace.clone();
            async move {
                trace.record("inner-called");
                if let Some(handle) = req.extensions().get::<CaptureHandle>().cloned()
                    && let Ok(mut slot) = seen.lock()
                {
                    *slot = Some(handle);
                }
                if read_body {
                    let _ = axum::body::to_bytes(req.into_body(), usize::MAX).await;
                    trace.record("handler-read-body");
                }
                Ok::<_, std::convert::Infallible>(Response::new(Body::empty()))
            }
        });

        let mut service = test_layer(settings).layer(inner);
        let _response = service
            .call(request)
            .await
            .expect("inner service is infallible");
        let handle = seen
            .lock()
            .expect("handle slot")
            .clone()
            .expect("the capture layer must publish a handle in the request extensions");
        Arc::clone(handle.scope())
    }

    /// Codex review on #3222: the layer reads at the request start whether
    /// the app state has a cache. Code can resolve the cache when
    /// the app is built, before any request scope exists.
    #[tokio::test]
    async fn the_capture_layer_records_a_state_cache() {
        for present in [true, false] {
            let seen: Arc<Mutex<Option<CaptureHandle>>> = Arc::new(Mutex::new(None));
            let inner_seen = Arc::clone(&seen);
            let inner = tower::service_fn(move |req: Request<Body>| {
                let seen = Arc::clone(&inner_seen);
                async move {
                    if let Ok(mut slot) = seen.lock() {
                        *slot = req.extensions().get::<CaptureHandle>().cloned();
                    }
                    Ok::<_, std::convert::Infallible>(Response::new(Body::empty()))
                }
            });
            let mut service = test_layer(CaptureSettings::default())
                .with_state_cache_probe(Arc::new(move || present))
                .layer(inner);
            let _ = service
                .call(Request::new(Body::empty()))
                .await
                .expect("infallible");
            let handle = seen.lock().expect("slot").clone().expect("handle");
            assert_eq!(handle.scope().had_state_cache(), present);
        }
    }

    /// An inner service that does its work in `call` itself, the way a real
    /// Tower middleware does — not inside the future it returns, the way
    /// `service_fn` does.
    #[derive(Clone)]
    struct SyncProbe {
        saw_scope: Arc<Mutex<Option<bool>>>,
    }

    impl Service<Request<Body>> for SyncProbe {
        type Response = Response<Body>;
        type Error = std::convert::Infallible;
        type Future =
            Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;

        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
            Poll::Ready(Ok(()))
        }

        fn call(&mut self, _req: Request<Body>) -> Self::Future {
            if let Ok(mut slot) = self.saw_scope.lock() {
                *slot = Some(current_scope().is_some());
            }
            Box::pin(async { Ok(Response::new(Body::empty())) })
        }
    }

    #[tokio::test]
    async fn an_inner_service_sees_the_scope_from_call_not_only_from_its_future() {
        // `CAPSULE_SCOPE.scope(scope, inner.call(req))` evaluates its argument
        // *first*, so an inner service that records during `call` — which is
        // what a hand-written Tower middleware does, and what
        // `TrustedProxiesService` does when it stamps the client identity —
        // ran outside the task-local and found no scope at all. Every layer
        // inner to this one is affected, so the fix belongs here rather than
        // in each of them.
        //
        // `service_fn` hides this: its closure body is the returned future, so
        // it always runs inside the scope. Only a service that works in `call`
        // itself can catch the regression.
        let saw_scope = Arc::new(Mutex::new(None));
        let probe = SyncProbe {
            saw_scope: Arc::clone(&saw_scope),
        };

        let mut service = test_layer(CaptureSettings::default()).layer(probe);
        let _response = service
            .call(
                Request::get("/x")
                    .body(Body::empty())
                    .expect("request builds"),
            )
            .await
            .expect("probe is infallible");

        assert_eq!(
            *saw_scope.lock().expect("probe slot"),
            Some(true),
            "an inner service must see the capture scope from `call`, not only from its future"
        );
    }

    #[tokio::test]
    async fn call_does_not_read_the_body_before_the_inner_service_runs() {
        // The capture layer sits *outside* the request-timeout layer, so any
        // byte it reads off the socket itself is read before the deadline
        // starts. A slow client dripping a small body would otherwise hold a
        // worker open forever. Bytes must only be read by the handler.
        let trace = Trace::default();
        let request = Request::post("/x")
            .header(axum::http::header::CONTENT_LENGTH, "7")
            .body(Body::new(WatchedBody {
                trace: trace.clone(),
                chunks: vec![b"payload"],
            }))
            .expect("request builds");

        let _scope = run_capture(CaptureSettings::default(), request, trace.clone(), true).await;

        let entries = trace.entries();
        assert_eq!(
            entries.first(),
            Some(&"inner-called"),
            "capture must not touch the request body before the inner service \
             (and therefore the request timeout) is running, got {entries:?}"
        );
    }

    #[tokio::test]
    async fn teed_body_is_captured_whole_when_the_handler_reads_it() {
        let trace = Trace::default();
        let request = Request::post("/x")
            .header(axum::http::header::CONTENT_LENGTH, "10")
            .body(Body::new(WatchedBody {
                trace: trace.clone(),
                chunks: vec![b"hello", b"world"],
            }))
            .expect("request builds");

        let scope = run_capture(CaptureSettings::default(), request, trace, true).await;

        match scope.captured_body() {
            CapturedBody::Buffered(bytes) => assert_eq!(&bytes[..], b"helloworld"),
            other => panic!("a fully read body must be captured whole, got {other:?}"),
        }
        assert_eq!(
            scope.body_note(),
            None,
            "a complete body needs no caveat in the capsule"
        );
    }

    #[tokio::test]
    async fn body_the_handler_never_reads_leaves_a_note_not_a_capture() {
        let trace = Trace::default();
        let request = Request::post("/x")
            .header(axum::http::header::CONTENT_LENGTH, "7")
            .body(Body::new(WatchedBody {
                trace: trace.clone(),
                chunks: vec![b"payload"],
            }))
            .expect("request builds");

        let scope = run_capture(CaptureSettings::default(), request, trace.clone(), false).await;

        assert!(
            !trace.entries().contains(&"body-polled"),
            "nothing may read a body the handler ignored, got {:?}",
            trace.entries()
        );
        assert!(matches!(scope.captured_body(), CapturedBody::Absent));
        assert_eq!(
            scope.body_note(),
            Some(BODY_PARTIAL_NOTE),
            "the capsule must say the body is incomplete rather than imply the \
             request had none"
        );
    }

    #[tokio::test]
    async fn streamed_body_with_no_length_and_no_transfer_encoding_is_still_teed() {
        // HTTP/2 (and /3) have no `Transfer-Encoding`: a streamed h2 request
        // arrives with no `Content-Length`, no exact size hint and no header to
        // hint at one. Deciding "does this request have a body" from the
        // HTTP/1.1 header alone classifies it as having none, and the capsule
        // then replays a request whose body has silently vanished.
        let trace = Trace::default();
        let request = Request::post("/x")
            .version(axum::http::Version::HTTP_2)
            .body(Body::new(WatchedBody {
                trace: trace.clone(),
                chunks: vec![b"h2-", b"payload"],
            }))
            .expect("request builds");

        let scope = run_capture(CaptureSettings::default(), request, trace, true).await;

        match scope.captured_body() {
            CapturedBody::Buffered(bytes) => assert_eq!(&bytes[..], b"h2-payload"),
            other => panic!(
                "a body with no declared length must be teed, not assumed absent, got {other:?}"
            ),
        }
    }

    #[tokio::test]
    async fn a_request_with_no_body_at_all_is_recorded_as_absent() {
        // The other side of the same predicate: an empty body is at
        // end-of-stream from the start, so nothing is wrapped and the capsule
        // records a request that genuinely had no body.
        let request = Request::get("/x")
            .body(Body::empty())
            .expect("request builds");
        let scope = run_capture(CaptureSettings::default(), request, Trace::default(), true).await;

        assert!(matches!(scope.captured_body(), CapturedBody::Absent));
        assert_eq!(
            scope.body_note(),
            None,
            "a request with no body needs no caveat"
        );
    }

    #[tokio::test]
    async fn streamed_body_over_the_cap_is_dropped_mid_stream() {
        // No declared length, so the layer cannot skip up front: the cap has to
        // hold while the frames arrive.
        let trace = Trace::default();
        let request = Request::post("/x")
            .header(axum::http::header::TRANSFER_ENCODING, "chunked")
            .body(Body::new(WatchedBody {
                trace: trace.clone(),
                chunks: vec![b"1234", b"5678", b"9012"],
            }))
            .expect("request builds");

        let settings = CaptureSettings {
            max_body_bytes: 6,
            ..CaptureSettings::default()
        };
        let scope = run_capture(settings, request, trace, true).await;

        assert!(
            matches!(
                scope.captured_body(),
                CapturedBody::Skipped { declared_len: None }
            ),
            "a body that outgrows the cap mid-stream must be dropped, got {:?}",
            scope.captured_body()
        );
        assert_eq!(scope.body_note(), Some(BODY_OVERFLOW_NOTE));
    }

    #[tokio::test]
    async fn body_declared_over_the_cap_is_never_wrapped() {
        let trace = Trace::default();
        let request = Request::post("/x")
            .header(axum::http::header::CONTENT_LENGTH, "4096")
            .body(Body::new(WatchedBody {
                trace: trace.clone(),
                chunks: vec![b"1234"],
            }))
            .expect("request builds");

        let settings = CaptureSettings {
            max_body_bytes: 16,
            ..CaptureSettings::default()
        };
        let scope = run_capture(settings, request, trace, true).await;

        assert!(
            matches!(
                scope.captured_body(),
                CapturedBody::Skipped {
                    declared_len: Some(4096)
                }
            ),
            "an oversized upload must be recorded as skipped, got {:?}",
            scope.captured_body()
        );
        assert_eq!(
            scope.body_note(),
            None,
            "skipping a declared-oversized body is the documented behaviour, \
             not a degraded capture"
        );
    }

    #[tokio::test]
    async fn the_scope_closes_when_the_request_ends() {
        // The scope outlives the request — the reporting layer writes the
        // capsule from a detached task — so "closed" is what stops a pooled
        // connection's next liveness ping from being recorded as something
        // this request did.
        let request = Request::get("/x")
            .body(Body::empty())
            .expect("request builds");
        let scope = run_capture(CaptureSettings::default(), request, Trace::default(), false).await;

        assert!(
            scope.is_closed(),
            "a finished request must stop accepting effects"
        );
        assert!(
            scope_by_id(scope.id()).is_none(),
            "and must no longer be reachable by a connection marker"
        );
    }

    /// A lock poisoned by a panic mid-record means the capsule is missing
    /// whatever was being written. Returning the degraded value alone would
    /// make that indistinguishable from "the request did none of this".
    /// `http_client::encode_body` skips an oversized body before `fill_http`
    /// ever sees it, and a skipped body weighs nothing — so a check written
    /// only against the weight would miss precisely the case it exists for.
    #[test]
    fn a_body_skipped_before_it_reached_the_recorder_still_truncates() {
        let scope = CaptureScope::new(
            "pre".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let slot = scope.reserve_http().expect("a slot is available");
        scope.fill_http(
            slot,
            HttpEffect {
                method: "POST".to_owned(),
                url: "https://api.example/upload".to_owned(),
                // What the outbound client hands over for a body past the cap.
                request_body: CapsuleBody::Skipped {
                    declared_len: Some(50_000_000),
                },
                status: 200,
                ..Default::default()
            },
        );
        let _ = scope.effects_snapshot();
        assert!(
            scope.is_truncated(),
            "a capsule that cannot compare its own outbound body must say so"
        );
        assert!(
            scope
                .notes()
                .iter()
                .any(|note| note == HTTP_BODY_SKIPPED_NOTE),
            "{:?}",
            scope.notes()
        );
    }

    /// The replay comparison treats a skipped body as matching anything, so a
    /// capsule holding one must not present as complete — otherwise the mail
    /// contents could change freely and still replay `reproduced`.
    #[test]
    fn a_mail_body_too_large_to_record_marks_the_capsule_incomplete() {
        let settings = CaptureSettings {
            max_body_bytes: 8,
            ..CaptureSettings::default()
        };
        let scope = CaptureScope::new(
            "mail".to_owned(),
            Arc::new(settings),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let slot = scope.reserve_mail().expect("a slot is available");
        scope.fill_mail(
            slot,
            MailEffect {
                to: vec!["a@example.com".to_owned()],
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("a body well past the cap".to_owned()),
                ..Default::default()
            },
        );
        let effects = scope.effects_snapshot();
        assert!(
            matches!(effects.mail[0].body, CapsuleBody::Skipped { .. }),
            "the oversized body is not kept: {:?}",
            effects.mail[0].body
        );
        assert!(
            scope.is_truncated(),
            "and the capsule says so rather than replaying any body clean"
        );
        assert!(
            scope
                .notes()
                .iter()
                .any(|note| note == MAIL_BODY_SKIPPED_NOTE),
            "{:?}",
            scope.notes()
        );
    }

    /// #2351 item 10: the alternate half of a multipart message has the same
    /// per-body cap as the main body.
    #[test]
    fn an_oversized_alternate_mail_body_is_skipped_and_truncates() {
        let settings = CaptureSettings {
            max_body_bytes: 8,
            ..CaptureSettings::default()
        };
        let scope = CaptureScope::new(
            "mail".to_owned(),
            Arc::new(settings),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let slot = scope.reserve_mail().expect("a slot is available");
        scope.fill_mail(
            slot,
            MailEffect {
                to: vec!["a@example.com".to_owned()],
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("short".to_owned()),
                alternate_body: CapsuleBody::Text("<p>an html half past the cap</p>".to_owned()),
                ..Default::default()
            },
        );
        let effects = scope.effects_snapshot();
        assert!(
            matches!(effects.mail[0].alternate_body, CapsuleBody::Skipped { .. }),
            "{:?}",
            effects.mail[0].alternate_body
        );
        assert!(scope.is_truncated());
    }

    /// #2351 item 10: every retained mail field counts against the capsule
    /// budget.
    #[test]
    fn the_alternate_mail_body_is_charged_against_the_capsule_budget() {
        let settings = CaptureSettings {
            max_body_bytes: 1 << 20,
            max_capsule_bytes: 256,
            ..CaptureSettings::default()
        };
        let scope = CaptureScope::new(
            "mail".to_owned(),
            Arc::new(settings),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let slot = scope.reserve_mail().expect("a slot is available");
        scope.fill_mail(
            slot,
            MailEffect {
                to: vec!["a@example.com".to_owned()],
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("short".to_owned()),
                alternate_body: CapsuleBody::Text("x".repeat(4096)),
                ..Default::default()
            },
        );
        assert!(scope.is_truncated());
    }

    /// Codex review on #3222: every attachment field is charged.
    #[test]
    fn attachment_metadata_is_charged_against_the_capsule_budget() {
        let settings = CaptureSettings {
            max_capsule_bytes: 256,
            ..CaptureSettings::default()
        };
        let scope = CaptureScope::new(
            "mail".to_owned(),
            Arc::new(settings),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let slot = scope.reserve_mail().expect("a slot is available");
        scope.fill_mail(
            slot,
            MailEffect {
                to: vec!["a@example.com".to_owned()],
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("short".to_owned()),
                attachments: vec![crate::capsule::schema::MailAttachmentEffect {
                    filename: "a.pdf".to_owned(),
                    content_type: "x".repeat(4096),
                    len: 1,
                    sha256: String::new(),
                }],
                ..Default::default()
            },
        );
        assert!(scope.is_truncated());
    }

    /// #2351 item 19: an enqueue error is retained, so it is charged.
    #[test]
    fn an_enqueue_error_is_charged_against_the_capsule_budget() {
        let settings = CaptureSettings {
            max_capsule_bytes: 256,
            ..CaptureSettings::default()
        };
        let scope = CaptureScope::new(
            "job".to_owned(),
            Arc::new(settings),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let slot = scope.reserve_job_enqueue().expect("a slot is available");
        scope.fill_job_enqueue(
            slot,
            JobEffect {
                name: "j".to_owned(),
                payload: serde_json::json!({}),
                error: Some("x".repeat(4096)),
                ..Default::default()
            },
        );
        assert!(scope.is_truncated());
    }

    /// A seam reserves its tape position when the effect starts and fills it
    /// when the effect finishes. A cancelled future — the losing branch of a
    /// `tokio::select!`, a timeout — never fills it, and persisting the
    /// placeholder as if it were an outcome would hand replay a backend
    /// failure the run never had.
    #[test]
    fn an_effect_reserved_but_never_completed_marks_the_capsule_incomplete() {
        let scope = CaptureScope::new(
            "sel".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let cancelled = scope.reserve_http().expect("a slot is available");
        let finished = scope.reserve_http().expect("and a second one");
        scope.fill_http(
            finished,
            HttpEffect {
                method: "GET".to_owned(),
                url: "https://a.example/one".to_owned(),
                request_headers: Vec::new(),
                request_body: CapsuleBody::Absent,
                status: 200,
                response_headers: Vec::new(),
                response_body: CapsuleBody::Absent,
                error: None,
                ..Default::default()
            },
        );
        // `cancelled` is deliberately never filled — its future was dropped.
        let _ = cancelled;

        let effects = scope.effects_snapshot();
        assert_eq!(effects.http.len(), 2, "both slots are still on the tape");
        assert!(
            scope.is_truncated(),
            "a recording with an unfinished effect must not present as complete"
        );
        assert!(
            scope
                .notes()
                .iter()
                .any(|note| note == UNFINISHED_EFFECT_NOTE),
            "and must say why: {:?}",
            scope.notes()
        );
    }

    /// Every slot completed: an ordinary recording is not truncated.
    #[test]
    fn effects_that_all_completed_leave_the_capsule_whole() {
        let scope = CaptureScope::new(
            "ok".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        let slot = scope.reserve_http().expect("a slot is available");
        scope.fill_http(
            slot,
            HttpEffect {
                method: "GET".to_owned(),
                url: "https://a.example/one".to_owned(),
                request_headers: Vec::new(),
                request_body: CapsuleBody::Absent,
                status: 200,
                response_headers: Vec::new(),
                response_body: CapsuleBody::Absent,
                error: None,
                ..Default::default()
            },
        );
        let _ = scope.effects_snapshot();
        assert!(!scope.is_truncated(), "{:?}", scope.notes());
    }

    /// A handler that reads exactly `Content-Length` bytes and stops has the
    /// whole body; it just never polled again for the end-of-stream that sets
    /// the flag. Calling that partial would refuse a faithful capsule.
    #[test]
    fn a_body_read_to_its_declared_length_is_not_partial() {
        let scope = CaptureScope::new(
            "body".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        scope.arm_body(BodyTap::Teeing {
            declared_len: Some(5),
            buf: b"hello".to_vec(),
            end_stream: false,
            overflowed: false,
        });
        assert_eq!(
            scope.body_note(),
            None,
            "a body captured up to its declared length is complete"
        );

        // One byte short is genuinely partial, and must still say so.
        let scope = CaptureScope::new(
            "body".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        scope.arm_body(BodyTap::Teeing {
            declared_len: Some(5),
            buf: b"hell".to_vec(),
            end_stream: false,
            overflowed: false,
        });
        assert_eq!(scope.body_note(), Some(BODY_PARTIAL_NOTE));

        // A body of undeclared length has nothing to compare against, so the
        // end-of-stream flag remains the only evidence.
        let scope = CaptureScope::new(
            "body".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        );
        scope.arm_body(BodyTap::Teeing {
            declared_len: None,
            buf: b"hello".to_vec(),
            end_stream: false,
            overflowed: false,
        });
        assert_eq!(scope.body_note(), Some(BODY_PARTIAL_NOTE));
    }

    /// A body that reports end-of-stream *with* its last frame rather than on
    /// a following poll.
    struct EagerEndBody {
        chunk: Option<&'static [u8]>,
    }

    impl http_body::Body for EagerEndBody {
        type Data = Bytes;
        type Error = axum::Error;

        fn poll_frame(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
        ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
            let this = self.get_mut();
            this.chunk.take().map_or(Poll::Ready(None), |chunk| {
                Poll::Ready(Some(Ok(Frame::data(Bytes::from_static(chunk)))))
            })
        }

        fn is_end_stream(&self) -> bool {
            self.chunk.is_none()
        }
    }

    #[test]
    fn a_body_that_ends_with_its_last_frame_is_not_partial() {
        // A streaming body with no `Content-Length` can announce its end
        // alongside the final frame, and a handler is entitled to stop there
        // rather than poll again for `Ready(None)`. There is no declared
        // length to compare against, so the end-of-stream flag is the only
        // evidence — and waiting for the extra poll to set it had replay
        // refuse a capsule whose body was complete.
        let scope = Arc::new(CaptureScope::new(
            "body".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        ));
        scope.arm_body(BodyTap::Teeing {
            declared_len: None,
            buf: Vec::new(),
            end_stream: false,
            overflowed: false,
        });

        let mut tee = TeeBody {
            inner: Body::new(EagerEndBody {
                chunk: Some(b"hello"),
            }),
            scope: Arc::clone(&scope),
        };
        let waker = std::task::Waker::noop();
        let mut cx = Context::from_waker(waker);
        let polled = http_body::Body::poll_frame(Pin::new(&mut tee), &mut cx);

        assert!(
            matches!(polled, Poll::Ready(Some(Ok(_)))),
            "the frame is passed through"
        );
        assert_eq!(
            scope.body_note(),
            None,
            "a body that ended with its last frame is complete, not partial"
        );
    }

    #[test]
    fn a_poisoned_buffer_marks_the_capsule_truncated() {
        let scope = Arc::new(CaptureScope::new(
            "poisoned".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        ));
        let panicking = Arc::clone(&scope);
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            panicking.with_db(|_| panic!("recording interrupted"));
        }));

        assert!(
            scope.db_snapshot().is_none(),
            "an unreachable buffer yields no tape"
        );
        assert!(
            scope.is_truncated(),
            "and the capsule must say it is incomplete rather than imply the request \
             never touched the database"
        );

        let body_scope = Arc::new(CaptureScope::new(
            "poisoned-body".to_owned(),
            Arc::new(CaptureSettings::default()),
            Arc::new(ParameterFilter::new(&[], &[])),
        ));
        let panicking = Arc::clone(&body_scope);
        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _guard = panicking.body.lock();
            panic!("body copy interrupted");
        }));
        assert!(matches!(body_scope.captured_body(), CapturedBody::Absent));
        assert!(
            body_scope.is_truncated(),
            "a body that could not be read back is a truncated capture, not an absent body"
        );
    }

    #[test]
    fn scope_ids_are_bounded_and_charset_checked() {
        assert!(is_valid_scope_id("018f-4b2c_AB"));
        assert!(!is_valid_scope_id(""));
        assert!(!is_valid_scope_id("has space"));
        assert!(!is_valid_scope_id("quote'; DROP TABLE users; --"));
        assert!(!is_valid_scope_id(&"a".repeat(MAX_SCOPE_ID_LEN + 1)));
    }

    #[test]
    fn db_buffer_charges_against_the_budget() {
        let mut buffer = DbBuffer::default();
        assert!(buffer.charge(400, 1000));
        assert!(buffer.charge(600, 1000));
        assert!(!buffer.charge(1, 1000), "the budget must eventually stop");
        assert_eq!(buffer.charged_bytes(), 1001);
    }

    #[test]
    fn db_buffer_snapshots_tapes_in_first_use_order() {
        // Connection ids are process-wide birth order, which says nothing about
        // the order *this* request reached for them: a long-lived pooled
        // connection 2 can easily be checked out after a freshly minted 7.
        // Replay hands tape *i* to the *i*-th connection its pool opens, so the
        // capsule must list them in the order the request first used them or
        // the tapes get swapped and both connections diverge.
        let mut buffer = DbBuffer::default();
        buffer.tape_mut(7);
        buffer.tape_mut(2);
        buffer.tape_mut(7);
        let snapshot = buffer.snapshot().expect("tapes were created");
        let ids: Vec<u64> = snapshot.connections.iter().map(|tape| tape.id).collect();
        assert_eq!(
            ids,
            vec![7, 2],
            "tapes must be listed in the order the request first used each connection"
        );
    }
}