autumn-web 0.9.0

//! Replay-side tape for the effect seams outside the request and the database.
//!
//! The database has a wire-level stub server; everything else a request can
//! touch — an outbound HTTP call, a job enqueue, a cache read, a mail send,
//! the resolved tenant — is served from here (#1634). The posture is exactly
//! the database tape's:
//!
//! * a recorded effect is served from the capsule, never performed live;
//! * an effect the capsule cannot answer is a **divergence**, not a fallback
//!   to the real thing;
//! * a recorded effect the replayed run never asked for is a divergence too,
//!   because reaching the recorded outcome without the recorded effects is not
//!   a reproduction.
//!
//! The tape is installed as a **task-local** rather than a process global, the
//! way [`clock::with_replay_request_scope`](crate::capsule::clock) marks the
//! replay request's task. That is what lets several capsules replay
//! concurrently in one `cargo test` process — which is exactly what a
//! committed regression corpus does.

// autumn-panic-gate: request-path module — production code path must be panic-free.
// The tape itself only ever serves a replay, but `current_tape`/`tape_active`
// are probed from the serving path (the HTTP client, job enqueue, the cache,
// the mailer, tenancy), so this module is gated like one.
// 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, VecDeque};
use std::future::Future;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

use base64::Engine as _;
use base64::engine::general_purpose::STANDARD;
use serde::Serialize;

use crate::capsule::schema::{
    CacheEffect, CapsuleBody, CapsuleEffects, HttpEffect, JobEffect, MailEffect, TenantEffect,
};

// ── Divergences ─────────────────────────────────────────────────────────────

/// Which effect seam a divergence happened on.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EffectSeam {
    /// Outbound HTTP through [`http_client`](crate::http_client).
    Http,
    /// Background job enqueue.
    Job,
    /// Cache read or write.
    Cache,
    /// Mail handed to the mailer.
    Mail,
    /// The tenant context the run resolved.
    Tenant,
    /// A draw from the framework's entropy source.
    Random,
    /// Work started on a detached task (`capsule::spawn`).
    Detached,
}

impl EffectSeam {
    /// Short label used in the human summary.
    #[must_use]
    pub const fn label(self) -> &'static str {
        match self {
            Self::Http => "outbound http",
            Self::Job => "job enqueue",
            Self::Cache => "cache",
            Self::Mail => "mail",
            Self::Tenant => "tenancy",
            Self::Random => "randomness",
            Self::Detached => "detached task",
        }
    }
}

/// Why a replayed effect did not line up with the tape.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EffectDivergenceKind {
    /// The tape holds no recording of this effect at all.
    Unrecorded,
    /// The next recorded effect on this seam was a different one.
    Mismatch,
    /// The seam ran past the end of its recorded effects.
    Exhausted,
    /// The run finished with recorded effects it never asked for.
    Unconsumed,
}

impl EffectDivergenceKind {
    /// Short label used in the human summary.
    #[must_use]
    pub const fn label(self) -> &'static str {
        match self {
            Self::Unrecorded => "unrecorded effect",
            Self::Mismatch => "effect mismatch",
            Self::Exhausted => "tape exhausted",
            Self::Unconsumed => "unconsumed effects",
        }
    }
}

/// One place where a replayed run's effects left the capsule.
#[derive(Debug, Clone, Serialize)]
pub struct EffectDivergence {
    /// Which seam.
    pub seam: EffectSeam,
    /// What went wrong.
    pub kind: EffectDivergenceKind,
    /// Position in the seam's recorded list.
    pub index: usize,
    /// What the tape expected next, when it expected anything.
    pub expected: Option<String>,
    /// What the replayed run actually did.
    pub actual: String,
    /// Human-readable explanation, safe to print.
    pub detail: String,
}

/// What the tape can say about a cache read.
///
/// A named three-state answer rather than a nested `Option`, because the three
/// cases mean genuinely different things to the caller and to the verdict.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CachedValue {
    /// The capsule never saw this key. A divergence has been logged; the read
    /// replays as a miss.
    Unrecorded,
    /// The recording read this key and missed.
    Miss,
    /// The recording read this key and got these bytes.
    Hit(Vec<u8>),
}

/// What the tape can say about a job enqueue.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum EnqueueVerdict {
    /// Matched a recorded enqueue that succeeded; nothing reaches a queue.
    Queued,
    /// Matched a recorded enqueue that the backend **rejected**; the caller
    /// reproduces the error, with its recorded status when there is one.
    Failed(String, Option<u16>),
    /// Did not match the recording. A divergence has been logged.
    Diverged,
}

/// What the tape can say about a mail send.
///
/// Gated with the seam that consumes it: a build without `mail` has no
/// `Mailer::send` to serve, though a capsule recorded by a `mail` build still
/// carries — and still reports — its mail tape.
#[cfg(any(test, feature = "mail"))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum MailVerdict {
    /// Matched a recorded successful send; nothing is delivered.
    Sent,
    /// Matched a recorded send that *failed*; the caller reproduces the error.
    ///
    /// Carries the recorded kind alongside the text so the caller can rebuild
    /// the variant a handler branches on, not only what it prints.
    Failed(String, Option<crate::capsule::schema::MailErrorKind>),
    /// Did not match the recording. A divergence has been logged.
    Diverged,
}

/// What the tape can say about a tenant lookup.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum TenantVerdict {
    /// The recording resolved this tenant.
    Resolved(String),
    /// The recording tried and failed. The caller runs the resolver again on
    /// the restored request, so the same error comes back.
    RecordedFailure,
    /// The recording has no tenant lookup. A divergence has been logged; the
    /// caller fails closed.
    Unrecorded,
    /// A v3 capsule has no tenant lookup: v3 did not record a failed one.
    /// The caller runs the resolver and does not check the result.
    Unchecked,
}

// ── The tape ────────────────────────────────────────────────────────────────

/// An ordered seam's recorded entries and how far the run has consumed them.
#[derive(Debug)]
struct Ordered<T> {
    pending: VecDeque<T>,
    consumed: usize,
}

impl<T> Ordered<T> {
    fn new(entries: Vec<T>) -> Self {
        Self {
            pending: entries.into(),
            consumed: 0,
        }
    }
}

/// The recorded effects a replayed run is served from.
#[derive(Debug)]
pub struct ReplayEffects {
    http: Mutex<Ordered<HttpEffect>>,
    jobs: Mutex<Ordered<JobEffect>>,
    mail: Mutex<Ordered<MailEffect>>,
    /// Recorded cache *reads*, by key rather than in order: one key is
    /// legitimately read many times in a run, and a recorded read order would
    /// make an innocent extra hit look like a divergence.
    cache: Mutex<BTreeMap<String, CacheSlot>>,
    /// Recorded cache *writes*, in order. Writes are ordered effects like an
    /// enqueue or a send — dropping one, or changing what it stores, is a
    /// behaviour change the capsule can see — so they are consumed and
    /// compared rather than merely applied.
    cache_writes: Mutex<Ordered<CacheWrite>>,
    /// The recorded tenant, and whether the replayed run ever asked for it.
    ///
    /// Tracked like every other seam: a router whose updated code no longer
    /// resolves a tenant would otherwise leave the recording untouched and
    /// still report `Reproduced` on an unchanged response.
    tenant: RecordedTenant,
    tenant_read: std::sync::atomic::AtomicBool,
    /// The capsule is format version 3. Seams that version did not record
    /// (cache removals, untyped reads, failed tenant lookups, draw widths)
    /// replay with the v3 rules, so an older corpus keeps its verdicts.
    legacy_v3: bool,
    /// Values that stood behind a placeholder in a recorded effect the run
    /// matched. The actual outcome is masked with them before it is judged,
    /// as persistence masked the recorded one (#2351 item 16).
    observed: Mutex<crate::capsule::redact::RedactedValues>,
    divergences: Mutex<Vec<EffectDivergence>>,
    /// Total effects the tape was built with, so the verdict can say how much
    /// of the recording the run actually met.
    recorded: usize,
    served: AtomicUsize,
    /// Set when the replayed router actually entered the consuming scope.
    ///
    /// `ReportingLayer` establishes that scope, so a router assembled without
    /// it — a hand-built `axum::Router`, or a regression test whose factory
    /// skips `TestApp` — would run with the clock and the entropy source
    /// serving stable non-consuming values and still be able to report
    /// `Reproduced`. The verdict warns instead of pretending.
    scope_entered: std::sync::atomic::AtomicBool,
    /// `register_after_commit` callbacks that a rollback has not dropped.
    after_commit: AtomicUsize,
}

/// One recorded cache write: a value written, or entries removed.
#[derive(Debug, Clone, PartialEq, Eq)]
enum CacheWrite {
    Insert {
        key: String,
        /// `None` when the recorded value was not decodable; the key still has
        /// to match.
        value: Option<Vec<u8>>,
        /// The expiry the recorded write asked for, in seconds.
        ttl_secs: Option<u64>,
    },
    Invalidate {
        key: String,
        error: Option<crate::capsule::schema::CacheInvalidationError>,
    },
    InvalidateNamespace {
        namespace: String,
        error: Option<crate::capsule::schema::CacheInvalidationError>,
    },
    Clear,
}

impl CacheWrite {
    /// The write tape entry a recorded effect describes, or `None` for a read.
    fn from_effect(effect: &CacheEffect) -> Option<Self> {
        match effect {
            CacheEffect::Get { .. } => None,
            CacheEffect::Insert {
                key,
                value,
                ttl_secs,
            } => Some(Self::Insert {
                key: key.clone(),
                value: decode(value),
                ttl_secs: *ttl_secs,
            }),
            CacheEffect::Invalidate { key, error } => Some(Self::Invalidate {
                key: key.clone(),
                error: error.clone(),
            }),
            CacheEffect::InvalidateNamespace { namespace, error } => {
                Some(Self::InvalidateNamespace {
                    namespace: namespace.clone(),
                    error: error.clone(),
                })
            }
            CacheEffect::Clear => Some(Self::Clear),
        }
    }

    /// Whether this recorded write is the one the replayed run made.
    ///
    /// The recorded key and value may have been masked on their way to disk,
    /// so they compare the way every other redacted recording does. The expiry
    /// is part of the mutation: a five-second entry and a permanent one are
    /// different writes, whatever the bytes say.
    fn matches(&self, actual: &Self) -> bool {
        match (self, actual) {
            (
                Self::Insert {
                    key,
                    value,
                    ttl_secs,
                },
                Self::Insert {
                    key: actual_key,
                    value: actual_value,
                    ttl_secs: actual_ttl,
                },
            ) => {
                matches_redacted(key, actual_key)
                    && ttl_secs == actual_ttl
                    && value.as_ref().is_none_or(|recorded| {
                        actual_value
                            .as_ref()
                            .is_some_and(|actual| bytes_match_redacted(recorded, actual))
                    })
            }
            (Self::Invalidate { key, .. }, Self::Invalidate { key: actual, .. })
            | (
                Self::InvalidateNamespace { namespace: key, .. },
                Self::InvalidateNamespace {
                    namespace: actual, ..
                },
            ) => matches_redacted(key, actual),
            (Self::Clear, Self::Clear) => true,
            _ => false,
        }
    }

    /// How the write is named in a divergence report.
    fn describe(&self) -> String {
        match self {
            Self::Insert { key, .. } => key.clone(),
            Self::Invalidate { key, .. } => format!("invalidate {key}"),
            Self::InvalidateNamespace { namespace, .. } => {
                format!("invalidate namespace {namespace}")
            }
            Self::Clear => "clear".to_owned(),
        }
    }

    /// The key or namespace the write names.
    fn identifier(&self) -> Option<&str> {
        match self {
            Self::Insert { key, .. } | Self::Invalidate { key, .. } => Some(key),
            Self::InvalidateNamespace { namespace, .. } => Some(namespace),
            Self::Clear => None,
        }
    }

    /// [`describe`](Self::describe) with no key or namespace.
    fn describe_withheld(&self) -> String {
        match self {
            Self::Insert { .. } => "a cache write (key withheld)".to_owned(),
            Self::Invalidate { .. } => "invalidate (key withheld)".to_owned(),
            Self::InvalidateNamespace { .. } => {
                "invalidate namespace (namespace withheld)".to_owned()
            }
            Self::Clear => "clear".to_owned(),
        }
    }
}

/// Whether a recorded key is in `namespace`: `"{namespace}:..."`, the shape
/// [`make_cache_key`](crate::cache::make_cache_key) makes. A recorded key can
/// be masked, so its prefix compares like other redacted text.
fn in_namespace(key: &str, namespace: &str) -> bool {
    if key
        .strip_prefix(namespace)
        .is_some_and(|rest| rest.starts_with(':'))
    {
        return true;
    }
    // The key suffix can hold a colon too, so try each boundary.
    key.match_indices(':')
        .filter_map(|(index, _)| key.get(..index))
        .any(|prefix| matches_redacted(prefix, namespace))
}

