camel_integration_test/

runner.rs

//! The scenario action runner (ADR-0069 §5, §7).
//!
//! Executes a scenario's ordered actions against a
//! [`PartnerRouter`](crate::adapters::PartnerRouter): `send`
//! dispatches through the adapter, `receive` awaits with the action's
//! deadline and applies `extract` into [`ScenarioVars`], `sleep` uses
//! tokio time, and `validate` evaluates the matcher grammar against
//! the last received message or an extracted variable, or asserts a
//! partner's recorded-request count (feature `http`).
//!
//! Failure taxonomy (ADR-0069 §7), encoded by variant and named in
//! `Display`, never by message text alone:
//!
//! - Verdict class — the scenario ran and the system under test
//!   failed it: [`ScenarioFailure::ReceiveTimeout`],
//!   [`ScenarioFailure::ValidationMismatch`],
//!   [`ScenarioFailure::VarUnresolved`].
//! - Apparatus class — the scenario never got a meaningful answer:
//!   [`ScenarioFailure::ActionTransport`],
//!   [`ScenarioFailure::PartnerStartup`],
//!   [`ScenarioFailure::ShutdownFailure`].
//!
//! Verdict-class failures map to exit 1 at the CLI; apparatus-class
//! failures map to exit 2, as do doc-validation failures before the
//! runner ever runs.
//!
//! Every await is bounded: `receive` carries the action deadline, and
//! `send` is bounded by the document's `sendDeadline`, defaulting to
//! [`SEND_DEADLINE`].

use std::collections::BTreeMap;
use std::time::{Duration, Instant};

use camel_api::{Body, Exchange, Value};
use camel_matchers::{expectation_matches, stringify};

use crate::adapters::redact_wire_path;
use crate::adapters::{
    IncomingMessage, OutgoingMessage, PartnerRouter, ReceiveError, TransportError, lanes_suffix,
};
use crate::document::{
    EndpointRef, Expectation, Provisioning, ScenarioAction, ScenarioDocument, ScenarioTarget,
    ValidateExpectation,
};

/// Partner verification for the `validate` action's `partner` target
/// (ADR-0069 §5): the filtered recorded-request count, the deadline
/// poll, and the mismatch-detail renderers.
mod partner_validate;

// Test-only re-exports: these primitives are exercised directly by
// `runner_test`, while the runner itself only calls
// `partner_validate_action`.
use partner_validate::partner_validate_action;
#[cfg(all(test, feature = "http"))]
pub(crate) use partner_validate::{
    matching_requests, partner_mismatch_detail, render_bound, render_filters,
};

/// The default bounded deadline for every `send` action (ADR-0069
/// §7: every adapter operation carries a deadline). A document-level
/// `sendDeadline` overrides it (rc-tr4w).
const SEND_DEADLINE: Duration = Duration::from_secs(30);

/// The effective send bound of a document: its declared
/// `sendDeadline`, or the thirty-second [`SEND_DEADLINE`] default.
/// Real time only (ADR-0069 §6: no virtual time).
pub(crate) fn effective_send_deadline(doc: &ScenarioDocument) -> Duration {
    doc.send_deadline.unwrap_or(SEND_DEADLINE)
}

/// Mutable run state carried across actions: scenario variables set by
/// `extract`, and the last message received per endpoint for
/// `lastReceived` validation.
#[derive(Debug, Default)]
pub struct ScenarioVars {
    /// Variables extracted from received messages, by name.
    variables: BTreeMap<String, Value>,
    /// Last message received per endpoint URI.
    last_received: BTreeMap<String, IncomingMessage>,
}

impl ScenarioVars {
    /// Empty run state.
    pub fn new() -> Self {
        Self::default()
    }

    /// The variable set by an earlier `extract`, if any.
    pub fn get(&self, name: &str) -> Option<&Value> {
        self.variables.get(name)
    }

    /// Sets a variable, overwriting any earlier value.
    pub fn set(&mut self, name: impl Into<String>, value: Value) {
        self.variables.insert(name.into(), value);
    }

    /// The last message received on the endpoint URI, if any.
    pub fn last_received(&self, endpoint: &str) -> Option<&IncomingMessage> {
        self.last_received.get(endpoint)
    }

    /// Records the last message received on an endpoint URI.
    fn remember(&mut self, endpoint: String, message: IncomingMessage) {
        self.last_received.insert(endpoint, message);
    }
}