/// Which read-map slots a replayed write changes.
#[derive(Debug, Clone, Copy)]
enum SlotUpdate<'a> {
    Key(&'a str),
    Namespace(&'a str),
    All,
}

/// The recorded slot a live key reads from: the exact key, or the one masked
/// key it matches.
///
/// Exactly one match, or none. A masked key such as `user:[FILTERED]:profile`
/// matches the entry of each subject. If replay served the first match, a read
/// of the wrong subject would get the recorded value. The bug would then look
/// like a clean reproduction.
fn resolve_slot(cache: &BTreeMap<String, CacheSlot>, key: &str) -> Option<String> {
    if cache.contains_key(key) {
        return Some(key.to_owned());
    }
    let mut candidates = cache
        .keys()
        .filter(|recorded| matches_redacted(recorded, key));
    match (candidates.next().cloned(), candidates.next()) {
        (Some(only), None) => Some(only),
        _ => None,
    }
}

/// What the capsule recorded about the tenant lookup.
#[derive(Debug, Clone, PartialEq, Eq)]
enum RecordedTenant {
    /// No lookup was recorded.
    None,
    /// The lookup resolved this tenant.
    Resolved(String),
    /// The lookup failed.
    Failed,
}

/// One cache key's recorded value and whether the run has read it.
#[derive(Debug)]
struct CacheSlot {
    /// `None` is a recorded miss (or a hit the backend could not serialize —
    /// see [`ReplayEffects::cache_get`]).
    value: Option<Vec<u8>>,
    /// Whether the recording ever *read* this key, as opposed to only writing
    /// it. Only recorded reads are owed back by the replay.
    was_read: bool,
    /// Whether the replayed run has read it. A write does not count: a
    /// recorded read the run dropped is still owed.
    touched: bool,
}

impl ReplayEffects {
    /// Build the tape a capsule's recorded effects describe.
    #[must_use]
    pub fn new(effects: CapsuleEffects) -> Self {
        // Seeded from the recorded **reads** only, and from the *first* read of
        // each key.
        //
        // Folding writes in as well would break the commonest cache shape
        // there is: a handler that misses, computes, fills, and (later, or on
        // the next request) hits. The recorded `Get{value: None}` and the
        // `Insert{…}` that followed it would collapse into one pre-seeded hit,
        // the replayed handler would take the hit branch its recording never
        // took, and the capsule would report a mismatch caused entirely by
        // replay. A write the replayed run performs is applied by
        // `cache_insert` at the moment it happens, which is what makes
        // fill-then-read-back work with the recorded ordering intact.
        let mut cache: BTreeMap<String, CacheSlot> = BTreeMap::new();
        for entry in &effects.cache {
            if let CacheEffect::Get { key, value } = entry {
                cache.entry(key.clone()).or_insert_with(|| CacheSlot {
                    value: value.as_ref().and_then(|encoded| decode(encoded)),
                    was_read: true,
                    touched: false,
                });
            }
        }
        let cache_writes: Vec<CacheWrite> = effects
            .cache
            .iter()
            .filter_map(CacheWrite::from_effect)
            .collect();
        let recorded = effects
            .http
            .len()
            .saturating_add(effects.jobs.len())
            .saturating_add(effects.mail.len())
            .saturating_add(cache.len())
            .saturating_add(cache_writes.len());
        Self {
            http: Mutex::new(Ordered::new(effects.http)),
            jobs: Mutex::new(Ordered::new(effects.jobs)),
            mail: Mutex::new(Ordered::new(effects.mail)),
            cache: Mutex::new(cache),
            cache_writes: Mutex::new(Ordered::new(cache_writes)),
            tenant: match effects.tenant {
                None => RecordedTenant::None,
                Some(TenantEffect { id: Some(id) }) => RecordedTenant::Resolved(id),
                Some(TenantEffect { id: None }) => RecordedTenant::Failed,
            },
            legacy_v3: false,
            tenant_read: std::sync::atomic::AtomicBool::new(false),
            observed: Mutex::new(crate::capsule::redact::RedactedValues::default()),
            divergences: Mutex::new(Vec::new()),
            recorded,
            served: AtomicUsize::new(0),
            scope_entered: std::sync::atomic::AtomicBool::new(false),
            after_commit: AtomicUsize::new(0),
        }
    }

    /// Replay with the rules of the capsule's format version. See
    /// [`CAPSULE_FORMAT_VERSION`](crate::capsule::CAPSULE_FORMAT_VERSION).
    #[must_use]
    pub const fn for_format_version(mut self, format_version: u32) -> Self {
        self.legacy_v3 = format_version < 4;
        self
    }

    /// Whether the capsule is format version 3.
    #[must_use]
    pub(crate) const fn legacy_v3(&self) -> bool {
        self.legacy_v3
    }

    /// Record that the replayed router entered the consuming scope.
    pub(crate) fn mark_scope_entered(&self) {
        self.scope_entered.store(true, Ordering::SeqCst);
    }

    /// Whether the replayed router entered the consuming scope.
    #[must_use]
    pub fn scope_entered(&self) -> bool {
        self.scope_entered.load(Ordering::SeqCst)
    }

    /// How many recorded effects the tape holds.
    #[must_use]
    pub const fn recorded(&self) -> usize {
        self.recorded
    }

    /// How many of them the replayed run consumed.
    #[must_use]
    pub fn served(&self) -> usize {
        self.served.load(Ordering::SeqCst)
    }

    /// The divergences observed so far.
    #[must_use]
    pub fn divergences(&self) -> Vec<EffectDivergence> {
        self.divergences
            .lock()
            .map(|log| log.clone())
            .unwrap_or_default()
    }

    fn diverge(&self, divergence: EffectDivergence) {
        if let Ok(mut log) = self.divergences.lock() {
            log.push(divergence);
        }
    }

    /// The response the capsule recorded for the next outbound call, or `None`
    /// when the tape cannot answer it — in which case a divergence has been
    /// logged and the caller must fail the call rather than dial the peer.
    #[must_use]
    pub(crate) fn next_http(&self, request: &OutboundRequest<'_>) -> Option<HttpEffect> {
        let (method, raw_url) = (request.method, request.url);
        self.observe_live_userinfo(raw_url);
        // The URL parser ignores ASCII tab/CR/LF, so two spellings that differ
        // only by those are one URL: compare and report the parser's view.
        let live = crate::capsule::redact::parser_view(raw_url);
        let url = live.as_ref();
        let Ok(mut seam) = self.http.lock() else {
            return None;
        };
        let index = seam.consumed;
        let Some(next) = seam.pending.front() else {
            drop(seam);
            let actual = describe_http(method, url);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Http,
                kind: if self.recorded_http_is_empty() {
                    EffectDivergenceKind::Unrecorded
                } else {
                    EffectDivergenceKind::Exhausted
                },
                index,
                expected: None,
                actual: actual.clone(),
                detail: format!(
                    "the replayed handler made an outbound request the capsule has no \
                     recording for ({actual}); it was refused rather than sent"
                ),
            });
            return None;
        };
        if !next.method.eq_ignore_ascii_case(method)
            || !matches_redacted(&crate::capsule::redact::parser_view(&next.url), url)
        {
            let expected = describe_http(&next.method, &next.url);
            drop(seam);
            let actual = describe_http(method, url);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Http,
                kind: EffectDivergenceKind::Mismatch,
                index,
                expected: Some(expected.clone()),
                actual: actual.clone(),
                detail: format!(
                    "the replayed handler called {actual}, but the capsule's next recorded \
                     outbound call was {expected}"
                ),
            });
            return None;
        }
        // Same endpoint, different message. Serving the recorded response here
        // would hide the change *and* leave the audit nothing to notice, so an
        // unchanged final failure would read as a reproduction of a run that
        // sent something else entirely.
        if !headers_match_redacted(&next.request_headers, request.headers)
            || !body_matches_redacted(&next.request_body, request.body)
        {
            let expected = describe_http(&next.method, &next.url);
            let changed = describe_request_mismatch(next, request);
            let recorded_url = next.url.clone();
            drop(seam);
            // The URL matched: learn what its masked values stood for before
            // the report prints the live URL.
            self.observe_url(&recorded_url, url);
            let actual = describe_http(method, url);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Http,
                kind: EffectDivergenceKind::Mismatch,
                index,
                expected: Some(expected),
                actual: actual.clone(),
                detail: format!(
                    "the replayed handler called {actual} as the capsule recorded, but changed \
                     {changed}; the recorded response was not served"
                ),
            });
            return None;
        }
        let served = seam.pending.pop_front();
        seam.consumed = seam.consumed.saturating_add(1);
        drop(seam);
        if let Some(recorded) = served.as_ref() {
            self.observe_url(&recorded.url, url);
            self.observe_headers(&recorded.request_headers, request.headers);
            self.observe_body(&recorded.request_body, request.body);
        }
        self.served.fetch_add(1, Ordering::SeqCst);
        served
    }

    /// Whether the capsule recorded no outbound calls at all — the difference
    /// between "this run never called out" and "it ran off the end".
    fn recorded_http_is_empty(&self) -> bool {
        self.http
            .lock()
            .is_ok_and(|seam| seam.consumed == 0 && seam.pending.is_empty())
    }

    /// Whether the next recorded enqueue is the one the run just made.
    ///
    /// `true` means the enqueue is satisfied from the tape and must **not** be
    /// written to a queue; `false` means it diverged and the caller should fail
    /// it. The schedule is compared on the same terms the caller stated it: a
    /// job that used to run immediately and now runs in an hour, or one whose
    /// absolute deadline moved, is a behaviour change the capsule can see.
    #[must_use]
    pub(crate) fn next_job(
        &self,
        name: &str,
        payload: &serde_json::Value,
        schedule: crate::job::EnqueueSchedule,
    ) -> EnqueueVerdict {
        let Ok(mut seam) = self.jobs.lock() else {
            return EnqueueVerdict::Diverged;
        };
        let index = seam.consumed;
        let Some(next) = seam.pending.front() else {
            drop(seam);
            let actual = describe_job(name, payload);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Job,
                kind: EffectDivergenceKind::Unrecorded,
                index,
                expected: None,
                actual: actual.clone(),
                detail: format!(
                    "the replayed run enqueued a job the capsule has no recording for \
                     ({actual}); nothing was written to a queue"
                ),
            });
            return EnqueueVerdict::Diverged;
        };
        // Each schedule is compared on its own terms. A relative enqueue is
        // compared by delay; an absolute one by its deadline, because the delay
        // recorded for it is `deadline - now` and would differ between capture
        // and replay purely because time passed. `Immediate` states nothing
        // about timing, so there is nothing to compare — an entry point that
        // takes no schedule cannot have changed one.
        let schedule_diverged = match schedule {
            // A plain enqueue states no timing, so it needs an entry that ran
            // at once: no positive delay and no future deadline (#2351 item
            // 17). A zero delay is still immediate.
            crate::job::EnqueueSchedule::Immediate => {
                next.delay_secs.is_some_and(|delay| delay > 0)
                    || next.due_at.is_some()
                    || next.requested_due_at.is_some()
            }
            // Capture rounds a sub-second remainder up. A v3 capture cut it
            // off, so a v3 capsule is compared the way it was recorded.
            crate::job::EnqueueSchedule::After(delay) => {
                let recorded = if self.legacy_v3 {
                    i64::try_from(delay.as_secs()).unwrap_or(i64::MAX)
                } else {
                    crate::job::delay_seconds(delay)
                };
                // Capture runs a zero delay at once and records no delay.
                let ran_at_once = delay.is_zero()
                    && next.delay_secs.is_none()
                    && next.due_at.is_none()
                    && next.requested_due_at.is_none();
                next.delay_secs != Some(recorded) && !ran_at_once
            }
            // Capture runs a past deadline at once and records `due_at: None`.
            // It keeps the deadline the caller gave in `requested_due_at`, so
            // replay compares that one (#2351 item 18).
            crate::job::EnqueueSchedule::At(deadline) => {
                next.due_at != Some(deadline)
                    && !(next.due_at.is_none() && next.requested_due_at == Some(deadline))
            }
        };
        if next.name != name || schedule_diverged || !json_matches_redacted(&next.payload, payload)
        {
            let expected = describe_job(&next.name, &next.payload);
            drop(seam);
            let actual = describe_job(name, payload);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Job,
                kind: EffectDivergenceKind::Mismatch,
                index,
                expected: Some(expected.clone()),
                actual: actual.clone(),
                detail: format!(
                    "the replayed run enqueued {actual}, but the capsule's next recorded \
                     enqueue was {expected}"
                ),
            });
            return EnqueueVerdict::Diverged;
        }
        let recorded = seam.pending.pop_front();
        seam.consumed = seam.consumed.saturating_add(1);
        drop(seam);
        if let Some(recorded) = recorded.as_ref() {
            self.observe_json(&recorded.payload, payload);
        }
        self.served.fetch_add(1, Ordering::SeqCst);
        recorded
            .and_then(|recorded| Some((recorded.error?, recorded.error_status)))
            .map_or(EnqueueVerdict::Queued, |(error, status)| {
                EnqueueVerdict::Failed(error, status)
            })
    }

    /// What the capsule recorded for a cache key.
    ///
    /// [`CachedValue::Unrecorded`] is a divergence: a replayed run reading a
    /// key the recording never read has taken a different path through the
    /// handler.
    #[must_use]
    pub(crate) fn cache_get(&self, key: &str) -> CachedValue {
        let Ok(mut cache) = self.cache.lock() else {
            return CachedValue::Unrecorded;
        };
        let recorded_key = resolve_slot(&cache, key);
        let Some(slot) = recorded_key
            .as_ref()
            .and_then(|recorded| cache.get_mut(recorded))
        else {
            drop(cache);
            // No recorded key teaches the verdict what to mask, and a key can
            // hold a secret, so it is not printed.
            self.diverge(EffectDivergence {
                seam: EffectSeam::Cache,
                kind: EffectDivergenceKind::Unrecorded,
                index: 0,
                expected: None,
                actual: "a cache read (key withheld)".to_owned(),
                detail: "the replayed run read a cache key the capsule has no recording for; \
                         it was served as a miss"
                    .to_owned(),
            });
            return CachedValue::Unrecorded;
        };
        let first_touch = !slot.touched;
        slot.touched = true;
        let was_read = slot.was_read;
        let value = slot.value.clone();
        drop(cache);
        if let Some(recorded_key) = recorded_key {
            self.observe_text(&recorded_key, key);
        }
        if !was_read {
            // The recording wrote this key and never read it. A read here is a
            // new cache dependency, not a read-back (#2351 item 15). The run's
            // own write is still served, as production would serve it.
            // The slot can hold the run's own unrecorded key, which no
            // recorded identifier masks, so the key is not printed.
            self.diverge(EffectDivergence {
                seam: EffectSeam::Cache,
                kind: EffectDivergenceKind::Unrecorded,
                index: 0,
                expected: None,
                actual: "a cache read (key withheld)".to_owned(),
                detail: "the replayed run read a cache key that the recording wrote but never \
                         read; the run's own write was served"
                    .to_owned(),
            });
            return value.map_or(CachedValue::Miss, CachedValue::Hit);
        }
        if first_touch {
            self.served.fetch_add(1, Ordering::SeqCst);
        }
        value.map_or(CachedValue::Miss, CachedValue::Hit)
    }

    /// Apply a cache write made during replay, and check it against the
    /// recording.
    ///
    /// Two jobs at once. The value lands in the read map so a read-back later
    /// in the same run finds it, exactly as it did in production. And the write
    /// is *consumed* from the recorded write tape and compared — a write the
    /// code dropped, added, or changed the value of is a behaviour change, and
    /// without this it could ride along under an unchanged response and still
    /// report `Reproduced`. Nothing is ever sent to a live backend.
    pub(crate) fn cache_insert(&self, key: &str, value: &[u8], ttl_secs: Option<u64>) {
        self.update_slots(SlotUpdate::Key(key), Some(value));
        let _ = self.next_cache_write(&CacheWrite::Insert {
            key: key.to_owned(),
            value: Some(value.to_vec()),
            ttl_secs,
        });
    }

    /// Apply a key removal made during replay, and check it against the
    /// recording. Returns the recorded result.
    ///
    /// # Errors
    ///
    /// The recorded failure, or a failure when the removal does not match the
    /// recording (the divergence is logged).
    pub(crate) fn cache_invalidate(
        &self,
        key: &str,
    ) -> Result<(), crate::cache::InvalidationError> {
        // A v3 capsule did not record removals.
        let result = if self.legacy_v3 {
            Ok(())
        } else {
            self.removal_result(&CacheWrite::Invalidate {
                key: key.to_owned(),
                error: None,
            })
        };
        // A removal that failed in production kept the entry.
        if result.is_ok() {
            self.update_slots(SlotUpdate::Key(key), None);
        }
        result
    }

    /// Apply a namespace removal made during replay, and check it against the
    /// recording. Returns the recorded result.
    ///
    /// # Errors
    ///
    /// As [`Self::cache_invalidate`].
    pub(crate) fn cache_invalidate_namespace(
        &self,
        namespace: &str,
    ) -> Result<(), crate::cache::InvalidationError> {
        let result = if self.legacy_v3 {
            Ok(())
        } else {
            self.removal_result(&CacheWrite::InvalidateNamespace {
                namespace: namespace.to_owned(),
                error: None,
            })
        };
        if result.is_ok() {
            self.update_slots(SlotUpdate::Namespace(namespace), None);
        }
        result
    }

    /// Apply a removal of all entries made during replay, and check it
    /// against the recording.
    pub(crate) fn cache_clear(&self) {
        self.update_slots(SlotUpdate::All, None);
        if !self.legacy_v3 {
            let _ = self.next_cache_write(&CacheWrite::Clear);
        }
    }

    /// Serve an untyped read during replay. A recorded untyped read is always
    /// a miss: capture marks a capsule with an untyped hit incomplete.
    pub(crate) fn cache_untyped_get(&self, key: &str) {
        // A v3 capsule did not record untyped reads.
        if !self.legacy_v3 {
            let _ = self.cache_get(key);
        }
    }

    /// Log an untyped write made during replay.
    ///
    /// Capture cannot record an untyped value, and marks such a capsule
    /// incomplete, so a replayed untyped write is always unrecorded.
    pub(crate) fn cache_untyped_insert(&self, key: &str) {
        if self.legacy_v3 {
            return;
        }
        // No recorded key teaches the verdict what to mask, and a key can
        // hold a secret, so it is not printed.
        let _ = key;
        self.diverge(EffectDivergence {
            seam: EffectSeam::Cache,
            kind: EffectDivergenceKind::Unrecorded,
            index: 0,
            expected: None,
            actual: "an untyped cache write (key withheld)".to_owned(),
            detail: "the replayed run wrote an untyped cache value; a capsule cannot record \
                     one, so the write was not applied"
                .to_owned(),
        });
    }

    /// Log a direct `Cache::get_value` made during replay, outside the
    /// cache helpers.
    ///
    /// Capture marks a capsule with a direct read incomplete, so the
    /// recording had none. The read does not consume a read a helper
    /// recorded for the same key.
    pub(crate) fn cache_direct_get(&self, key: &str) {
        if self.legacy_v3 {
            return;
        }
        // As for an untyped write: no recorded key teaches the verdict what
        // to mask, so the key is not printed.
        let _ = key;
        self.diverge(EffectDivergence {
            seam: EffectSeam::Cache,
            kind: EffectDivergenceKind::Unrecorded,
            index: 0,
            expected: None,
            actual: "a direct cache read (key withheld)".to_owned(),
            detail: "the replayed run read the cache backend directly rather than through                      a cache helper; a capsule cannot record such a read, so it was a miss"
                .to_owned(),
        });
    }

    /// 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
    }

    /// Log work the replayed run started on a detached task. Capture marks
    /// a capsule with detached work incomplete, so the recording had none.
    /// It is logged at the spawn, so the verdict sees it even when the task
    /// runs after the response.
    pub(crate) fn detached_work_started(&self) {
        if self.legacy_v3 {
            return;
        }
        self.diverge(EffectDivergence {
            seam: EffectSeam::Detached,
            kind: EffectDivergenceKind::Unrecorded,
            index: 0,
            expected: None,
            actual: "a detached task".to_owned(),
            detail: "the replayed run started work on a detached task; the recorded run \
                     started none"
                .to_owned(),
        });
    }

    /// Set the read map the way a write sets the live cache, so a later read
    /// in the same run finds what production found.
    fn update_slots(&self, target: SlotUpdate<'_>, value: Option<&[u8]>) {
        let Ok(mut cache) = self.cache.lock() else {
            return;
        };
        match target {
            SlotUpdate::Key(key) => {
                // Resolved like a read, so a write to a key the capsule holds
                // masked lands on the recorded slot.
                if let Some(slot) =
                    resolve_slot(&cache, key).and_then(|recorded| cache.get_mut(&recorded))
                {
                    slot.value = value.map(<[u8]>::to_vec);
                } else if let Some(value) = value {
                    cache.insert(
                        key.to_owned(),
                        CacheSlot {
                            value: Some(value.to_vec()),
                            was_read: false,
                            touched: false,
                        },
                    );
                }
            }
            SlotUpdate::Namespace(namespace) => {
                for (key, slot) in cache.iter_mut() {
                    if in_namespace(key, namespace) {
                        slot.value = None;
                    }
                }
            }
            SlotUpdate::All => {
                for slot in cache.values_mut() {
                    slot.value = None;
                }
            }
        }
    }

    /// The result a replayed removal returns: the recorded one, or a failure
    /// when the removal diverged.
    fn removal_result(&self, actual: &CacheWrite) -> Result<(), crate::cache::InvalidationError> {
        match self.next_cache_write(actual) {
            Some(
                CacheWrite::Invalidate {
                    error: Some(error), ..
                }
                | CacheWrite::InvalidateNamespace {
                    error: Some(error), ..
                },
            ) => Err(crate::cache::InvalidationError::new(
                error.attempts,
                error.reason,
            )),
            Some(_) => Ok(()),
            None => Err(crate::cache::InvalidationError::new(
                1,
                "the replayed run removed a cache entry the capsule has no recording for",
            )),
        }
    }

    /// Consume the next recorded cache write when it is the one the run just
    /// made. Logs a divergence and returns `None` when it is not.
    fn next_cache_write(&self, actual: &CacheWrite) -> Option<CacheWrite> {
        let Ok(mut seam) = self.cache_writes.lock() else {
            return None;
        };
        let index = seam.consumed;
        let Some(next) = seam.pending.front() else {
            drop(seam);
            // No recorded identifier teaches the verdict what to mask, and a
            // key can hold a secret, so it is not printed.
            let described = actual.describe_withheld();
            self.diverge(EffectDivergence {
                seam: EffectSeam::Cache,
                kind: EffectDivergenceKind::Unrecorded,
                index,
                expected: None,
                actual: described.clone(),
                detail: format!(
                    "the replayed run made cache write {described:?}, which the capsule has no \
                     recording for"
                ),
            });
            return None;
        };
        if !next.matches(actual) {
            let expected = next.describe();
            let recorded_identifier = next.identifier().map(ToOwned::to_owned);
            drop(seam);
            // The verdict masks a value it learned. A masked identifier that
            // does not match teaches nothing, so it is not printed.
            let described = if self.observe_identifier(recorded_identifier.as_deref(), actual) {
                actual.describe()
            } else {
                actual.describe_withheld()
            };
            self.diverge(EffectDivergence {
                seam: EffectSeam::Cache,
                kind: EffectDivergenceKind::Mismatch,
                index,
                expected: Some(expected.clone()),
                actual: described.clone(),
                detail: format!(
                    "the replayed run made cache write {described:?}, but the capsule's next \
                     recorded write was {expected:?}"
                ),
            });
            return None;
        }
        let recorded = seam.pending.pop_front();
        seam.consumed = seam.consumed.saturating_add(1);
        drop(seam);
        if let Some(recorded) = recorded.as_ref() {
            self.observe_write(recorded, actual);
        }
        self.served.fetch_add(1, Ordering::SeqCst);
        recorded
    }

    /// Whether the next recorded mail send is the one the run just made, and
    /// the delivery error the recording produced for it.
    ///
    #[cfg(any(test, feature = "mail"))]
    #[must_use]
    pub(crate) fn next_mail(&self, sent: &SentMail<'_>) -> MailVerdict {
        let (to, subject) = (sent.to, sent.subject);
        let Ok(mut seam) = self.mail.lock() else {
            return MailVerdict::Diverged;
        };
        let index = seam.consumed;
        let Some(next) = seam.pending.front() else {
            drop(seam);
            let actual = describe_mail(to, subject);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Mail,
                kind: EffectDivergenceKind::Unrecorded,
                index,
                expected: None,
                actual: actual.clone(),
                detail: format!(
                    "the replayed run sent mail the capsule has no recording for ({actual}); \
                     nothing was delivered"
                ),
            });
            return MailVerdict::Diverged;
        };
        let recipients_match = next.to.len() == to.len()
            && next
                .to
                .iter()
                .zip(to.iter())
                .all(|(recorded, actual)| matches_redacted(recorded, actual));
        if !recipients_match || !matches_redacted(&next.subject, subject) {
            let expected = describe_mail(&next.to, &next.subject);
            drop(seam);
            let actual = describe_mail(to, subject);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Mail,
                kind: EffectDivergenceKind::Mismatch,
                index,
                expected: Some(expected.clone()),
                actual: actual.clone(),
                detail: format!(
                    "the replayed run sent {actual}, but the capsule's next recorded send was \
                     {expected}"
                ),
            });
            return MailVerdict::Diverged;
        }
        // Same envelope, different letter. Consuming the send here would hand
        // back the recorded success for a message whose contents — or whose
        // sender — the code now gets wrong, and the run would grade clean.
        // The sender is compared only when the replayed run produced one. A
        // message that sets no `from` inherits `[mail] from` at send time, so
        // the recorded address is deployment *configuration* rather than
        // handler behaviour — and a replay deliberately boots without mail
        // config. Reading that absence as "the code changed its sender" is the
        // same false signal as reporting a subsystem replay never booted. A
        // sender the run did choose, and chose differently, is a real change.
        let changed = mail_change(next, sent);
        if let Some(changed) = changed {
            let expected = describe_mail(&next.to, &next.subject);
            let (recorded_to, recorded_subject) = (next.to.clone(), next.subject.clone());
            drop(seam);
            // The envelope matched: learn what its masked values stood for
            // before the report prints the live subject and domains.
            for (recorded, actual) in recorded_to.iter().zip(to.iter()) {
                self.observe_text(recorded, actual);
            }
            self.observe_text(&recorded_subject, subject);
            let actual = describe_mail(to, subject);
            self.diverge(EffectDivergence {
                seam: EffectSeam::Mail,
                kind: EffectDivergenceKind::Mismatch,
                index,
                expected: Some(expected),
                actual: actual.clone(),
                detail: format!(
                    "the replayed run sent {actual} to the recipients the capsule recorded, but \
                     changed {changed}; nothing was delivered"
                ),
            });
            return MailVerdict::Diverged;
        }
        let recorded = seam.pending.pop_front();
        seam.consumed = seam.consumed.saturating_add(1);
        drop(seam);
        if let Some(recorded) = recorded.as_ref() {
            self.observe_mail(recorded, sent);
        }
        let error =
            recorded.and_then(|recorded| recorded.error.map(|text| (text, recorded.error_kind)));
        self.served.fetch_add(1, Ordering::SeqCst);
        error.map_or(MailVerdict::Sent, |(text, kind)| {
            MailVerdict::Failed(text, kind)
        })
    }

    /// What the recording says about the tenant lookup.
    ///
    /// A lookup the recording never made is a divergence (#2351 item 13):
    /// live resolution would succeed on input the recording never had.
    #[must_use]
    pub(crate) fn tenant(&self) -> TenantVerdict {
        let first_read = !self.tenant_read.swap(true, Ordering::SeqCst);
        match &self.tenant {
            RecordedTenant::Resolved(id) => TenantVerdict::Resolved(id.clone()),
            RecordedTenant::Failed => TenantVerdict::RecordedFailure,
            // A v3 capsule did not record a failed lookup.
            RecordedTenant::None if self.legacy_v3 => TenantVerdict::Unchecked,
            RecordedTenant::None => {
                if first_read {
                    self.diverge(EffectDivergence {
                        seam: EffectSeam::Tenant,
                        kind: EffectDivergenceKind::Unrecorded,
                        index: 0,
                        expected: None,
                        actual: "tenant lookup".to_owned(),
                        detail: "the replayed run resolved a tenant, which the capsule has no \
                                 recording for; live tenant resolution was not consulted"
                            .to_owned(),
                    });
                }
                TenantVerdict::Unrecorded
            }
        }
    }

    /// Log a lookup the recording saw fail and the replayed run resolved.
    /// The caller fails closed.
    pub(crate) fn tenant_resolved_after_recorded_failure(&self) {
        self.diverge(EffectDivergence {
            seam: EffectSeam::Tenant,
            kind: EffectDivergenceKind::Mismatch,
            index: 0,
            expected: Some("a failed tenant lookup".to_owned()),
            actual: "a resolved tenant".to_owned(),
            detail: "the recorded tenant lookup failed, but the replayed run resolved a tenant; \
                     the run would continue under a tenant production never had"
                .to_owned(),
        });
    }

    /// Log egress a subsystem tried to make outside the outbound-HTTP seam.
    /// The caller refuses the call.
    pub(crate) fn refuse_unrecorded_egress(&self, subsystem: &str, method: &str, url: &str) {
        let index = self.http.lock().map_or(0, |seam| seam.consumed);
        let actual = describe_http(method, url);
        self.diverge(EffectDivergence {
            seam: EffectSeam::Http,
            kind: EffectDivergenceKind::Unrecorded,
            index,
            expected: None,
            actual: actual.clone(),
            detail: format!(
                "the replayed run's {subsystem} made {actual} outside the recorded outbound-HTTP \
                 seam; it was refused rather than sent"
            ),
        });
    }

    /// The values that stood behind a placeholder in the recorded effects the
    /// run matched.
    #[must_use]
    pub(crate) fn observed_redactions(&self) -> crate::capsule::redact::RedactedValues {
        self.observed
            .lock()
            .map(|values| values.clone())
            .unwrap_or_default()
    }

    /// Keep what each placeholder in `recorded` stood for in `actual`.
    fn observe_text(&self, recorded: &str, actual: &str) {
        let Some(spans) = redacted_spans(recorded, actual) else {
            return;
        };
        if spans.is_empty() {
            return;
        }
        // A URL-encoded placeholder stood for an encoded value. Capture kept
        // its decoded spelling too, so keep it here.
        if let Ok(mut observed) = self.observed.lock() {
            for (span, encoded) in spans {
                observed.insert(span.as_bytes());
                if encoded {
                    let decoded = form_decoded(span);
                    if !decoded.is_empty() && decoded != span {
                        observed.insert(decoded.as_bytes());
                    }
                }
            }
        }
    }

    /// [`Self::observe_text`] for a URL, which also decomposes a masked
    /// userinfo the way capture retained it (user, password, decoded forms).
    fn observe_url(&self, recorded: &str, actual: &str) {
        let recorded = crate::capsule::redact::parser_view(recorded);
        let actual = crate::capsule::redact::parser_view(actual);
        let (recorded, actual) = (recorded.as_ref(), actual.as_ref());
        self.observe_text(recorded, actual);
        if crate::capsule::redact::url_userinfo(recorded).as_deref() == Some("[FILTERED]") {
            self.observe_live_userinfo(actual);
        }
    }

    /// Keep the live URL's userinfo, so an outcome that quotes it is scrubbed
    /// even when the call diverges before the recorded URL is compared.
    fn observe_live_userinfo(&self, url: &str) {
        if let Some(userinfo) = crate::capsule::redact::url_userinfo(url)
            && let Ok(mut observed) = self.observed.lock()
        {
            crate::capsule::redact::record_userinfo_components(&userinfo, &mut observed);
        }
    }

    fn observe_headers(&self, recorded: &[(String, String)], actual: &[(String, String)]) {
        for ((name, recorded), (_, actual)) in recorded.iter().zip(actual.iter()) {
            self.observe_text(recorded, actual);
            // Capture also kept the parts of a credential header (the token
            // of `Bearer <token>`), so an outcome that quotes one part was
            // masked. Keep the same parts here.
            if recorded != actual
                && let Ok(mut observed) = self.observed.lock()
            {
                crate::capsule::redact::record_credential_components(
                    name,
                    actual.as_bytes(),
                    &mut observed,
                );
            }
        }
    }

    fn observe_body(&self, recorded: &CapsuleBody, actual: &CapsuleBody) {
        let (recorded, actual) = (body_bytes(recorded), body_bytes(actual));
        if let (Ok(recorded), Ok(actual)) =
            (std::str::from_utf8(&recorded), std::str::from_utf8(&actual))
        {
            self.observe_text(recorded, actual);
        }
    }

    fn observe_json(&self, recorded: &serde_json::Value, actual: &serde_json::Value) {
        use serde_json::Value;
        match (recorded, actual) {
            (Value::String(recorded), Value::String(actual)) => {
                self.observe_text(recorded, actual);
            }
            (Value::String(recorded), actual) if recorded == FILTERED => {
                if let Ok(mut observed) = self.observed.lock() {
                    crate::capsule::redact::record_masked_value(actual, &mut observed);
                }
            }
            (Value::Object(recorded), Value::Object(actual)) => {
                for (key, recorded) in recorded {
                    if let Some(actual) = actual.get(key) {
                        self.observe_json(recorded, actual);
                    }
                }
            }
            (Value::Array(recorded), Value::Array(actual)) => {
                for (recorded, actual) in recorded.iter().zip(actual.iter()) {
                    self.observe_json(recorded, actual);
                }
            }
            _ => {}
        }
    }

    #[cfg(any(test, feature = "mail"))]
    fn observe_mail(&self, recorded: &MailEffect, sent: &SentMail<'_>) {
        for (recorded, actual) in recorded.to.iter().zip(sent.to.iter()) {
            self.observe_text(recorded, actual);
        }
        self.observe_text(&recorded.subject, sent.subject);
        if let (Some(recorded), Some(actual)) = (recorded.from.as_deref(), sent.from) {
            self.observe_text(recorded, actual);
        }
        self.observe_body(&recorded.body, sent.body);
        self.observe_body(&recorded.alternate_body, sent.alternate_body);
        if let (Some(recorded), Some(actual)) = (recorded.reply_to.as_deref(), sent.reply_to) {
            self.observe_text(recorded, actual);
        }
        if let (Some(recorded), Some(actual)) =
            (recorded.list_unsubscribe.as_deref(), sent.list_unsubscribe)
        {
            self.observe_text(recorded, actual);
        }
        self.observe_headers(&recorded.extra_headers, sent.extra_headers);
        for (recorded, actual) in recorded.attachments.iter().zip(sent.attachments.iter()) {
            self.observe_text(&recorded.filename, &actual.filename);
        }
    }

    /// Learn what a masked recorded key or namespace stood for in `actual`.
    /// Returns whether `actual`'s identifier is safe to print.
    fn observe_identifier(&self, recorded: Option<&str>, actual: &CacheWrite) -> bool {
        let (Some(recorded), Some(actual)) = (recorded, actual.identifier()) else {
            return true;
        };
        // An unmasked recorded identifier: print only the same one. Another
        // one can hold a secret that the scrub has not learned.
        if !recorded.contains(FILTERED) && !recorded.contains(FILTERED_URLENCODED) {
            return recorded == actual;
        }
        if redacted_spans(recorded, actual).is_none() {
            return false;
        }
        self.observe_text(recorded, actual);
        true
    }

    fn observe_write(&self, recorded: &CacheWrite, actual: &CacheWrite) {
        match (recorded, actual) {
            (
                CacheWrite::Insert { key, value, .. },
                CacheWrite::Insert {
                    key: actual_key,
                    value: actual_value,
                    ..
                },
            ) => {
                self.observe_text(key, actual_key);
                if let (Some(recorded), Some(actual)) = (value, actual_value)
                    && let (Ok(recorded), Ok(actual)) =
                        (std::str::from_utf8(recorded), std::str::from_utf8(actual))
                {
                    self.observe_text(recorded, actual);
                }
            }
            (CacheWrite::Invalidate { key, .. }, CacheWrite::Invalidate { key: actual, .. })
            | (
                CacheWrite::InvalidateNamespace { namespace: key, .. },
                CacheWrite::InvalidateNamespace {
                    namespace: actual, ..
                },
            ) => self.observe_text(key, actual),
            _ => {}
        }
    }

    /// Close the tape and report every recorded effect the run never asked
    /// for, alongside the divergences it already logged.
    #[must_use]
    pub fn finish(&self) -> Vec<EffectDivergence> {
        let mut divergences = self.divergences();
        if let Ok(seam) = self.http.lock()
            && !seam.pending.is_empty()
        {
            divergences.push(unconsumed(
                EffectSeam::Http,
                seam.consumed,
                seam.pending.len(),
                seam.pending
                    .front()
                    .map(|next| format!("{} {}", next.method, next.url)),
            ));
        }
        if let Ok(seam) = self.jobs.lock()
            && !seam.pending.is_empty()
        {
            divergences.push(unconsumed(
                EffectSeam::Job,
                seam.consumed,
                seam.pending.len(),
                seam.pending.front().map(|next| next.name.clone()),
            ));
        }
        if let Ok(seam) = self.mail.lock()
            && !seam.pending.is_empty()
        {
            divergences.push(unconsumed(
                EffectSeam::Mail,
                seam.consumed,
                seam.pending.len(),
                seam.pending.front().map(|next| next.subject.clone()),
            ));
        }
        let recorded_tenant = match &self.tenant {
            RecordedTenant::None => None,
            RecordedTenant::Resolved(id) => Some(id.clone()),
            RecordedTenant::Failed => Some("a failed tenant lookup".to_owned()),
        };
        if let Some(recorded) = recorded_tenant
            && !self.tenant_read.load(Ordering::SeqCst)
        {
            divergences.push(unconsumed(EffectSeam::Tenant, 0, 1, Some(recorded)));
        }
        if let Ok(seam) = self.cache_writes.lock()
            && !seam.pending.is_empty()
        {
            divergences.push(unconsumed(
                EffectSeam::Cache,
                seam.consumed,
                seam.pending.len(),
                seam.pending.front().map(CacheWrite::describe),
            ));
        }
        if let Ok(cache) = self.cache.lock() {
            let missed: Vec<&String> = cache
                .iter()
                .filter(|(_, slot)| slot.was_read && !slot.touched)
                .map(|(key, _)| key)
                .collect();
            if !missed.is_empty() {
                divergences.push(unconsumed(
                    EffectSeam::Cache,
                    0,
                    missed.len(),
                    missed.first().map(|key| (*key).clone()),
                ));
            }
        }
        divergences
    }
}

/// The divergence a draw of a different width than the recording reports
/// (#2351 item 11).
pub(crate) fn random_width_divergence(
    index: usize,
    recorded: usize,
    actual: usize,
) -> EffectDivergence {
    EffectDivergence {
        seam: EffectSeam::Random,
        kind: EffectDivergenceKind::Mismatch,
        index,
        expected: Some(format!("{recorded} byte(s)")),
        actual: format!("{actual} byte(s)"),
        detail: format!(
            "the replayed run drew {actual} random byte(s) where the recording drew {recorded}; \
             the value it minted is not the one production used"
        ),
    }
}

/// The divergence a seam reports for recorded effects nobody asked for.
fn unconsumed(
    seam: EffectSeam,
    consumed: usize,
    remaining: usize,
    next: Option<String>,
) -> EffectDivergence {
    let label = seam.label();
    let next_text = next.unwrap_or_else(|| "?".to_owned());
    EffectDivergence {
        seam,
        kind: EffectDivergenceKind::Unconsumed,
        index: consumed,
        expected: Some(next_text.clone()),
        actual: String::new(),
        detail: format!(
            "{remaining} recorded {label} effect(s) were never asked for by the replayed run \
             (the next one was {next_text}); reaching the recorded outcome without the \
             recorded effects is not a reproduction"
        ),
    }
}