/// Resolves `${name}` placeholders in a scenario string against `vars`.
///
/// Grammar: `$${` escapes to a literal `${`; `${name}` substitutes the
/// variable when `name` matches `[A-Za-z0-9_]+` and is immediately
/// followed by `}`. Anything else — including `${env:FOO}`, where a
/// colon follows the name — stays literal, so `${env:}` never resolves
/// in scenarios. A non-string variable substitutes its JSON
/// representation (`Value::to_string`), so a number 42 yields `42`.
/// Substituted text is not re-scanned. An unset variable fails with
/// [`ScenarioFailure::VarUnresolved`].
pub(crate) fn resolve_placeholders(
    input: &str,
    vars: &ScenarioVars,
) -> Result<String, ScenarioFailure> {
    let bytes = input.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'$' {
            // `$${` escapes to a literal `${`.
            if i + 2 < bytes.len() && bytes[i + 1] == b'$' && bytes[i + 2] == b'{' {
                out.extend_from_slice(b"${");
                i += 3;
                continue;
            }
            // `${name}` with name in [A-Za-z0-9_]+ immediately followed
            // by `}`; a colon or any other character after the name
            // keeps the whole span literal.
            if i + 1 < bytes.len() && bytes[i + 1] == b'{' {
                let name_start = i + 2;
                let mut j = name_start;
                while j < bytes.len() && (bytes[j].is_ascii_alphanumeric() || bytes[j] == b'_') {
                    j += 1;
                }
                if j > name_start && j < bytes.len() && bytes[j] == b'}' {
                    let name = &input[name_start..j];
                    match vars.get(name) {
                        Some(value) => {
                            let replacement = stringify(value);
                            out.extend_from_slice(replacement.as_bytes());
                            i = j + 1;
                            continue;
                        }
                        None => {
                            return Err(ScenarioFailure::VarUnresolved {
                                name: name.to_string(),
                            });
                        }
                    }
                }
            }
            out.push(b'$');
            i += 1;
            continue;
        }
        out.push(bytes[i]);
        i += 1;
    }
    // The output is a byte-for-byte copy of the input except for
    // substituted spans, so it stays valid UTF-8.
    Ok(String::from_utf8(out).expect("placeholder output preserves input UTF-8")) // allow-unwrap
}

/// Recursively interpolates `${name}` placeholders in a value: maps
/// and arrays are rebuilt with interpolated values, string leaves go
/// through [`resolve_placeholders`], and every other leaf is cloned
/// untouched. An unset variable propagates
/// [`ScenarioFailure::VarUnresolved`] from any depth.
pub(crate) fn interpolate_value(
    value: &Value,
    vars: &ScenarioVars,
) -> Result<Value, ScenarioFailure> {
    match value {
        Value::String(text) => Ok(Value::String(resolve_placeholders(text, vars)?)),
        Value::Array(items) => items
            .iter()
            .map(|item| interpolate_value(item, vars))
            .collect::<Result<Vec<_>, _>>()
            .map(Value::Array),
        Value::Object(map) => {
            let rebuilt = map
                .iter()
                .map(|(key, item)| Ok((key.clone(), interpolate_value(item, vars)?)))
                .collect::<Result<_, _>>()?;
            Ok(Value::Object(rebuilt))
        }
        other => Ok(other.clone()),
    }
}

/// The outcome of a scenario that ran to completion: every action
/// succeeded and every validation passed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ScenarioVerdict {
    /// All actions completed and all validations passed.
    Pass,
}