/// Decode a base64 field, treating an undecodable one as absent.
fn decode(encoded: &str) -> Option<Vec<u8>> {
    STANDARD.decode(encoded.as_bytes()).ok()
}

/// The recorded body of an effect, as the bytes a replayed caller receives.
#[must_use]
pub fn body_bytes(body: &CapsuleBody) -> Vec<u8> {
    match body {
        CapsuleBody::Absent | CapsuleBody::Skipped { .. } => Vec::new(),
        CapsuleBody::Text(text) => text.as_bytes().to_vec(),
        CapsuleBody::Base64(encoded) => decode(encoded).unwrap_or_default(),
    }
}

/// How an outbound call is named in a divergence report.
///
/// The query string is dropped. The `actual` side of a divergence is what the
/// *live* run produced — a real API key, a real token — and the report is
/// printed to a terminal and into CI logs, right beside a recorded side that
/// redaction dutifully masked. Reporting the path is what an operator needs to
/// see; reporting the credential is a leak the capsule format went to some
/// trouble to prevent.
fn describe_http(method: &str, url: &str) -> String {
    let path = url.split_once('?').map_or(url, |(head, _)| head);
    let elided = if path.len() < url.len() { "?…" } else { "" };
    // The live URL's userinfo is a credential, and this text reaches replay's
    // JSON and terminal output, so it is dropped like the query string is.
    let path = crate::capsule::redact::hide_url_userinfo(path);
    format!("{method} {path}{elided}")
}

/// How a job enqueue is named in a divergence report.
///
/// The job name and the payload's *shape* — its keys — but not its values, for
/// the same reason [`describe_http`] drops the query string.
fn describe_job(name: &str, payload: &serde_json::Value) -> String {
    match payload {
        serde_json::Value::Object(map) => {
            let keys: Vec<&str> = map.keys().map(String::as_str).collect();
            format!("{name} {{{}}}", keys.join(", "))
        }
        serde_json::Value::Null => format!("{name} (null payload)"),
        other => format!("{name} ({} payload)", json_kind(other)),
    }
}

/// The bare type name of a JSON value, for [`describe_job`].
const fn json_kind(value: &serde_json::Value) -> &'static str {
    match value {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(_) => "boolean",
        serde_json::Value::Number(_) => "number",
        serde_json::Value::String(_) => "string",
        serde_json::Value::Array(_) => "array",
        serde_json::Value::Object(_) => "object",
    }
}

/// What a replayed letter changed from the recorded one with the same
/// envelope, if anything.
#[cfg(any(test, feature = "mail"))]
fn mail_change(next: &MailEffect, sent: &SentMail<'_>) -> Option<&'static str> {
    let sender_match = sent.from.is_none_or(|actual| {
        next.from
            .as_deref()
            .is_some_and(|recorded| matches_redacted(recorded, actual))
    });
    // Everything a recipient would notice, not only what addresses the
    // envelope: an invoice removed, an HTML template rewritten under an
    // untouched text fallback, an unsubscribe header dropped. Each is a
    // materially different email, and serving the recorded success for one
    // would leave the final audit nothing to catch.
    if !sender_match {
        Some("its sender")
    } else if !body_matches_redacted(&next.body, sent.body) {
        Some("its body")
    } else if !body_matches_redacted(&next.alternate_body, sent.alternate_body) {
        Some("the other half of its multipart body")
    } else if !optional_matches_redacted(next.reply_to.as_deref(), sent.reply_to) {
        Some("its reply-to")
    } else if !optional_matches_redacted(next.list_unsubscribe.as_deref(), sent.list_unsubscribe) {
        Some("its list-unsubscribe header")
    } else if !headers_match_redacted(&next.extra_headers, sent.extra_headers) {
        Some("its headers")
    } else if !attachments_match_redacted(&next.attachments, sent.attachments) {
        Some("its attachments")
    } else {
        None
    }
}

/// How a mail send is named in a divergence report.
///
/// Recipients are the PII here, so only their count and domains reach the
/// report — the report is printed to a terminal and to CI logs.
#[cfg(any(test, feature = "mail"))]
fn describe_mail(to: &[String], subject: &str) -> String {
    let domains: Vec<&str> = to
        .iter()
        .map(|address| address.rsplit('@').next().unwrap_or("?"))
        .collect();
    format!(
        "{subject:?} -> {} recipient(s) at {}",
        to.len(),
        domains.join(", ")
    )
}

// ── Matching against redacted recordings ────────────────────────────────────

/// The placeholder redaction leaves behind, treated as a wildcard when the
/// tape is matched against a live call.
const FILTERED: &str = crate::log::filter::FILTERED_PLACEHOLDER;
/// Its percent-encoded spelling, which is what a masked *query* value looks
/// like in a recorded URL.
const FILTERED_URLENCODED: &str = "%5BFILTERED%5D";

/// Whether a recorded string matches what the replayed run actually produced.
///
/// This is the whole reason effect matching is not `==`. The capsule holds the
/// **redacted** spelling — `Bearer [FILTERED]`, `?token=%5BFILTERED%5D`,
/// `{"api_key":"[FILTERED]"}` — while the replayed handler produces the real
/// one, because it reads the real configuration and the real recorded request.
/// Comparing them literally would make every capsule that redacted anything
/// report a divergence on the first outbound call, which is precisely the false
/// "the code changed" verdict the whole design exists to avoid.
///
/// So each placeholder is a wildcard: the literal segments around it must
/// appear, in order, and nothing is asserted about what stood where the
/// placeholder is — the capsule does not carry those bytes, so nothing *can*
/// be asserted about them.
fn matches_redacted(recorded: &str, actual: &str) -> bool {
    redacted_spans(recorded, actual).is_some()
}

/// A form-encoded value, decoded as capture decodes a query or form value.
fn form_decoded(value: &str) -> String {
    url::form_urlencoded::parse(format!("={value}").as_bytes())
        .next()
        .map(|(_, decoded)| decoded.into_owned())
        .unwrap_or_default()
}

/// [`matches_redacted`], and on a match the text each placeholder stood for,
/// with whether that placeholder was the URL-encoded one.
///
/// A recorded text can hold both spellings (a value echoed into a path, and
/// a masked query parameter), so both are matched. Empty spans are left out:
/// there is nothing to mask.
fn redacted_spans<'a>(recorded: &str, actual: &'a str) -> Option<Vec<(&'a str, bool)>> {
    if recorded == actual {
        return Some(Vec::new());
    }
    let (segments, encoded) = split_placeholders(recorded);
    if encoded.is_empty() {
        // No placeholder and not equal: an ordinary difference.
        return None;
    }
    let mut spans = Vec::new();
    let mut push = |span: &'a str, encoded: bool| {
        if !span.is_empty() {
            spans.push((span, encoded));
        }
    };
    let mut rest = actual;
    let mut segments = segments.into_iter();
    rest = rest.strip_prefix(segments.next().unwrap_or_default())?;
    let mut segments = segments.zip(encoded).peekable();
    while let Some((segment, encoded)) = segments.next() {
        let last = segments.peek().is_none();
        if segment.is_empty() {
            // Back-to-back placeholders, or one at the very end: nothing to
            // anchor on. At the end, the placeholder takes what remains.
            if last {
                push(rest, encoded);
                rest = "";
            }
        } else if last {
            // The final literal segment has to be the *tail*, or a recorded
            // `?a=1` would match an actual `?a=1&b=2`.
            let head = rest.strip_suffix(segment)?;
            push(head, encoded);
            rest = "";
        } else {
            let found = rest.find(segment)?;
            push(rest.get(..found).unwrap_or_default(), encoded);
            let after = found.saturating_add(segment.len());
            rest = rest.get(after..).unwrap_or_default();
        }
    }
    Some(spans)
}

/// Whether two placeholders touch in a recorded text, so that nothing marks
/// where one masked value ends and the next begins.
pub(crate) fn has_adjacent_placeholders(recorded: &str) -> bool {
    let (segments, _) = split_placeholders(recorded);
    let between = segments.len().saturating_sub(2);
    segments
        .iter()
        .skip(1)
        .take(between)
        .any(|segment| segment.is_empty())
}

/// Split a recorded text at its placeholders, in either spelling. Returns
/// the literal segments (one more than the placeholders), and for each
/// placeholder whether it was the URL-encoded one.
fn split_placeholders(recorded: &str) -> (Vec<&str>, Vec<bool>) {
    let mut segments = Vec::new();
    let mut encoded = Vec::new();
    let mut rest = recorded;
    loop {
        let next = match (rest.find(FILTERED), rest.find(FILTERED_URLENCODED)) {
            (Some(raw), Some(url)) if url < raw => Some((url, FILTERED_URLENCODED, true)),
            (Some(raw), _) => Some((raw, FILTERED, false)),
            (None, Some(url)) => Some((url, FILTERED_URLENCODED, true)),
            (None, None) => None,
        };
        let Some((at, placeholder, is_encoded)) = next else {
            segments.push(rest);
            return (segments, encoded);
        };
        segments.push(rest.get(..at).unwrap_or_default());
        encoded.push(is_encoded);
        rest = rest
            .get(at.saturating_add(placeholder.len())..)
            .unwrap_or_default();
    }
}

/// The request half of an outbound call, as replay sees it.
///
/// The same three fields `OutboundRecorder::arm` records, so the two halves of
/// a comparison are derived the same way: matching on method and URL alone
/// would serve a recorded response to a handler that now sends a *different*
/// body to the same endpoint — charging a different amount — and then, with no
/// effect left over for the final audit to notice, grade the unchanged failure
/// `reproduced`.
pub(crate) struct OutboundRequest<'a> {
    /// The HTTP method, as the caller spelled it.
    pub method: &'a str,
    /// The full URL, query string included.
    pub url: &'a str,
    /// The headers the *caller* set, in the order they were set — the same
    /// subset `arm` records, so client-injected headers cannot diverge a run.
    pub headers: &'a [(String, String)],
    /// The body the caller supplied.
    pub body: &'a CapsuleBody,
}

/// One message a replayed run handed to the mailer.
///
/// Derived exactly as `mail_effect` derives the recorded side, so the two
/// compare like with like: matching on recipients and subject alone would
/// serve the recorded success to a run that now sends different contents, or
/// sends them from the wrong address, and leave the audit nothing to notice.
#[cfg(any(test, feature = "mail"))]
pub(crate) struct SentMail<'a> {
    /// The recipients, in the order the caller listed them.
    pub to: &'a [String],
    /// The envelope sender, when the message set one.
    pub from: Option<&'a str>,
    /// The subject line.
    pub subject: &'a str,
    /// The body, preferring plain text over HTML — the same preference the
    /// recorder applies.
    pub body: &'a CapsuleBody,
    /// The other alternative of a multipart message.
    pub alternate_body: &'a CapsuleBody,
    /// `Reply-To`, when set.
    pub reply_to: Option<&'a str>,
    /// `List-Unsubscribe`, when set.
    pub list_unsubscribe: Option<&'a str>,
    /// Caller-set headers beyond the named ones.
    pub extra_headers: &'a [(String, String)],
    /// What the message carries, by name and digest.
    pub attachments: &'a [crate::capsule::schema::MailAttachmentEffect],
}

/// Whether a recorded body matches what the replayed run produced.
///
/// A `Skipped` recording is treated as matching: the bytes were deliberately
/// never captured, so there is nothing to compare against, and a capsule
/// carrying one is refused before it reaches replay anyway.
fn body_matches_redacted(recorded: &CapsuleBody, actual: &CapsuleBody) -> bool {
    if matches!(recorded, CapsuleBody::Skipped { .. }) {
        return true;
    }
    match (recorded, actual) {
        (CapsuleBody::Absent, CapsuleBody::Absent) => true,
        // Absent and empty are different facts on the wire — a POST with no
        // body is not a POST with a zero-length one — so they are not folded
        // together here.
        (CapsuleBody::Absent, _) | (_, CapsuleBody::Absent) => false,
        _ => bytes_match_redacted(&body_bytes(recorded), &body_bytes(actual)),
    }
}

/// Whether the caller-set headers a run sent match the ones recorded.
///
/// Order and count are compared exactly — a header the code stopped sending is
/// a change — while a masked value matches whatever stood in its place, for the
/// same reason [`matches_redacted`] exists.
/// Whether the attachments a run sent match the ones recorded.
///
/// The filename is free-form text the redaction sweep masks like any other, so
/// it compares through [`matches_redacted`] — a masked one stands for whatever
/// was really there, exactly as a masked header value does. Everything else is
/// exact: a content type, a length and a digest are not text a filter would
/// ever rewrite, and they are what actually establish that the same file is
/// still being sent.
#[cfg(any(test, feature = "mail"))]
fn attachments_match_redacted(
    recorded: &[crate::capsule::schema::MailAttachmentEffect],
    actual: &[crate::capsule::schema::MailAttachmentEffect],
) -> bool {
    recorded.len() == actual.len()
        && recorded
            .iter()
            .zip(actual.iter())
            .all(|(recorded, actual)| {
                matches_redacted(&recorded.filename, &actual.filename)
                    && recorded.content_type == actual.content_type
                    && recorded.len == actual.len
                    && recorded.sha256 == actual.sha256
            })
}

/// Whether an optional recorded field matches what the replayed run set.
///
/// Both absent matches; one present and one absent is a change; two present
/// compare through [`matches_redacted`], so a masked value still stands for
/// whatever was really there.
///
/// Gated with its callers: every one of them is in the mail comparison.
#[cfg(any(test, feature = "mail"))]
fn optional_matches_redacted(recorded: Option<&str>, actual: Option<&str>) -> bool {
    match (recorded, actual) {
        (None, None) => true,
        (Some(recorded), Some(actual)) => matches_redacted(recorded, actual),
        _ => false,
    }
}

fn headers_match_redacted(recorded: &[(String, String)], actual: &[(String, String)]) -> bool {
    recorded.len() == actual.len()
        && recorded.iter().zip(actual.iter()).all(
            |((recorded_name, recorded_value), (actual_name, actual_value))| {
                recorded_name.eq_ignore_ascii_case(actual_name)
                    && matches_redacted(recorded_value, actual_value)
            },
        )
}

/// Which halves of an outbound request half differed, named but never quoted.
///
/// A header *value* and a body are production data — the whole reason the
/// capsule redacts — and a divergence report is printed to a terminal and into
/// CI logs. So this says *what* changed, and for headers which names, and
/// leaves the operator to read the capsule for the rest.
fn describe_request_mismatch(recorded: &HttpEffect, actual: &OutboundRequest<'_>) -> String {
    let mut parts = Vec::new();
    if !headers_match_redacted(&recorded.request_headers, actual.headers) {
        let names: Vec<&str> = recorded
            .request_headers
            .iter()
            .map(|(name, _)| name.as_str())
            .collect();
        parts.push(format!("its headers ({} recorded)", names.join(", ")));
    }
    if !body_matches_redacted(&recorded.request_body, actual.body) {
        parts.push("its body".to_owned());
    }
    parts.join(" and ")
}

/// Whether recorded bytes match what the replayed run produced, tolerating
/// redaction placeholders in a UTF-8 recording.
///
/// Binary values are compared exactly: there is nothing in them for a
/// placeholder to stand in.
fn bytes_match_redacted(recorded: &[u8], actual: &[u8]) -> bool {
    if recorded == actual {
        return true;
    }
    match (std::str::from_utf8(recorded), std::str::from_utf8(actual)) {
        (Ok(recorded), Ok(actual)) => matches_redacted(recorded, actual),
        _ => false,
    }
}

/// Whether a recorded JSON payload matches the one the replayed run produced,
/// with redaction placeholders treated as wildcards.
///
/// Structure is compared exactly — a payload that gained or lost a key *is* a
/// change — but a masked leaf matches whatever stood there, for the same reason
/// [`matches_redacted`] exists.
fn json_matches_redacted(recorded: &serde_json::Value, actual: &serde_json::Value) -> bool {
    use serde_json::Value;
    match (recorded, actual) {
        (Value::String(recorded), Value::String(actual)) => matches_redacted(recorded, actual),
        // A masked value need not have been a string: `{"pin": 1234}` is
        // recorded as `{"pin": "[FILTERED]"}`.
        (Value::String(recorded), _) if recorded == FILTERED => true,
        (Value::Object(recorded), Value::Object(actual)) => {
            recorded.len() == actual.len()
                && recorded.iter().all(|(key, recorded_value)| {
                    actual.get(key).is_some_and(|actual_value| {
                        json_matches_redacted(recorded_value, actual_value)
                    })
                })
        }
        (Value::Array(recorded), Value::Array(actual)) => {
            recorded.len() == actual.len()
                && recorded
                    .iter()
                    .zip(actual.iter())
                    .all(|(recorded, actual)| json_matches_redacted(recorded, actual))
        }
        _ => recorded == actual,
    }
}

// ── Task-local installation ─────────────────────────────────────────────────

tokio::task_local! {
    /// The effect tape serving the replayed run on this task.
    ///
    /// Task-local for the same reason the replay clock's marker is: a
    /// `tokio::spawn`ed side task carried no capture scope when the failure was
    /// recorded, so it must not consume tape entries the recorded handler is
    /// still owed — and, just as importantly, several capsules must be able to
    /// replay concurrently inside one `cargo test` process.
    ///
    /// `capsule::spawn` and the framework's detached work carry the tape into
    /// their task on purpose. Capture marks such a capsule incomplete, so
    /// replay refuses it; the carried tape only makes a changed run fail
    /// closed rather than reach a live service.
    static REPLAY_EFFECTS: Arc<ReplayEffects>;
}

/// Run `future` with `tape` serving its effect seams.
pub async fn with_effect_tape<F: Future>(tape: Arc<ReplayEffects>, future: F) -> F::Output {
    REPLAY_EFFECTS.scope(tape, future).await
}

/// The effect tape serving the current task, if any.
#[must_use]
pub fn current_tape() -> Option<Arc<ReplayEffects>> {
    REPLAY_EFFECTS.try_with(Arc::clone).ok()
}

/// Whether a tape is serving the current task, without cloning the handle.
///
/// The presence test alone, for callers on the request path that would
/// otherwise pay an atomic refcount pair per request just to ask.
#[must_use]
pub fn tape_active() -> bool {
    REPLAY_EFFECTS
        .try_with(|tape| tape.mark_scope_entered())
        .is_ok()
}

#[cfg(test)]
mod tests {
    use crate::capsule::effects::{
        CachedValue, EffectDivergenceKind, EffectSeam, EnqueueVerdict, FILTERED, MailVerdict,
        OutboundRequest, ReplayEffects, SentMail,
    };

    /// An absent body to point at when a fixture has no alternative half.
    ///
    /// A `static` rather than a `const` so `&NO_BODY` borrows for `'static`:
    /// `CapsuleBody` owns a `String`, so a const would only ever yield a
    /// temporary that cannot outlive the `SentMail` borrowing it.
    static NO_BODY: CapsuleBody = CapsuleBody::Absent;
    use crate::capsule::schema::{
        CacheEffect, CapsuleBody, CapsuleEffects, HttpEffect, JobEffect, MailEffect, TenantEffect,
    };

    fn http(method: &str, url: &str, status: u16) -> HttpEffect {
        HttpEffect {
            method: method.to_owned(),
            url: url.to_owned(),
            request_headers: Vec::new(),
            request_body: CapsuleBody::Absent,
            status,
            response_headers: Vec::new(),
            response_body: CapsuleBody::Text("{}".to_owned()),
            error: None,
            ..Default::default()
        }
    }

    /// A request half with nothing but a method and a URL — what most of these
    /// fixtures record, so the request-half comparison is a no-op for them and
    /// each test keeps testing the one thing it names.
    fn call<'a>(method: &'a str, url: &'a str) -> OutboundRequest<'a> {
        OutboundRequest {
            method,
            url,
            headers: &[],
            body: &CapsuleBody::Absent,
        }
    }