/// Why a scenario failed (ADR-0069 §7). Verdict-class variants mean
/// the system under test failed the scenario; apparatus-class
/// variants mean the scenario never got a meaningful answer. The
/// CLI maps verdict-class failures to exit 1 and apparatus-class
/// failures to exit 2; doc validation also maps to exit 2.
#[derive(Debug, Clone, PartialEq, thiserror::Error)]
#[non_exhaustive]
pub enum ScenarioFailure {
    /// Nothing reached the partner before the deadline (verdict
    /// class, `receive-timeout`).
    #[error("receive-timeout: {endpoint} delivered nothing within {deadline:?}{lanes}")]
    ReceiveTimeout {
        /// The endpoint URI that delivered nothing.
        endpoint: String,
        /// The deadline that elapsed.
        deadline: Duration,
        /// Rendered lane evidence (already redacted, ADR-0051): the
        /// `; no arrival matched; lanes recorded: [...]` suffix, or
        /// empty when the construction site saw no lanes.
        lanes: String,
    },
    /// A validation failed (verdict class, `validation-mismatch`).
    #[error("validation-mismatch: action {action}: {detail}")]
    ValidationMismatch {
        /// Index of the failing action, zero-based.
        action: usize,
        /// What was expected and what arrived.
        detail: String,
    },
    /// A referenced variable was never set (verdict class,
    /// `scenario-var-unresolved`).
    #[error("scenario-var-unresolved: {name}")]
    VarUnresolved {
        /// The variable name no `extract` ever set.
        name: String,
    },
    /// A send or receive failed at the transport before any assertion
    /// ran (apparatus class, `action-transport-failure`).
    #[error("action-transport-failure: action {action}: {source}")]
    ActionTransport {
        /// Index of the failing action, zero-based.
        action: usize,
        /// The transport failure.
        source: TransportError,
    },
    /// A partner listener bound but its handler failed to start
    /// (apparatus class, `partner-startup-failure`). Reserved in v1:
    /// no adapter separates bind from handler start, and the CLI maps
    /// bind failures to `partner-bind-failure` doc errors.
    #[error("partner-startup-failure: {message}")]
    PartnerStartup {
        /// Startup failure detail.
        message: String,
    },
    /// The partner's arrival lane dropped arrivals while the scenario
    /// was not receiving (apparatus class, `arrival-lane-overflow`):
    /// the harness lost them before the system under test could fail
    /// the scenario on substance.
    #[error("arrival-lane-overflow: {endpoint} dropped {dropped} arrivals")]
    ArrivalLaneOverflow {
        /// The endpoint URI whose lane dropped arrivals.
        endpoint: String,
        /// How many arrivals the lane dropped while full.
        dropped: usize,
    },
    /// Teardown of the boot or a partner timed out or erred after the
    /// verdict was recorded (apparatus class, `shutdown-failure`).
    #[error("shutdown-failure: {message}")]
    ShutdownFailure {
        /// Teardown failure detail.
        message: String,
    },
}

/// Fills the harness bind variables into `vars` (ADR-0069 §9): every
/// wired reference with `provisioning: harness` and a `bindVar` gets
/// its partner's bound `host:port` authority from the router, so a
/// scenario string can address the partner as
/// `http://${NAME}/path`.
///
/// Two-layer split: the scenario variable carries `host:port` only;
/// the env-tier binding that route files interpolate keeps its
/// `http://host:port` form (owned by the CLI driver, unchanged here).
/// A reference with no registered adapter or no bound authority is
/// skipped: the variable stays unset, and a later use fails with the
/// verdict-class `VarUnresolved`.
pub fn fill_bind_vars(wired: &[EndpointRef], router: &PartnerRouter, vars: &mut ScenarioVars) {
    for reference in wired {
        if reference.provisioning != Some(Provisioning::Harness) {
            continue;
        }
        let Some(bind_var) = reference.bind_var.as_deref() else {
            continue;
        };
        let Some(authority) = router
            .adapter(&reference.endpoint)
            .and_then(|adapter| adapter.bound_authority())
        else {
            continue;
        };
        vars.set(bind_var, Value::String(authority));
    }
}

/// Runs a scenario's actions in order against the router.
///
/// On success every action completed; on failure the variant names
/// the ADR-0069 §7 class. `vars` carries extraction results and
/// last-received state both into and out of the run.
pub async fn run_scenario(
    doc: &ScenarioDocument,
    router: &PartnerRouter,
    vars: &mut ScenarioVars,
) -> Result<ScenarioVerdict, ScenarioFailure> {
    // The scenario-start anchor every `elapsedAtLeast` bound measures
    // against; taken once per run, before the first action.
    let started_at = Instant::now();
    let send_deadline = effective_send_deadline(doc);
    for (index, action) in doc.scenario.iter().enumerate() {
        run_action(action, index, router, vars, started_at, send_deadline).await?;
    }
    Ok(ScenarioVerdict::Pass)
}

/// The outcome of executing a whole scenario document
/// (ADR-0069 sections 5 and 7).
///
/// [`run_scenario_document`](self::run_scenario_document) fills
/// `per_action` with one outcome per executed action and stops at the
/// first failure; `verdict` is `Some(Pass)` only when every action
/// passed. `final_failure` is the post-verdict slot: the caller that
/// owns the boot (the CLI, after `BootHandle::shutdown`) records a
/// `ShutdownFailure` there without masking the recorded verdict.
#[derive(Debug, Clone, PartialEq)]
pub struct DocumentOutcome {
    /// One outcome per executed action, in action order; actions after
    /// the first failure never ran.
    pub per_action: Vec<Result<ScenarioVerdict, ScenarioFailure>>,
    /// `Some(Pass)` when every action completed; `None` after any
    /// failure.
    pub verdict: Option<ScenarioVerdict>,
    /// Post-verdict shutdown failure, recorded by the boot-owning
    /// caller; empty when teardown is clean or never ran.
    pub final_failure: Option<ScenarioFailure>,
    /// The bound address of the document's `inbound:` listener
    /// (rc-5yon, ADR-0070), filled by the boot-owning caller from
    /// [`crate::boot_scenario::ScenarioRun::inbound_bound`] after the
    /// boot, so tests target the ephemeral listener without re-deriving
    /// it. `None` when the document declares no `inbound:` listener or
    /// the caller never filled it; the post-boot slot, as
    /// `final_failure` is the post-verdict slot.
    pub inbound_bound: Option<std::net::SocketAddr>,
}

/// Executes a scenario document's actions in order against the
/// router, one recorded outcome per action, stopping at the first
/// failure (the whole-document contract, library-level).
///
/// Partners route through `router`; a `send` addressed to a context
/// component reaches the booted system under test through the
/// context-stimulus adapter the caller registered for that endpoint
/// (see [`crate::adapters`]). The single-action
/// [`run_scenario`] loop and this loop share [`run_action`].
pub async fn run_scenario_document(
    doc: &ScenarioDocument,
    router: &PartnerRouter,
    vars: &mut ScenarioVars,
) -> DocumentOutcome {
    // The scenario-start anchor every `elapsedAtLeast` bound measures
    // against; taken once per run, before the first action.
    let started_at = Instant::now();
    let send_deadline = effective_send_deadline(doc);
    let mut per_action = Vec::with_capacity(doc.scenario.len());
    let mut failed = false;
    for (index, action) in doc.scenario.iter().enumerate() {
        if failed {
            break;
        }
        match run_action(action, index, router, vars, started_at, send_deadline).await {
            Ok(()) => per_action.push(Ok(ScenarioVerdict::Pass)),
            Err(failure) => {
                per_action.push(Err(failure));
                failed = true;
            }
        }
    }
    let verdict = if failed {
        None
    } else {
        Some(ScenarioVerdict::Pass)
    };
    DocumentOutcome {
        per_action,
        verdict,
        final_failure: None,
        inbound_bound: None,
    }
}

/// Executes one action at its scenario index. The shared primitive of
/// [`run_scenario`] and [`run_scenario_document`]; every failure
/// carries the action index.
async fn run_action(
    action: &ScenarioAction,
    index: usize,
    router: &PartnerRouter,
    vars: &mut ScenarioVars,
    started_at: Instant,
    send_deadline: Duration,
) -> Result<(), ScenarioFailure> {
    match action {
        ScenarioAction::Send {
            to,
            body,
            headers,
            method,
            expect_reply,
        } => {
            send_action(
                index,
                to,
                body.as_ref(),
                headers.as_ref(),
                method,
                expect_reply.as_ref(),
                router,
                vars,
                send_deadline,
            )
            .await?;
        }
        ScenarioAction::Receive {
            from,
            deadline,
            extract,
        } => {
            receive_action(index, from, *deadline, extract.as_ref(), router, vars).await?;
        }
        ScenarioAction::Sleep { duration } => {
            tokio::time::sleep(*duration).await;
        }
        ScenarioAction::Validate { .. } => {
            validate_action(action, index, started_at, router, vars).await?;
        }
    }
    Ok(())
}