    /// A message with the body and sender these fixtures record, so the
    /// contents comparison is satisfied and each test keeps testing the one
    /// thing it names.
    fn sent<'a>(to: &'a [String], subject: &'a str, body: &'a CapsuleBody) -> SentMail<'a> {
        SentMail {
            to,
            from: None,
            subject,
            body,
            alternate_body: &NO_BODY,
            reply_to: None,
            list_unsubscribe: None,
            extra_headers: &[],
            attachments: &[],
        }
    }

    /// A message with every delivery-relevant part spelled out, for the tests
    /// that are about those parts rather than about the envelope.
    fn sent_full<'a>(
        to: &'a [String],
        body: &'a CapsuleBody,
        alternate_body: &'a CapsuleBody,
        attachments: &'a [crate::capsule::schema::MailAttachmentEffect],
    ) -> SentMail<'a> {
        SentMail {
            to,
            from: Some("billing@shop.example"),
            subject: "Receipt",
            body,
            alternate_body,
            reply_to: None,
            list_unsubscribe: None,
            extra_headers: &[],
            attachments,
        }
    }

    #[test]
    fn a_redacted_recording_still_matches_the_call_the_handler_makes() {
        // The capsule holds the *masked* spelling; the replayed handler
        // produces the real one. Comparing literally would report a divergence
        // on the first outbound call of every capsule that redacted anything.
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![http(
                "GET",
                "https://api.example/items?key=%5BFILTERED%5D&page=2",
                200,
            )],
            jobs: vec![JobEffect {
                name: "notify".to_owned(),
                payload: serde_json::json!({"to": "a@example.com", "token": "[FILTERED]"}),
                delay_secs: None,
                due_at: None,
                requested_due_at: None,
                error: None,
                error_status: None,
            }],
            ..CapsuleEffects::default()
        });

        assert!(
            tape.next_http(&call(
                "GET",
                "https://api.example/items?key=sk-live-42&page=2"
            ))
            .is_some(),
            "a masked query value must match whatever really stood there"
        );
        assert_eq!(
            tape.next_job(
                "notify",
                &serde_json::json!({"to": "a@example.com", "token": "real-token"}),
                crate::job::EnqueueSchedule::Immediate
            ),
            EnqueueVerdict::Queued
        );
        assert!(tape.divergences().is_empty(), "{:?}", tape.divergences());
    }

    #[test]
    fn a_wildcard_does_not_swallow_a_genuine_difference() {
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![http(
                "GET",
                "https://api.example/items?key=%5BFILTERED%5D",
                200,
            )],
            ..CapsuleEffects::default()
        });
        // Same masked prefix, different path: still a divergence.
        assert!(
            tape.next_http(&call("GET", "https://api.example/other?key=sk-live-42"))
                .is_none()
        );
        assert_eq!(tape.divergences().len(), 1);
    }

    #[test]
    fn a_masked_json_payload_still_compares_structurally() {
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![JobEffect {
                name: "notify".to_owned(),
                payload: serde_json::json!({"token": "[FILTERED]"}),
                delay_secs: None,
                due_at: None,
                requested_due_at: None,
                error: None,
                error_status: None,
            }],
            ..CapsuleEffects::default()
        });
        // An extra key is a real change, placeholder or not.
        assert_eq!(
            tape.next_job(
                "notify",
                &serde_json::json!({"token": "real", "extra": 1}),
                crate::job::EnqueueSchedule::Immediate
            ),
            EnqueueVerdict::Diverged
        );
        assert_eq!(tape.divergences().len(), 1);
    }

    #[test]
    fn an_outbound_call_is_served_from_the_tape_in_recorded_order() {
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![
                http("GET", "https://a.example/one", 200),
                http("GET", "https://a.example/two", 503),
            ],
            ..CapsuleEffects::default()
        });

        let first = tape
            .next_http(&call("GET", "https://a.example/one"))
            .expect("the first recorded exchange serves the first call");
        assert_eq!(first.status, 200);
        let second = tape
            .next_http(&call("GET", "https://a.example/two"))
            .expect("the second call takes the second exchange");
        assert_eq!(second.status, 503);
        assert!(tape.divergences().is_empty(), "a faithful run diverges not");
    }

    #[test]
    fn an_outbound_call_the_capsule_never_recorded_is_a_divergence_not_a_live_call() {
        let tape = ReplayEffects::new(CapsuleEffects::default());
        assert!(
            tape.next_http(&call("POST", "https://payments.example/charge"))
                .is_none(),
            "an unrecorded call must never fall through to the network"
        );
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1);
        assert_eq!(divergences[0].seam, EffectSeam::Http);
        assert!(
            divergences[0].detail.contains("payments.example"),
            "the report must name the call: {}",
            divergences[0].detail
        );
    }

    #[test]
    fn an_outbound_call_to_a_different_url_than_recorded_is_a_divergence() {
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![http("GET", "https://a.example/one", 200)],
            ..CapsuleEffects::default()
        });
        assert!(
            tape.next_http(&call("GET", "https://b.example/other"))
                .is_none()
        );
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1);
        assert_eq!(
            divergences[0].expected.as_deref(),
            Some("GET https://a.example/one")
        );
    }

    #[test]
    fn a_recorded_call_the_replay_never_made_is_a_divergence_too() {
        // Symmetric with the database tape's `UnconsumedExchanges`: reaching
        // the recorded outcome without the recorded effects is not a
        // reproduction.
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![http("GET", "https://a.example/one", 200)],
            ..CapsuleEffects::default()
        });
        let divergences = tape.finish();
        assert_eq!(divergences.len(), 1);
        assert!(
            divergences[0].detail.contains("never"),
            "the report must say the recorded call was never made: {}",
            divergences[0].detail
        );
    }

    #[test]
    fn an_enqueue_is_asserted_against_the_tape_and_never_reaches_a_queue() {
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![JobEffect {
                name: "send_receipt".to_owned(),
                payload: serde_json::json!({"order": 7}),
                delay_secs: None,
                due_at: None,
                requested_due_at: None,
                error: None,
                error_status: None,
            }],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({"order": 7}),
                crate::job::EnqueueSchedule::Immediate
            ),
            EnqueueVerdict::Queued,
            "the recorded enqueue is satisfied from the tape"
        );
        assert!(tape.divergences().is_empty());
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({"order": 8}),
                crate::job::EnqueueSchedule::Immediate
            ),
            EnqueueVerdict::Diverged,
            "an enqueue the recording never made must not be satisfied"
        );
        assert_eq!(tape.divergences().len(), 1);
    }

    #[test]
    fn a_rescheduled_enqueue_is_a_divergence() {
        // A job that used to run immediately and now runs in an hour is a
        // behaviour change the capsule can see.
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![JobEffect {
                name: "send_receipt".to_owned(),
                payload: serde_json::json!({}),
                delay_secs: None,
                due_at: None,
                requested_due_at: None,
                error: None,
                error_status: None,
            }],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                crate::job::EnqueueSchedule::After(std::time::Duration::from_secs(3600))
            ),
            EnqueueVerdict::Diverged
        );
        assert_eq!(tape.divergences().len(), 1);
    }

    /// #2351 item 17: a plain `enqueue` must not consume a recorded delayed
    /// or future-deadline entry.
    #[test]
    fn a_relative_delay_is_compared_the_way_its_format_recorded_it() {
        let recorded = |delay_secs: i64| CapsuleEffects {
            jobs: vec![JobEffect {
                name: "send_receipt".to_owned(),
                payload: serde_json::json!({}),
                delay_secs: Some(delay_secs),
                ..JobEffect::default()
            }],
            ..CapsuleEffects::default()
        };
        let after = |millis: u64| {
            crate::job::EnqueueSchedule::After(std::time::Duration::from_millis(millis))
        };
        let verdict = |tape: &ReplayEffects, millis: u64| {
            tape.next_job("send_receipt", &serde_json::json!({}), after(millis))
        };
        // v4 rounds up: 500 ms was recorded as 1 s.
        assert_eq!(
            verdict(&ReplayEffects::new(recorded(1)), 500),
            EnqueueVerdict::Queued
        );
        // v3 cut it off: 500 ms was recorded as 0 s.
        let v3 = |secs| ReplayEffects::new(recorded(secs)).for_format_version(3);
        assert_eq!(verdict(&v3(0), 500), EnqueueVerdict::Queued);
        // Codex review on #3222: a whole-second change is still a change in v3.
        assert_eq!(verdict(&v3(5), 6_000), EnqueueVerdict::Diverged);
        assert_eq!(verdict(&v3(5), 5_000), EnqueueVerdict::Queued);
    }

    /// Codex review on #3222: capture runs an `enqueue_in` with a zero delay
    /// at once and records no delay, so a zero delay matches that entry.
    #[test]
    fn a_zero_relative_delay_matches_the_immediate_entry_it_recorded() {
        let immediate = CapsuleEffects {
            jobs: vec![JobEffect {
                name: "send_receipt".to_owned(),
                payload: serde_json::json!({}),
                ..JobEffect::default()
            }],
            ..CapsuleEffects::default()
        };
        let after = |millis: u64| {
            crate::job::EnqueueSchedule::After(std::time::Duration::from_millis(millis))
        };
        let tape = ReplayEffects::new(immediate.clone());
        assert_eq!(
            tape.next_job("send_receipt", &serde_json::json!({}), after(0)),
            EnqueueVerdict::Queued
        );
        assert!(tape.finish().is_empty(), "{:?}", tape.finish());
        // A positive delay is still not the immediate entry.
        let tape = ReplayEffects::new(immediate);
        assert_eq!(
            tape.next_job("send_receipt", &serde_json::json!({}), after(1_000)),
            EnqueueVerdict::Diverged
        );
    }

    #[test]
    fn an_immediate_enqueue_does_not_satisfy_a_recorded_delayed_one() {
        let delayed = JobEffect {
            name: "send_receipt".to_owned(),
            payload: serde_json::json!({}),
            delay_secs: Some(3600),
            ..JobEffect::default()
        };
        let future = JobEffect {
            due_at: Some(chrono::Utc::now() + chrono::Duration::hours(1)),
            ..delayed.clone()
        };
        for recorded in [delayed, future] {
            let tape = ReplayEffects::new(CapsuleEffects {
                jobs: vec![recorded],
                ..CapsuleEffects::default()
            });
            assert_eq!(
                tape.next_job(
                    "send_receipt",
                    &serde_json::json!({}),
                    crate::job::EnqueueSchedule::Immediate
                ),
                EnqueueVerdict::Diverged
            );
        }
        // A zero delay is still immediate.
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![JobEffect {
                name: "send_receipt".to_owned(),
                payload: serde_json::json!({}),
                delay_secs: Some(0),
                ..JobEffect::default()
            }],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                crate::job::EnqueueSchedule::Immediate
            ),
            EnqueueVerdict::Queued
        );
    }

    /// #2351 item 18: capture records a past deadline as an immediate
    /// enqueue, and keeps the deadline the caller asked for. Replay compares
    /// the same deadline.
    #[test]
    fn a_past_deadline_enqueue_at_matches_its_normalized_recording() {
        let past = chrono::Utc::now() - chrono::Duration::hours(1);
        let recorded = JobEffect {
            name: "send_receipt".to_owned(),
            payload: serde_json::json!({}),
            requested_due_at: Some(past),
            ..JobEffect::default()
        };
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![recorded.clone()],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                crate::job::EnqueueSchedule::At(past)
            ),
            EnqueueVerdict::Queued,
            "{:?}",
            tape.divergences()
        );
        // A plain enqueue is not the recorded `enqueue_at` (Codex review on
        // #3222).
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![recorded.clone()],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                crate::job::EnqueueSchedule::Immediate
            ),
            EnqueueVerdict::Diverged
        );
        // A different past deadline is a changed call.
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![recorded],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                crate::job::EnqueueSchedule::At(past - chrono::Duration::hours(1))
            ),
            EnqueueVerdict::Diverged
        );
    }

    /// Review fix: a v3 capsule did not record removals, untyped reads or a
    /// failed tenant lookup, so replay does not check those seams for it.
    #[test]
    fn a_v3_capsule_replays_with_the_v3_rules() {
        let tape = ReplayEffects::new(CapsuleEffects::default()).for_format_version(3);
        assert_eq!(tape.cache_invalidate("k"), Ok(()));
        tape.cache_clear();
        assert_eq!(tape.cache_invalidate_namespace("ns"), Ok(()));
        tape.cache_untyped_insert("k");
        tape.cache_untyped_get("k");
        tape.cache_direct_get("k");
        assert_eq!(tape.tenant(), super::TenantVerdict::Unchecked);
        assert!(tape.finish().is_empty(), "{:?}", tape.finish());
    }

    /// Review fix: a write does not stand in for a recorded read the run
    /// dropped.
    #[test]
    fn a_dropped_read_is_owed_even_when_the_key_was_written() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "widgets".to_owned(),
                    value: Some(base64_of(b"41")),
                },
                CacheEffect::Invalidate {
                    key: "widgets".to_owned(),
                    error: None,
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(tape.cache_invalidate("widgets"), Ok(()));
        let divergences = tape.finish();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].kind, EffectDivergenceKind::Unconsumed);
    }

    /// Codex review on #3222: a masked namespace matches when the key
    /// suffix holds a colon.
    #[test]
    fn a_masked_namespace_matches_a_key_with_a_colon_in_its_suffix() {
        assert!(super::in_namespace(
            "tenant:[FILTERED]:user:42",
            "tenant:secret"
        ));
        assert!(super::in_namespace("ns:k", "ns"));
        assert!(!super::in_namespace("other:k", "ns"));
    }

    /// Review fix: a namespace removal clears a recorded key that is masked.
    #[test]
    fn a_namespace_removal_clears_a_masked_recorded_key() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "t:[FILTERED]:x".to_owned(),
                    value: Some(base64_of(b"41")),
                },
                CacheEffect::InvalidateNamespace {
                    namespace: "t:[FILTERED]".to_owned(),
                    error: None,
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(tape.cache_get("t:acme:x"), CachedValue::Hit(b"41".to_vec()));
        assert_eq!(tape.cache_invalidate_namespace("t:acme"), Ok(()));
        assert_eq!(tape.cache_get("t:acme:x"), CachedValue::Miss);
        assert!(tape.finish().is_empty(), "{:?}", tape.finish());
    }

    /// Review fix: the part of a credential header that stood behind the
    /// placeholder is masked in the outcome, as capture masked it.
    #[test]
    fn a_credential_header_token_is_observed() {
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![HttpEffect {
                request_headers: vec![("authorization".to_owned(), "[FILTERED]".to_owned())],
                ..http("POST", "https://api.example/charge", 502)
            }],
            ..CapsuleEffects::default()
        });
        let headers = vec![(
            "authorization".to_owned(),
            "Bearer sk-live-42-long-token".to_owned(),
        )];
        let body = CapsuleBody::Absent;
        let served = tape.next_http(&OutboundRequest {
            method: "POST",
            url: "https://api.example/charge",
            headers: &headers,
            body: &body,
        });
        assert!(served.is_some());
        let masked = crate::capsule::redact::mask_echoes(
            "upstream rejected sk-live-42-long-token",
            &tape.observed_redactions(),
        );
        assert!(!masked.contains("sk-live-42"), "{masked}");
    }

    /// Codex review on #3222: a URL-encoded masked value is observed in its
    /// decoded spelling too, as capture kept both.
    #[test]
    fn a_url_encoded_masked_value_is_observed_decoded() {
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![http(
                "GET",
                "https://api.example/x?token=%5BFILTERED%5D",
                502,
            )],
            ..CapsuleEffects::default()
        });
        let body = CapsuleBody::Absent;
        let served = tape.next_http(&OutboundRequest {
            method: "GET",
            url: "https://api.example/x?token=sk%2Flive+42",
            headers: &[],
            body: &body,
        });
        assert!(served.is_some());
        let masked = crate::capsule::redact::mask_echoes(
            "upstream rejected sk/live 42",
            &tape.observed_redactions(),
        );
        assert!(!masked.contains("sk/live"), "{masked}");
    }

    /// Codex review on #3222: a cache write mismatch does not print a
    /// secret that stood behind a masked recorded key.
    #[test]
    fn a_cache_write_mismatch_does_not_print_a_masked_key() {
        let recorded = |key: &str| CapsuleEffects {
            cache: vec![CacheEffect::Invalidate {
                key: key.to_owned(),
                error: None,
            }],
            ..CapsuleEffects::default()
        };
        // The identifier matches, so its value is learned and masked.
        let tape = ReplayEffects::new(recorded("user:[FILTERED]"));
        let _ = tape.cache_invalidate_namespace("user:sk-live-42");
        assert!(tape.observed_redactions().contains(b"sk-live-42"));
        // The identifier does not match, so it is not printed.
        let tape = ReplayEffects::new(recorded("tok:[FILTERED]:x"));
        let _ = tape.cache_invalidate_namespace("other:sk-live-42");
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert!(!divergences[0].actual.contains("sk-live-42"));
        assert!(!divergences[0].detail.contains("sk-live-42"));
        // With no recorded write, the identifier is not printed.
        let tape = ReplayEffects::new(CapsuleEffects::default());
        let _ = tape.cache_invalidate("token:sk-live-42");
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert!(!divergences[0].actual.contains("sk-live-42"));
        assert!(!divergences[0].detail.contains("sk-live-42"));
    }

    /// Codex review on #3222: an invalidation of another key prints no key
    /// when the recorded key has no placeholder. The live key can hold a
    /// secret, and the scrub has not learned it.
    #[test]
    fn a_mismatched_invalidation_prints_no_live_key() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![CacheEffect::Invalidate {
                key: "user:7".to_owned(),
                error: None,
            }],
            ..CapsuleEffects::default()
        });
        let _ = tape.cache_invalidate("token:sk-live-42");
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert!(!divergences[0].actual.contains("sk-live-42"));
        assert!(!divergences[0].detail.contains("sk-live-42"));
    }

    /// Codex review on #3222: a request that changed only its message
    /// learns the masked URL value first, so the scrub masks it in the report.
    #[test]
    fn a_request_mismatch_learns_the_masked_url_value() {
        let mut recorded = http("POST", "https://api.example/users/[FILTERED]/charge", 200);
        recorded.request_body = CapsuleBody::Text("amount=10".to_owned());
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![recorded],
            ..CapsuleEffects::default()
        });
        let body = CapsuleBody::Text("amount=99".to_owned());
        let served = tape.next_http(&OutboundRequest {
            method: "POST",
            url: "https://api.example/users/sk-live-42/charge",
            headers: &[],
            body: &body,
        });
        assert!(served.is_none(), "the body changed");
        assert_eq!(tape.divergences().len(), 1);
        assert!(tape.observed_redactions().contains(b"sk-live-42"));
    }

    /// Codex review on #3222: a letter that changed only its contents
    /// learns the masked envelope first, so the scrub masks it in the report.
    #[test]
    fn a_mail_content_mismatch_learns_the_masked_envelope() {
        let tape = ReplayEffects::new(CapsuleEffects {
            mail: vec![MailEffect {
                to: vec!["[FILTERED]@shop.example".to_owned()],
                subject: "Code [FILTERED]".to_owned(),
                body: CapsuleBody::Text("hi".to_owned()),
                error: None,
                ..Default::default()
            }],
            ..CapsuleEffects::default()
        });
        let to = ["sk-live-42@shop.example".to_owned()];
        let body = CapsuleBody::Text("bye".to_owned());
        assert_eq!(
            tape.next_mail(&sent(&to, "Code 991827", &body)),
            MailVerdict::Diverged
        );
        let learned = tape.observed_redactions();
        assert!(learned.contains(b"sk-live-42"));
        assert!(learned.contains(b"991827"));
    }

    /// Codex review on #3222: a read-back of a key that only the replayed
    /// run wrote prints no key.
    #[test]
    fn a_read_back_of_an_unrecorded_write_prints_no_key() {
        let tape = ReplayEffects::new(CapsuleEffects::default());
        tape.cache_insert("token:sk-live-42", b"1", None);
        let _ = tape.cache_get("token:sk-live-42");
        let divergences = tape.divergences();
        assert!(!divergences.is_empty());
        for divergence in &divergences {
            assert!(!divergence.actual.contains("sk-live-42"), "{divergence:?}");
            assert!(!divergence.detail.contains("sk-live-42"), "{divergence:?}");
        }
    }

    /// Codex review on #3222: one recorded text can hold both placeholder
    /// spellings. Both match, and only the URL-encoded span is decoded.
    #[test]
    fn both_placeholder_spellings_match_in_one_text() {
        let recorded = "https://api.example/u/[FILTERED]/x?token=%5BFILTERED%5D";
        let actual = "https://api.example/u/alice/x?token=sk%2Flive";
        assert!(super::matches_redacted(recorded, actual));
        assert_eq!(
            super::redacted_spans(recorded, actual),
            Some(vec![("alice", false), ("sk%2Flive", true)])
        );
        assert!(!super::matches_redacted(
            recorded,
            "https://api.example/u/alice/y?token=sk%2Flive"
        ));
    }

    /// #2351 item 13: an active tape with no tenant entry fails closed.
    #[test]
    fn a_tenant_lookup_the_capsule_never_recorded_is_a_divergence() {
        let tape = ReplayEffects::new(CapsuleEffects::default());
        assert_eq!(tape.tenant(), super::TenantVerdict::Unrecorded);
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].seam, EffectSeam::Tenant);
        assert_eq!(divergences[0].kind, EffectDivergenceKind::Unrecorded);

        // A recorded failure is not a divergence: the resolver runs again on
        // the restored request.
        let tape = ReplayEffects::new(CapsuleEffects {
            tenant: Some(TenantEffect { id: None }),
            ..CapsuleEffects::default()
        });
        assert_eq!(tape.tenant(), super::TenantVerdict::RecordedFailure);
        assert!(tape.finish().is_empty());
    }

    #[test]
    fn a_recorded_miss_replays_as_a_miss_even_when_the_run_filled_the_key() {
        // The commonest cache shape there is: miss, compute, fill. Folding the
        // recorded write into the initial map would pre-seed a hit and send the
        // replayed handler down a branch the recording never took.
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "widgets".to_owned(),
                    value: None,
                },
                CacheEffect::Insert {
                    key: "widgets".to_owned(),
                    value: base64_of(b"41"),
                    ttl_secs: None,
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.cache_get("widgets"),
            CachedValue::Miss,
            "the recorded miss must replay as a miss"
        );
        tape.cache_insert("widgets", b"41", None);
        assert_eq!(
            tape.cache_get("widgets"),
            CachedValue::Hit(b"41".to_vec()),
            "the run's own fill is readable back, as it was in production"
        );
        assert!(tape.divergences().is_empty());
    }

    #[test]
    fn an_ambiguous_masked_cache_key_is_treated_as_unrecorded() {
        // `user:[FILTERED]:profile` matches every subject's entry; serving the
        // first would hand a handler that read the *wrong* subject's entry the
        // recorded value and report no divergence.
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "user:[FILTERED]:a".to_owned(),
                    value: Some(base64_of(b"1")),
                },
                CacheEffect::Get {
                    key: "user:[FILTERED]:b".to_owned(),
                    value: Some(base64_of(b"2")),
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.cache_get("user:alice:a"),
            CachedValue::Hit(b"1".to_vec())
        );
        // A key that matches both recorded patterns is refused rather than
        // guessed.
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "user:[FILTERED]".to_owned(),
                    value: Some(base64_of(b"1")),
                },
                CacheEffect::Get {
                    key: "user:[FILTERED]x".to_owned(),
                    value: Some(base64_of(b"2")),
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(tape.cache_get("user:bobx"), CachedValue::Unrecorded);
        assert_eq!(tape.divergences().len(), 1);
    }

    #[test]
    fn a_cache_read_is_served_by_key_and_an_unrecorded_key_diverges() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "user:7".to_owned(),
                    value: Some(base64_of(b"{\"a\":1}")),
                },
                CacheEffect::Get {
                    key: "user:9".to_owned(),
                    value: None,
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.cache_get("user:7"),
            CachedValue::Hit(b"{\"a\":1}".to_vec()),
            "a recorded hit serves its bytes"
        );
        assert_eq!(
            tape.cache_get("user:9"),
            CachedValue::Miss,
            "a recorded miss replays as a miss, not as a divergence"
        );
        assert_eq!(tape.cache_get("user:404"), CachedValue::Unrecorded);
        assert_eq!(tape.divergences().len(), 1, "the unrecorded key diverges");
    }

    #[test]
    fn a_mail_send_is_asserted_and_never_delivered() {
        let tape = ReplayEffects::new(CapsuleEffects {
            mail: vec![MailEffect {
                to: vec!["a@example.com".to_owned()],
                from: None,
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("thanks".to_owned()),
                error: None,
                ..Default::default()
            }],
            ..CapsuleEffects::default()
        });
        let body = CapsuleBody::Text("thanks".to_owned());
        assert_eq!(
            tape.next_mail(&sent(&["a@example.com".to_owned()], "Receipt", &body)),
            MailVerdict::Sent,
            "a recorded success is served as one, and nothing is delivered"
        );
        assert!(tape.divergences().is_empty());
        assert_eq!(
            tape.next_mail(&sent(&["b@example.com".to_owned()], "Receipt", &body)),
            MailVerdict::Diverged
        );
        assert_eq!(tape.divergences().len(), 1);
    }

    #[test]
    fn a_changed_outbound_body_diverges_rather_than_being_served_the_old_response() {
        // Same endpoint, different message. Serving the recorded response here
        // would hide the change and leave the final audit nothing to notice,
        // so an unchanged failure would grade `reproduced`.
        let mut recorded = http("POST", "https://payments.example/charge", 502);
        recorded.request_body = CapsuleBody::Text("{\"amount\":10}".to_owned());
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![recorded],
            ..CapsuleEffects::default()
        });
        let body = CapsuleBody::Text("{\"amount\":9999}".to_owned());
        assert!(
            tape.next_http(&OutboundRequest {
                method: "POST",
                url: "https://payments.example/charge",
                headers: &[],
                body: &body,
            })
            .is_none(),
            "a different amount to the same endpoint is not the recorded call"
        );
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1);
        assert!(
            divergences[0].detail.contains("its body"),
            "the report must name what changed: {}",
            divergences[0].detail
        );
        assert!(
            !divergences[0].detail.contains("9999"),
            "and must not quote the body itself: {}",
            divergences[0].detail
        );
    }

    #[test]
    fn a_changed_outbound_header_diverges_but_a_masked_one_still_matches() {
        let mut recorded = http("POST", "https://api.example/send", 200);
        recorded.request_headers = vec![
            ("authorization".to_owned(), FILTERED.to_owned()),
            ("x-idempotency-key".to_owned(), "abc".to_owned()),
        ];
        let tape = ReplayEffects::new(CapsuleEffects {
            http: vec![recorded.clone(), recorded],
            ..CapsuleEffects::default()
        });
        // The masked credential stands for whatever was really sent.
        let faithful = [
            ("authorization".to_owned(), "Bearer sk-live-42".to_owned()),
            ("x-idempotency-key".to_owned(), "abc".to_owned()),
        ];
        assert!(
            tape.next_http(&OutboundRequest {
                method: "POST",
                url: "https://api.example/send",
                headers: &faithful,
                body: &CapsuleBody::Absent,
            })
            .is_some(),
            "a masked header value must match whatever really stood there"
        );
        assert!(tape.divergences().is_empty(), "{:?}", tape.divergences());
        // An unmasked one that genuinely changed is a divergence.
        let changed = [
            ("authorization".to_owned(), "Bearer sk-live-42".to_owned()),
            ("x-idempotency-key".to_owned(), "zzz".to_owned()),
        ];
        assert!(
            tape.next_http(&OutboundRequest {
                method: "POST",
                url: "https://api.example/send",
                headers: &changed,
                body: &CapsuleBody::Absent,
            })
            .is_none()
        );
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1);
        assert!(
            divergences[0].detail.contains("x-idempotency-key")
                && !divergences[0].detail.contains("zzz"),
            "the report names the header but never its value: {}",
            divergences[0].detail
        );
    }

    #[test]
    fn a_changed_mail_body_or_sender_diverges_rather_than_replaying_clean() {
        let recorded = MailEffect {
            to: vec!["a@example.com".to_owned()],
            from: Some("billing@shop.example".to_owned()),
            subject: "Receipt".to_owned(),
            body: CapsuleBody::Text("you paid $10".to_owned()),
            error: None,
            ..Default::default()
        };
        let tape = ReplayEffects::new(CapsuleEffects {
            mail: vec![recorded.clone(), recorded],
            ..CapsuleEffects::default()
        });
        let to = ["a@example.com".to_owned()];
        let wrong_body = CapsuleBody::Text("you paid $10000".to_owned());
        assert_eq!(
            tape.next_mail(&SentMail {
                to: &to,
                from: Some("billing@shop.example"),
                subject: "Receipt",
                body: &wrong_body,
                alternate_body: &NO_BODY,
                reply_to: None,
                list_unsubscribe: None,
                extra_headers: &[],
                attachments: &[],
            }),
            MailVerdict::Diverged,
            "same envelope, different letter — the recorded success must not be served"
        );
        let right_body = CapsuleBody::Text("you paid $10".to_owned());
        assert_eq!(
            tape.next_mail(&SentMail {
                to: &to,
                from: Some("noreply@shop.example"),
                subject: "Receipt",
                body: &right_body,
                alternate_body: &NO_BODY,
                reply_to: None,
                list_unsubscribe: None,
                extra_headers: &[],
                attachments: &[],
            }),
            MailVerdict::Diverged,
            "and mail sent from a different address is a change too"
        );
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 2);
        assert!(
            divergences[0].detail.contains("its body")
                && divergences[1].detail.contains("its sender"),
            "each report names what changed: {divergences:?}"
        );
    }

    /// An unset sender is the replay *environment*, not the code: a message
    /// that names no `from` inherits `[mail] from` at send time, and a replay
    /// boots without mail configuration on purpose. Reading that as a changed
    /// sender would diverge every capsule from an app that configures one.
    #[test]
    fn a_sender_the_replay_never_configured_is_not_a_changed_sender() {
        let tape = ReplayEffects::new(CapsuleEffects {
            mail: vec![MailEffect {
                to: vec!["a@example.com".to_owned()],
                from: Some("noreply@shop.example".to_owned()),
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("thanks".to_owned()),
                error: None,
                ..Default::default()
            }],
            ..CapsuleEffects::default()
        });
        let to = ["a@example.com".to_owned()];
        let body = CapsuleBody::Text("thanks".to_owned());
        assert_eq!(
            tape.next_mail(&SentMail {
                to: &to,
                from: None,
                subject: "Receipt",
                body: &body,
                alternate_body: &NO_BODY,
                reply_to: None,
                list_unsubscribe: None,
                extra_headers: &[],
                attachments: &[],
            }),
            MailVerdict::Sent
        );
        assert!(tape.divergences().is_empty(), "{:?}", tape.divergences());
    }

    #[test]
    fn a_changed_mail_attachment_or_alternate_body_diverges() {
        // The envelope, subject, sender and preferred body all match; only the
        // parts a recipient would still plainly notice have changed.
        let recorded = MailEffect {
            to: vec!["a@example.com".to_owned()],
            from: Some("billing@shop.example".to_owned()),
            subject: "Receipt".to_owned(),
            body: CapsuleBody::Text("you paid $10".to_owned()),
            alternate_body: CapsuleBody::Text("<p>you paid $10</p>".to_owned()),
            attachments: vec![crate::capsule::schema::MailAttachmentEffect {
                filename: "invoice.pdf".to_owned(),
                content_type: "application/pdf".to_owned(),
                len: 12,
                sha256: "cd".repeat(32),
            }],
            ..Default::default()
        };
        let tape = ReplayEffects::new(CapsuleEffects {
            mail: vec![recorded.clone(), recorded.clone(), recorded],
            ..CapsuleEffects::default()
        });
        let to = ["a@example.com".to_owned()];
        let body = CapsuleBody::Text("you paid $10".to_owned());
        let same_html = CapsuleBody::Text("<p>you paid $10</p>".to_owned());
        let new_html = CapsuleBody::Text("<p>you paid $10000</p>".to_owned());
        let invoice = [crate::capsule::schema::MailAttachmentEffect {
            filename: "invoice.pdf".to_owned(),
            content_type: "application/pdf".to_owned(),
            len: 12,
            sha256: "cd".repeat(32),
        }];

        // The invoice was dropped.
        assert_eq!(
            tape.next_mail(&sent_full(&to, &body, &same_html, &[])),
            MailVerdict::Diverged,
            "an attachment the run stopped sending is a different email"
        );
        // The HTML half was rewritten while the text fallback stood still.
        assert_eq!(
            tape.next_mail(&sent_full(&to, &body, &new_html, &invoice)),
            MailVerdict::Diverged,
            "a rewritten HTML half must not ride along under an unchanged text fallback"
        );
        // Everything matching still replays.
        assert_eq!(
            tape.next_mail(&sent_full(&to, &body, &same_html, &invoice)),
            MailVerdict::Sent
        );
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 2);
        assert!(
            divergences[0].detail.contains("its attachments")
                && divergences[1].detail.contains("multipart"),
            "each report names what changed: {divergences:?}"
        );
    }

    #[test]
    fn a_recorded_mail_failure_replays_as_its_own_variant() {
        use crate::capsule::schema::MailErrorKind;
        let tape = ReplayEffects::new(CapsuleEffects {
            mail: vec![MailEffect {
                to: vec!["a@example.com".to_owned()],
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("thanks".to_owned()),
                error: Some(
                    "all recipients are on the mail suppression list; nothing was sent".to_owned(),
                ),
                error_kind: Some(MailErrorKind::AllRecipientsSuppressed),
                ..Default::default()
            }],
            ..CapsuleEffects::default()
        });
        let to = ["a@example.com".to_owned()];
        let body = CapsuleBody::Text("thanks".to_owned());
        // The kind travels with the text, so the caller can rebuild the very
        // variant a handler branches on rather than a catch-all.
        assert_eq!(
            tape.next_mail(&sent(&to, "Receipt", &body)),
            MailVerdict::Failed(
                "all recipients are on the mail suppression list; nothing was sent".to_owned(),
                Some(MailErrorKind::AllRecipientsSuppressed)
            )
        );
    }

    /// An absolute enqueue states its deadline, so it is compared on that.
    /// Comparing it by delay would either diverge on every faithful replay —
    /// the delay is `deadline - now`, and `now` moved — or, as it did, skip the
    /// comparison and let a moved deadline through.
    #[test]
    fn a_moved_absolute_deadline_is_a_divergence() {
        use crate::job::EnqueueSchedule;
        let deadline = chrono::DateTime::parse_from_rfc3339("2026-01-01T00:00:00Z")
            .expect("parses")
            .with_timezone(&chrono::Utc);
        let recorded = JobEffect {
            name: "send_receipt".to_owned(),
            payload: serde_json::json!({}),
            delay_secs: Some(3600),
            due_at: Some(deadline),
            ..Default::default()
        };
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![recorded.clone(), recorded],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                EnqueueSchedule::At(deadline + chrono::Duration::hours(1))
            ),
            EnqueueVerdict::Diverged,
            "a job rescheduled to a different instant is a behaviour change"
        );
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                EnqueueSchedule::At(deadline)
            ),
            EnqueueVerdict::Queued,
            "and the same instant still replays, whatever the clock now reads"
        );
    }

    /// A five-second entry and a permanent one are different cache mutations,
    /// however identical the bytes.
    #[test]
    fn a_changed_cache_expiry_is_a_divergence() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![CacheEffect::Insert {
                key: "user:7".to_owned(),
                value: "eyJhIjoxfQ==".to_owned(),
                ttl_secs: Some(5),
            }],
            ..CapsuleEffects::default()
        });
        tape.cache_insert("user:7", b"{\"a\":1}", Some(3600));
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].seam, EffectSeam::Cache);
    }

    /// A filename is free-form text the redaction sweep masks like any other,
    /// so a masked one must stand for whatever was really there — comparing it
    /// exactly would report a divergence for code nobody changed.
    #[test]
    fn a_masked_attachment_filename_still_matches_the_file_being_sent() {
        let attachment = |filename: &str| crate::capsule::schema::MailAttachmentEffect {
            filename: filename.to_owned(),
            content_type: "application/pdf".to_owned(),
            len: 12,
            sha256: "cd".repeat(32),
        };
        let tape = ReplayEffects::new(CapsuleEffects {
            mail: vec![MailEffect {
                to: vec!["a@example.com".to_owned()],
                from: Some("billing@shop.example".to_owned()),
                subject: "Receipt".to_owned(),
                body: CapsuleBody::Text("you paid $10".to_owned()),
                attachments: vec![attachment(FILTERED)],
                ..Default::default()
            }],
            ..CapsuleEffects::default()
        });
        let to = ["a@example.com".to_owned()];
        let body = CapsuleBody::Text("you paid $10".to_owned());
        let sent = [attachment("invoice-sk-live-42.pdf")];
        assert_eq!(
            tape.next_mail(&sent_full(&to, &body, &NO_BODY, &sent)),
            MailVerdict::Sent,
            "a masked filename stands for the real one"
        );
        assert!(tape.divergences().is_empty(), "{:?}", tape.divergences());
    }

    #[test]
    fn a_recorded_enqueue_failure_is_reproduced_as_a_failure() {
        // A handler whose 500 came from `enqueue(..).await?` must meet the
        // rejection again, not be handed the success it never got.
        let tape = ReplayEffects::new(CapsuleEffects {
            jobs: vec![JobEffect {
                name: "send_receipt".to_owned(),
                payload: serde_json::json!({}),
                delay_secs: None,
                due_at: None,
                requested_due_at: None,
                error: Some("queue is down".to_owned()),
                error_status: None,
            }],
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.next_job(
                "send_receipt",
                &serde_json::json!({}),
                crate::job::EnqueueSchedule::Immediate
            ),
            EnqueueVerdict::Failed("queue is down".to_owned(), None)
        );
        assert!(tape.divergences().is_empty());
    }

    #[test]
    fn a_cache_write_the_code_dropped_is_a_divergence() {
        // A recorded write nobody performs must not ride along under an
        // unchanged response.
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![CacheEffect::Insert {
                key: "widgets".to_owned(),
                value: base64_of(b"41"),
                ttl_secs: None,
            }],
            ..CapsuleEffects::default()
        });
        let divergences = tape.finish();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].kind, EffectDivergenceKind::Unconsumed);
    }

    #[test]
    fn a_cache_write_with_a_changed_value_is_a_divergence() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![CacheEffect::Insert {
                key: "widgets".to_owned(),
                value: base64_of(b"41"),
                ttl_secs: None,
            }],
            ..CapsuleEffects::default()
        });
        tape.cache_insert("widgets", b"42", None);
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].kind, EffectDivergenceKind::Mismatch);
    }

    #[test]
    fn a_read_of_a_key_the_recording_only_wrote_diverges() {
        // #2351 item 15: the recording wrote the key and did not read it. A
        // new read is a new cache dependency, not a read-back.
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![CacheEffect::Insert {
                key: "widgets".to_owned(),
                value: base64_of(b"41"),
                ttl_secs: None,
            }],
            ..CapsuleEffects::default()
        });
        tape.cache_insert("widgets", b"41", None);
        let _ = tape.cache_get("widgets");
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].kind, EffectDivergenceKind::Unrecorded);
        assert_eq!(divergences[0].seam, EffectSeam::Cache);
    }

    #[test]
    fn a_masked_fill_then_read_back_is_not_a_divergence() {
        // The key is masked on disk. The fill and the read-back on replay use
        // the clear key, and must find the one recorded slot.
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "user:[FILTERED]".to_owned(),
                    value: None,
                },
                CacheEffect::Insert {
                    key: "user:[FILTERED]".to_owned(),
                    value: base64_of(b"41"),
                    ttl_secs: None,
                },
                CacheEffect::Get {
                    key: "user:[FILTERED]".to_owned(),
                    value: Some(base64_of(b"41")),
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(tape.cache_get("user:alice"), CachedValue::Miss);
        tape.cache_insert("user:alice", b"41", None);
        assert_eq!(
            tape.cache_get("user:alice"),
            CachedValue::Hit(b"41".to_vec())
        );
        let divergences = tape.finish();
        assert!(divergences.is_empty(), "{divergences:?}");
    }

    #[test]
    fn a_recorded_invalidate_is_consumed_and_the_next_read_misses() {
        // #2351 item 2: an invalidate is a write. Replay consumes it and
        // never reaches a backend.
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "widgets".to_owned(),
                    value: Some(base64_of(b"41")),
                },
                CacheEffect::Invalidate {
                    key: "widgets".to_owned(),
                    error: None,
                },
                CacheEffect::Clear,
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(tape.cache_get("widgets"), CachedValue::Hit(b"41".to_vec()));
        assert_eq!(tape.cache_invalidate("widgets"), Ok(()));
        assert_eq!(tape.cache_get("widgets"), CachedValue::Miss);
        tape.cache_clear();
        let divergences = tape.finish();
        assert!(divergences.is_empty(), "{divergences:?}");
    }

    #[test]
    fn a_dropped_or_added_invalidate_is_a_divergence() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![CacheEffect::Invalidate {
                key: "widgets".to_owned(),
                error: None,
            }],
            ..CapsuleEffects::default()
        });
        let divergences = tape.finish();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].kind, EffectDivergenceKind::Unconsumed);

        let tape = ReplayEffects::new(CapsuleEffects::default());
        let _ = tape.cache_invalidate("widgets");
        tape.cache_clear();
        let _ = tape.cache_invalidate_namespace("widgets");
        tape.cache_untyped_insert("widgets");
        let divergences = tape.divergences();
        assert_eq!(divergences.len(), 4, "{divergences:?}");
        assert!(
            divergences
                .iter()
                .all(|d| d.kind == EffectDivergenceKind::Unrecorded)
        );
    }

    #[test]
    fn a_recorded_invalidate_failure_replays_as_a_failure() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Invalidate {
                    key: "widgets".to_owned(),
                    error: Some(crate::capsule::schema::CacheInvalidationError {
                        attempts: 3,
                        reason: "READONLY".to_owned(),
                    }),
                },
                CacheEffect::InvalidateNamespace {
                    namespace: "widgets".to_owned(),
                    error: Some(crate::capsule::schema::CacheInvalidationError {
                        attempts: 1,
                        reason: "no sweep".to_owned(),
                    }),
                },
            ],
            ..CapsuleEffects::default()
        });
        let err = tape
            .cache_invalidate("widgets")
            .expect_err("recorded failure");
        assert_eq!(err.attempts(), 3);
        assert_eq!(err.reason(), "READONLY");
        assert!(tape.cache_invalidate_namespace("widgets").is_err());
        assert!(tape.finish().is_empty());
    }

    /// Codex review on #3222: a removal that failed in production kept the
    /// entry, so a read after it still hits.
    #[test]
    fn a_failed_removal_keeps_the_entry() {
        let failed = Some(crate::capsule::schema::CacheInvalidationError {
            attempts: 1,
            reason: "READONLY".to_owned(),
        });
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Get {
                    key: "ns:k".to_owned(),
                    value: Some(base64_of(b"41")),
                },
                CacheEffect::Invalidate {
                    key: "ns:k".to_owned(),
                    error: failed.clone(),
                },
                CacheEffect::InvalidateNamespace {
                    namespace: "ns".to_owned(),
                    error: failed,
                },
            ],
            ..CapsuleEffects::default()
        });
        assert_eq!(tape.cache_get("ns:k"), CachedValue::Hit(b"41".to_vec()));
        assert!(tape.cache_invalidate("ns:k").is_err());
        assert!(tape.cache_invalidate_namespace("ns").is_err());
        assert_eq!(tape.cache_get("ns:k"), CachedValue::Hit(b"41".to_vec()));
        assert!(tape.finish().is_empty(), "{:?}", tape.finish());
    }

    #[test]
    fn a_matching_cache_write_is_consumed_and_readable_back() {
        let tape = ReplayEffects::new(CapsuleEffects {
            cache: vec![
                CacheEffect::Insert {
                    key: "widgets".to_owned(),
                    value: base64_of(b"41"),
                    ttl_secs: None,
                },
                CacheEffect::Get {
                    key: "widgets".to_owned(),
                    value: Some(base64_of(b"41")),
                },
            ],
            ..CapsuleEffects::default()
        });
        tape.cache_insert("widgets", b"41", None);
        assert_eq!(
            tape.cache_get("widgets"),
            CachedValue::Hit(b"41".to_vec()),
            "the run's own write is readable back, as it was in production"
        );
        assert!(tape.finish().is_empty(), "{:?}", tape.finish());
    }

    #[test]
    fn a_recorded_tenant_the_replayed_run_never_read_is_a_divergence() {
        // A router whose updated code no longer resolves a tenant would
        // otherwise leave the recording untouched and still report a clean
        // reproduction on an unchanged response.
        let tape = ReplayEffects::new(CapsuleEffects {
            tenant: Some(TenantEffect {
                id: Some("acme".to_owned()),
            }),
            ..CapsuleEffects::default()
        });
        let divergences = tape.finish();
        assert_eq!(divergences.len(), 1, "{divergences:?}");
        assert_eq!(divergences[0].seam, EffectSeam::Tenant);
    }

    #[test]
    fn the_recorded_tenant_is_served_without_consulting_live_config() {
        let tape = ReplayEffects::new(CapsuleEffects {
            tenant: Some(TenantEffect {
                id: Some("acme".to_owned()),
            }),
            ..CapsuleEffects::default()
        });
        assert_eq!(
            tape.tenant(),
            super::TenantVerdict::Resolved("acme".to_owned())
        );
    }

    fn base64_of(bytes: &[u8]) -> String {
        use base64::Engine as _;
        base64::engine::general_purpose::STANDARD.encode(bytes)
    }

    #[test]
    fn a_divergence_report_never_prints_url_userinfo() {
        let described = super::describe_http("GET", "https://svc:hunter2pw@api.example/x?k=v");
        assert_eq!(described, "GET https://[FILTERED]@api.example/x?…");
    }
}