/// Dispatches a `send` action, bounded by the document's effective
/// send deadline ([`effective_send_deadline`]: the declared
/// `sendDeadline`, or the thirty-second default).
///
/// The endpoint reference, the body's string leaves, and the header
/// values are the complete interpolation surface: each resolves its
/// `${name}` placeholders against `vars` before dispatch, and an
/// unresolved variable fails with the verdict-class `VarUnresolved`.
/// The dial target comes from the router's address math: a
/// harness-declared `:0` reference (or a dynamic reference resolving
/// to a partner authority) dials the partner's bound address with the
/// interpolated path preserved; anything else dials the interpolated
/// URI literally.
///
/// A declared `expectReply` (rc-qvz6, `direct:` sends only — the
/// grammar rejected every other scheme at load) asserts the
/// synchronous reply the adapter returned: a non-matching reply is a
/// verdict-class [`ScenarioFailure::ValidationMismatch`] naming the
/// rendered expectation and the actual body, and a missing reply is
/// an apparatus-class [`ScenarioFailure::ActionTransport`] — the
/// scenario never got an answer to assert against.
// The action's flat decomposition (index, endpoint, body, headers,
// method, reply expectation, router, vars) plus the document send
// bound threaded from run_action (rc-tr4w).
#[allow(clippy::too_many_arguments)]
async fn send_action(
    index: usize,
    to: &EndpointRef,
    body: Option<&Value>,
    headers: Option<&BTreeMap<String, Value>>,
    method: &str,
    expect_reply: Option<&Expectation>,
    router: &PartnerRouter,
    vars: &ScenarioVars,
    send_deadline: Duration,
) -> Result<(), ScenarioFailure> {
    let declared = to.endpoint.as_str();
    let interpolated = resolve_placeholders(declared, vars)?;
    let body = body
        .map(|value| interpolate_value(value, vars))
        .transpose()?;
    let headers = headers
        .map(|map| -> Result<BTreeMap<String, Value>, ScenarioFailure> {
            map.iter()
                .map(|(name, value)| Ok((name.clone(), interpolate_value(value, vars)?)))
                .collect()
        })
        .transpose()?;
    let msg = OutgoingMessage {
        body: body.unwrap_or(Value::Null),
        headers: headers.unwrap_or_default(),
        method: method.to_string(),
    };
    let bounded =
        tokio::time::timeout(send_deadline, router.send(declared, &interpolated, msg)).await;
    let sent = bounded.map_err(|_| ScenarioFailure::ActionTransport {
        action: index,
        source: TransportError::Deadline {
            after: send_deadline,
        },
    })?;
    let reply = sent.map_err(|source| {
        // Render-site defense: the lane key is the declared endpoint
        // URI, and a third-party adapter may hand the overflow over
        // RAW; the runner holds the secret set, and redaction is
        // idempotent on already-masked output (ADR-0051).
        let source = match source {
            TransportError::LaneFifoOverflow { lane_key, bound } => {
                TransportError::LaneFifoOverflow {
                    lane_key: redact_wire_path(&lane_key, &router.secret_query_keys()),
                    bound,
                }
            }
            other => other,
        };
        ScenarioFailure::ActionTransport {
            action: index,
            source,
        }
    })?;
    if let Some(expectation) = expect_reply {
        let Some(reply) = reply else {
            // Fail closed: the grammar promised a direct reply, but
            // the adapter produced none — an apparatus defect, never
            // a silently-skipped assertion.
            return Err(ScenarioFailure::ActionTransport {
                action: index,
                source: TransportError::Other {
                    message: "direct send produced no reply".to_string(),
                },
            });
        };
        let value = reply_body_value(&reply);
        if !expectation_matches(expectation, &value) {
            return Err(ScenarioFailure::ValidationMismatch {
                action: index,
                detail: format!(
                    "direct reply on {}: expected {}, got {}",
                    to.endpoint,
                    render_expectation(expectation),
                    stringify(&value)
                ),
            });
        }
    }
    Ok(())
}

/// Converts a synchronous `direct:` reply exchange's body into the
/// matcher value an `expectReply` assertion reads (rc-qvz6): the
/// reply message is the exchange's output when the route produced
/// one, the (route-mutated — `set_body` writes it) input otherwise.
/// Feature-free by design: the partner-body extractors stay
/// `http`-gated; this path never touches the wire. Crate-visible for
/// the runner's unit tests, like the interpolation primitives.
pub(crate) fn reply_body_value(exchange: &Exchange) -> Value {
    let message = exchange.output.as_ref().unwrap_or(&exchange.input);
    match &message.body {
        Body::Json(value) => value.clone(),
        Body::Text(text) => reply_bytes_value(text.as_bytes()),
        Body::Xml(text) => reply_bytes_value(text.as_bytes()),
        Body::Bytes(bytes) => reply_bytes_value(bytes),
        // Empty and consumed-stream bodies carry no reply bytes, and
        // foreign `#[non_exhaustive]` body kinds (none today) expose
        // none either; the value reads as the empty string.
        _ => Value::String(String::new()),
    }
}

/// Parses reply bytes as JSON, falling back to a lossy-UTF-8 string
/// when they are not JSON text: a text body holding JSON is observed
/// as the structured value the matcher verbs expect, and any other
/// text stays textual.
fn reply_bytes_value(bytes: &[u8]) -> Value {
    serde_json::from_slice(bytes)
        .unwrap_or_else(|_| Value::String(String::from_utf8_lossy(bytes).into_owned()))
}

/// Renders an expectation for an `expectReply` mismatch detail: the
/// verb and its payload in the document grammar's own terms.
fn render_expectation(expectation: &Expectation) -> String {
    match expectation {
        Expectation::Equals(expected) => format!("equals {expected}"),
        Expectation::Regex(pattern) => format!("matches regex `{pattern}`"),
        Expectation::Contains(needle) => format!("contains `{needle}`"),
        Expectation::StartsWith(prefix) => format!("startsWith `{prefix}`"),
        Expectation::EndsWith(suffix) => format!("endsWith `{suffix}`"),
        Expectation::Exists => "exists".to_string(),
        Expectation::JsonSubset(pattern) => format!("is a superset of {pattern}"),
        // Foreign `#[non_exhaustive]` variants (none today): no verb
        // renders, but the matcher already failed closed.
        _ => "the expected value".to_string(),
    }
}

/// Awaits a `receive` action until the deadline, records the message,
/// and applies `extract` into `vars`.
async fn receive_action(
    index: usize,
    from: &EndpointRef,
    deadline: Duration,
    extract: Option<&BTreeMap<String, String>>,
    router: &PartnerRouter,
    vars: &mut ScenarioVars,
) -> Result<(), ScenarioFailure> {
    // The lane is read under the two-key contract: the declared
    // string names the registered lane when it can, and the
    // interpolated URI resolves a dynamic reference's lane by
    // authority (`lane_key_for`).
    let declared = from.endpoint.as_str();
    let interpolated = resolve_placeholders(declared, vars)?;
    let message = router
        .receive(declared, &interpolated, deadline)
        .await
        .map_err(|source| {
            // Render-site defense: a third-party adapter may hand
            // over RAW endpoint and lane evidence; the runner holds
            // the secret set, and redaction is idempotent on
            // already-masked output (ADR-0051).
            let keys = router.secret_query_keys();
            match source {
                ReceiveError::Timeout(timeout) => ScenarioFailure::ReceiveTimeout {
                    endpoint: redact_wire_path(&timeout.endpoint, &keys),
                    deadline,
                    lanes: lanes_suffix(
                        &timeout
                            .lanes_recorded
                            .iter()
                            .map(|lane| redact_wire_path(lane, &keys))
                            .collect::<Vec<_>>(),
                    ),
                },
                ReceiveError::Overflow(overflow) => ScenarioFailure::ArrivalLaneOverflow {
                    endpoint: redact_wire_path(&overflow.endpoint, &keys),
                    dropped: overflow.dropped,
                },
                ReceiveError::Transport(source) => ScenarioFailure::ActionTransport {
                    action: index,
                    source,
                },
            }
        })?;
    if let Some(extract) = extract {
        for (name, selector) in extract {
            let value = select_from(&message, selector).ok_or_else(|| {
                ScenarioFailure::ValidationMismatch {
                    action: index,
                    detail: format!(
                        "extract of `{selector}` into variable `{name}` resolved to nothing"
                    ),
                }
            })?;
            vars.set(name.clone(), value);
        }
    }
    vars.remember(from.endpoint.clone(), message);
    Ok(())
}

/// Evaluates a `validate` action (ADR-0069 §5).
///
/// The `partner` target asserts the exact filtered count of the
/// requests the harness partner recorded, read as the router's
/// snapshot — one immediate read without a deadline, a polled one
/// with it ([`partner_validate_action`]). Every other target applies
/// the message grammar against `vars`; the deadline is partner-only
/// (the grammar rejected it on these targets at parse time, so the
/// message arm ignores it). An `elapsedAtLeast` bound on a
/// `lastReceived` target checks the message's wire arrival against
/// the scenario-start anchor before the grammar runs; the grammar
/// rejected it on every other target at parse time. Mismatch details
/// name the validation subject — the variable's name, the receiving
/// endpoint, or the partner URI — so a corrupted-header regression is
/// diagnosable from the failure text.
async fn validate_action(
    action: &ScenarioAction,
    index: usize,
    started_at: Instant,
    router: &PartnerRouter,
    vars: &ScenarioVars,
) -> Result<(), ScenarioFailure> {
    // run_action dispatches only the Validate variant here; the
    // fallback mirrors the impossible pairing arms below.
    let ScenarioAction::Validate {
        target,
        expectation,
        deadline,
        elapsed_at_least,
    } = action
    else {
        return Err(unpaired_validate(index));
    };
    match (target, expectation) {
        // The parser pairs a `partner` target with the partner count
        // grammar; this arm reads the router's snapshot and owns the
        // deadline.
        (ScenarioTarget::Partner(endpoint), ValidateExpectation::Partner(expected)) => {
            partner_validate_action(index, &endpoint.endpoint, expected, *deadline, router).await
        }
        (_, ValidateExpectation::Message(expectation)) => {
            let (value, subject) = match target {
                ScenarioTarget::LastReceived(endpoint) => {
                    // The declared endpoint may carry query bytes: the
                    // subject renders redacted like every diagnostic
                    // that quotes a wire path (ADR-0051).
                    let redacted =
                        redact_wire_path(&endpoint.endpoint, &router.secret_query_keys());
                    let message = vars.last_received(&endpoint.endpoint).ok_or_else(|| {
                        ScenarioFailure::ValidationMismatch {
                            action: index,
                            detail: format!(
                                "no message has been received on {redacted} to validate"
                            ),
                        }
                    })?;
                    // The elapsed bound anchors to the message's wire
                    // arrival, never the consumption time: a message
                    // consumed late can still have arrived early (the
                    // wire is the proof, ADR-0069 §5).
                    if let Some(bound) = elapsed_at_least {
                        let actual = message
                            .arrival
                            .checked_duration_since(started_at)
                            .unwrap_or_default();
                        if actual < *bound {
                            return Err(ScenarioFailure::ValidationMismatch {
                                action: index,
                                detail: format!(
                                    "{redacted}: arrived {} after the scenario started; `elapsedAtLeast` requires {}",
                                    humantime::format_duration(actual),
                                    humantime::format_duration(*bound)
                                ),
                            });
                        }
                    }
                    (
                        message.body.clone(),
                        format!("body last received on {redacted}"),
                    )
                }
                ScenarioTarget::Variable(name) => (
                    vars.get(name)
                        .cloned()
                        .ok_or_else(|| ScenarioFailure::VarUnresolved { name: name.clone() })?,
                    format!("variable `{name}`"),
                ),
                // Taken by the arm above: the grammar never pairs a
                // `partner` target with the message expectation.
                ScenarioTarget::Partner(_) => return Err(unpaired_validate(index)),
            };
            // The per-form booleans delegate to the shared core
            // (`camel_matchers::expectation_matches`); the detail
            // strings stay here, where subject rendering and
            // redaction live.
            match expectation {
                Expectation::Equals(expected) => check(
                    index,
                    expectation_matches(expectation, &value),
                    format!("{subject}: expected {expected}, got {value}"),
                ),
                // The parser pre-verifies regex patterns at load time,
                // so the invalid-regex arm is unreachable through the
                // harness; it stays for the byte-identical verdicts,
                // short-circuiting before the core delegation (core's
                // Regex arm returns false on compile-fail).
                Expectation::Regex(pattern) => {
                    if let Err(error) = regex::Regex::new(pattern) {
                        return Err(ScenarioFailure::ValidationMismatch {
                            action: index,
                            detail: format!("invalid regex `{pattern}`: {error}"),
                        });
                    }
                    check(
                        index,
                        expectation_matches(expectation, &value),
                        format!("{subject}: `{pattern}` did not match {value}"),
                    )
                }
                Expectation::Contains(needle) => check(
                    index,
                    expectation_matches(expectation, &value),
                    format!("{subject}: did not contain `{needle}`: {value}"),
                ),
                Expectation::StartsWith(prefix) => check(
                    index,
                    expectation_matches(expectation, &value),
                    format!("{subject}: did not start with `{prefix}`: {value}"),
                ),
                Expectation::EndsWith(suffix) => check(
                    index,
                    expectation_matches(expectation, &value),
                    format!("{subject}: did not end with `{suffix}`: {value}"),
                ),
                Expectation::Exists => check(
                    index,
                    expectation_matches(expectation, &value),
                    format!("{subject}: expected a value, got null"),
                ),
                Expectation::JsonSubset(pattern) => check(
                    index,
                    expectation_matches(expectation, &value),
                    format!("{subject}: not a superset of {pattern}: {value}"),
                ),
                // Foreign `#[non_exhaustive]` variants (none today):
                // the harness has no matcher for them, so they fail
                // closed.
                _ => Err(ScenarioFailure::ValidationMismatch {
                    action: index,
                    detail: "validate expectation kind is not supported by the message grammar"
                        .to_string(),
                }),
            }
        }
        // The parser never pairs a partner expectation with a
        // non-partner target.
        _ => Err(unpaired_validate(index)),
    }
}

/// The failure for a target/expectation pairing the grammar never
/// produces: the parser pairs `partner` targets with the partner
/// count grammar and every other target with the message grammar, so
/// only a caller bypassing the parser reaches these arms.
fn unpaired_validate(index: usize) -> ScenarioFailure {
    ScenarioFailure::ValidationMismatch {
        action: index,
        detail: "validate target kind does not pair with the expectation kind".to_string(),
    }
}

/// Turns a validation predicate into a [`ScenarioFailure`] on `false`.
fn check(index: usize, passed: bool, detail: String) -> Result<(), ScenarioFailure> {
    if passed {
        Ok(())
    } else {
        Err(ScenarioFailure::ValidationMismatch {
            action: index,
            detail,
        })
    }
}

/// Reads a value out of a received message by dotted selector.
///
/// Grammar: the first segment selects `body`, `headers`, `status`,
/// `method`, or `path`; the rest is a literal header name
/// (`headers.X-Id`, dots allowed in the name) or a dotted object path
/// under the body (`body.user.id`). A bare `body` or `headers` selects
/// the whole part.
///
/// Header lookup is ASCII-case-insensitive: adapters differ in header
/// casing (hyper lowercases wire names; the fake preserves author
/// casing), and the same selector must behave identically per adapter.
/// Wire recording stays lowercase.
fn select_from(message: &IncomingMessage, selector: &str) -> Option<Value> {
    let (head, rest) = match selector.split_once('.') {
        Some((head, rest)) => (head, Some(rest)),
        None => (selector, None),
    };
    match head {
        "body" => match rest {
            None => Some(message.body.clone()),
            Some(path) => walk_path(&message.body, path),
        },
        "headers" => match rest {
            None => Some(Value::Object(
                message
                    .headers
                    .iter()
                    .map(|(name, value)| (name.clone(), value.clone()))
                    .collect(),
            )),
            Some(name) => lookup_header(&message.headers, name),
        },
        // The transport-scalar heads carry no sub-path: `status.why` is
        // not part of the grammar and resolves to nothing.
        "status" if rest.is_none() => Some(
            message
                .status
                .map_or(Value::Null, |code| Value::Number(code.into())),
        ),
        "method" if rest.is_none() => {
            Some(message.method.clone().map_or(Value::Null, Value::String))
        }
        "path" if rest.is_none() => Some(message.path.clone().map_or(Value::Null, Value::String)),
        _ => None,
    }
}

/// Case-insensitive header lookup: the first header whose name matches
/// the selector ASCII-case-insensitively wins; header maps are
/// case-unique per adapter, so the fold is deterministic.
fn lookup_header(headers: &BTreeMap<String, Value>, name: &str) -> Option<Value> {
    headers
        .iter()
        .find(|(key, _)| key.eq_ignore_ascii_case(name))
        .map(|(_, value)| value.clone())
}

/// Walks a dotted object path under a body value; arrays and scalars
/// resolve to nothing.
fn walk_path(value: &Value, path: &str) -> Option<Value> {
    let mut current = value;
    for key in path.split('.') {
        current = current.as_object()?.get(key)?;
    }
    Some(current.clone())
}