rig_core/test_utils/

streaming_conformance.rs

//! Wire-sequence conformance scenarios for provider streaming pipelines.
//!
//! The streaming sibling of `rig-agent`'s `model_conformance`: each scenario
//! drives raw wire bytes (SSE or NDJSON) through a provider's *complete*
//! streaming path — bytes → decode → the writer's parts →
//! [`CompletionStream`](crate::streaming::CompletionStream) → its response
//! — and asserts the streaming contract: an error is the stream's last item,
//! a reply the provider did not end has no response, and every part the
//! stream emits reaches the response exactly once. Scenarios state the contract; a per-provider
//! [`ProviderWireFixture`] supplies the frames, since each wire format spells
//! the same event differently.
//!
//! Every sequence family here pins a shipped bug from the #2257 review rounds
//! (`rig-2257-code-review-findings-*.md`); the per-scenario comments cite the
//! specific finding.
//!
//! Suites are expanded per wire family by
//! [`streaming_conformance_suite!`](crate::streaming_conformance_suite).
//! Scenarios a wire cannot spell return an explicit
//! [`ScenarioOutcome::Skipped`] that the macro cross-checks against the
//! suite's declared [`SuiteCapabilities`] — a skip is always visible and can
//! never masquerade as a pass, so the executed count is exactly the declared
//! grid minus the named skips (#2258 review, F8 corpus honesty).

use bytes::Bytes;
use futures::StreamExt;
use futures::future::BoxFuture;

use crate::completion::{CompletionRequest, CompletionResponse};
use crate::error::ProviderError;
use crate::{
    completion::FinishReason,
    error::ErrorReport,
    http_client,
    message::AssistantContent,
    streaming::{Item, StreamEvent},
};

/// Typed failure from a wire-conformance scenario.
#[derive(Debug, thiserror::Error)]
pub enum ConformanceError {
    /// Opening the stream failed before any wire frame was consumed.
    #[error(transparent)]
    Completion(#[from] ProviderError),
    /// The pipeline violated the streaming contract table.
    #[error("{scenario} conformance failed for {provider}: {details}")]
    Contract {
        /// Stable scenario name.
        scenario: &'static str,
        /// Provider driver under test.
        provider: &'static str,
        /// Actionable observation explaining the failure.
        details: String,
    },
}

impl ConformanceError {
    fn contract(
        scenario: &'static str,
        provider: &'static str,
        details: impl Into<String>,
    ) -> Self {
        Self::Contract {
            scenario,
            provider,
            details: details.into(),
        }
    }
}

/// Outcome of a passing wire-conformance scenario.
#[derive(Debug)]
pub struct ScenarioReport {
    /// Stable scenario name.
    pub name: &'static str,
    /// Provider driver the scenario ran against.
    pub provider: &'static str,
    /// Human-readable observations, one per verified sub-case.
    pub observations: Vec<String>,
}

/// What a capability-gated scenario did: ran its assertions, or skipped
/// because the wire family cannot spell the sequence shape.
///
/// A skip is an explicit, named outcome — never a silent pass. The
/// [`streaming_conformance_suite!`](crate::streaming_conformance_suite)
/// macro cross-checks it against the suite's declared capability flags via
/// [`check_gated_outcome`], so a fixture cannot vacuously pass a scenario its
/// capabilities claim to cover (#2258 review, F8 corpus-honesty batch).
#[derive(Debug)]
pub enum ScenarioOutcome {
    /// The scenario ran and its assertions held.
    Ran(ScenarioReport),
    /// The fixture lacks the sequence shape; nothing was asserted.
    Skipped {
        /// Stable scenario name.
        name: &'static str,
        /// Provider driver under test.
        provider: &'static str,
        /// Why the wire family cannot spell the shape.
        reason: &'static str,
    },
}

/// Streaming-relevant capability flags for one wire family's conformance
/// suite: which optional sequence shapes the wire can spell.
///
/// Each flag mirrors an `Option` field on [`ProviderWireFixture`], and the
/// only constructor is [`ProviderWireFixture::capabilities`] — suites never
/// hand-write flags, so a flag structurally cannot drift from the wire
/// fixture that backs it (it *is* the fixture's populated-field set).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct SuiteCapabilities {
    /// The wire streams tool-call arguments incrementally
    /// (`partial_tool_call_frames`).
    pub partial_tool_args: bool,
    /// The wire has a genuine terminal that can omit usage metrics
    /// (`zero_usage_terminal_frames`).
    pub zero_usage_terminal: bool,
    /// The wire has a data-less terminal signal (`bare_terminal_frames`).
    pub bare_terminal: bool,
    /// A frame-level decode failure can be spelled (`malformed_frame`).
    pub malformed_frame: bool,
    /// An unknown event type can be spelled (`unknown_event_frame`).
    pub unknown_event_frame: bool,
    /// A known event with a schema-defective payload can be spelled
    /// (`defective_known_frame`).
    pub defective_known_frame: bool,
    /// The wire has a delta-less choice prelude shape
    /// (`delta_less_prelude_frame`).
    pub delta_less_prelude: bool,
    /// The wire has a refusal channel (`refusal`).
    pub refusal: bool,
    /// The wire mints a constant per-stream reasoning identity, so
    /// interleaving output is its only reasoning boundary
    /// (`interleaved_reasoning`).
    pub interleaved_reasoning: bool,
}

impl SuiteCapabilities {
    /// Build a capability set from manifest names. An unknown name is an
    /// error so a typo in a suite's `manifest:` list fails loudly rather
    /// than silently asserting an empty flag; the macro-expanded test
    /// asserts on it.
    pub fn from_names(names: &[&str]) -> Result<Self, String> {
        let mut caps = Self::default();
        for name in names {
            match *name {
                "partial_tool_args" => caps.partial_tool_args = true,
                "zero_usage_terminal" => caps.zero_usage_terminal = true,
                "bare_terminal" => caps.bare_terminal = true,
                "malformed_frame" => caps.malformed_frame = true,
                "unknown_event_frame" => caps.unknown_event_frame = true,
                "defective_known_frame" => caps.defective_known_frame = true,
                "delta_less_prelude" => caps.delta_less_prelude = true,
                "refusal" => caps.refusal = true,
                "interleaved_reasoning" => caps.interleaved_reasoning = true,
                other => {
                    return Err(format!(
                        "unknown capability name in suite manifest: {other}"
                    ));
                }
            }
        }
        Ok(caps)
    }
}

/// The canonical fixture-driven scenario set every wire-family suite must
/// expand — one named test each, compared against the macro's emitted list by
/// its `suite_is_complete` test (langchain's anti-tamper precedent).
pub const CANONICAL_SCENARIOS: &[&str] = &[
    "truncation_preserves_content_without_terminal",
    "transport_error_after_tool_call_yields_err_then_end",
    "malformed_frame_ends_the_reply",
    "unknown_event_is_skipped",
    "defective_known_event_ends_the_reply",
    "delta_less_choice_prelude_is_a_noop",
    "refusal_frames_deliver_text_without_error",
    "bare_terminal_after_only_unparseable_frames_fabricates_nothing",
    "usage_variants_are_reported_or_absent",
    "interleaved_constant_id_reasoning_preserves_order",
];

/// Every streaming wire family in the workspace. The workspace registry test
/// (`all_wire_families_have_conformance_suites`) fails CI when any family
/// lacks a [`streaming_conformance_suite!`](crate::streaming_conformance_suite)
/// invocation naming it.
pub const WIRE_FAMILIES: &[&str] = &[
    "openai_chat",
    "openai_responses",
    "openai_responses_websocket",
    "chatgpt",
    "anthropic",
    "gemini_rest",
    "gemini_interactions",
    "gemini_grpc",
    "xai",
    "copilot",
    "bedrock",
    "candle",
];

/// The sanctioned reason for an expected-failure scenario, from `xfail`
/// entries of the form `"scenario_name: reason (finding reference)"`.
pub fn xfail_reason<'a>(xfail: &[&'a str], scenario: &str) -> Option<&'a str> {
    xfail.iter().find_map(|entry| {
        let (name, reason) = entry.split_once(':')?;
        (name.trim() == scenario).then(|| reason.trim())
    })
}

/// `xfail` entries that do not name a canonical scenario or carry no reason.
pub fn invalid_xfail_entries(xfail: &[&str]) -> Vec<String> {
    xfail
        .iter()
        .filter(|entry| match entry.split_once(':') {
            Some((name, reason)) => {
                !CANONICAL_SCENARIOS.contains(&name.trim()) || reason.trim().is_empty()
            }
            None => true,
        })
        .map(std::string::ToString::to_string)
        .collect()
}

/// Enforce a capability-gated scenario's outcome against the suite's declared
/// capability flag and its `xfail` list.
///
/// A `Skipped` outcome passes only when the capability is disclaimed; a `Ran`
/// outcome passes only when it is declared — so a vacuous pass (fixture lacks
/// the shape but the suite claims to cover it) is impossible, and the skip is
/// visible in the test output.
pub fn check_gated_outcome(
    scenario: &'static str,
    capability: bool,
    xfail: &[&str],
    outcome: Result<ScenarioOutcome, ConformanceError>,
) -> Result<(), String> {
    match (xfail_reason(xfail, scenario), outcome) {
        (Some(reason), Err(error)) => {
            eprintln!("xfail {scenario}: {reason} ({error})");
            Ok(())
        }
        (Some(reason), Ok(_)) => Err(format!(
            "{scenario} passed but is listed as xfail ({reason}); remove the xfail entry"
        )),
        (None, Err(error)) => Err(format!("{scenario} failed: {error}")),
        (None, Ok(ScenarioOutcome::Ran(_))) => {
            if capability {
                Ok(())
            } else {
                Err(format!(
                    "{scenario} ran but the suite disclaims the capability; set the flag to true"
                ))
            }
        }
        (None, Ok(ScenarioOutcome::Skipped { reason, .. })) => {
            if capability {
                Err(format!(
                    "{scenario} skipped ({reason}) but the suite declares the capability; \
                     a declared capability's scenario must run"
                ))
            } else {
                eprintln!("skipped {scenario}: {reason}");
                Ok(())
            }
        }
    }
}

/// Enforce an always-runnable scenario's result against the `xfail` list.
pub fn check_ungated_outcome(
    scenario: &'static str,
    xfail: &[&str],
    result: Result<ScenarioReport, ConformanceError>,
) -> Result<(), String> {
    match (xfail_reason(xfail, scenario), result) {
        (Some(reason), Err(error)) => {
            eprintln!("xfail {scenario}: {reason} ({error})");
            Ok(())
        }
        (Some(reason), Ok(_)) => Err(format!(
            "{scenario} passed but is listed as xfail ({reason}); remove the xfail entry"
        )),
        (None, Err(error)) => Err(format!("{scenario} failed: {error}")),
        (None, Ok(_)) => Ok(()),
    }
}

/// One scripted wire input frame.
///
/// Byte-transport wires (SSE, NDJSON, websocket) script raw bytes fed through
/// the provider's HTTP layer; typed-event wires (bedrock, candle,
/// gemini-grpc) script already-typed SDK events fed to the adapter directly —
/// events-first, no mock transport — which the typed driver downcasts back.
#[derive(Clone)]
pub enum WireInput {
    /// A raw wire byte frame.
    Bytes(Bytes),
    /// An already-typed SDK event for a typed-event wire.
    Event(std::sync::Arc<dyn std::any::Any + Send + Sync>),
}

impl WireInput {
    /// The frame's raw bytes, when it is a byte frame.
    pub fn as_bytes(&self) -> Option<&Bytes> {
        match self {
            Self::Bytes(bytes) => Some(bytes),
            Self::Event(_) => None,
        }
    }

    /// The frame's typed event, when it is an event frame of type `T`.
    pub fn downcast_event<T: 'static>(&self) -> Option<&T> {
        match self {
            Self::Bytes(_) => None,
            Self::Event(event) => event.downcast_ref(),
        }
    }
}

impl From<Bytes> for WireInput {
    fn from(bytes: Bytes) -> Self {
        Self::Bytes(bytes)
    }
}

impl std::fmt::Debug for WireInput {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Bytes(bytes) => formatter.debug_tuple("Bytes").field(bytes).finish(),
            Self::Event(_) => formatter.write_str("Event(..)"),
        }
    }
}

/// Build a typed-event fixture frame.
pub fn event_frame<T: Send + Sync + 'static>(event: T) -> WireInput {
    WireInput::Event(std::sync::Arc::new(event))
}

/// The wire frames a driver feeds into the provider's pipeline. An `Err`
/// chunk models a mid-stream transport failure.
pub type WireChunks = Vec<http_client::Result<WireInput>>;

/// Build the chunk list for an all-delivered frame sequence.
pub fn ok_chunks(frames: impl IntoIterator<Item = impl Into<WireInput>>) -> WireChunks {
    frames.into_iter().map(|frame| Ok(frame.into())).collect()
}

/// A scripted mid-stream transport failure chunk: the connection dropped,
/// so there is no reply and no status.
pub fn transport_error_chunk() -> http_client::Result<WireInput> {
    Err(http_client::Error::instance(std::io::Error::new(
        std::io::ErrorKind::ConnectionReset,
        "connection reset",
    )))
}

/// Executable stream-lifecycle validator (#2258 C1).
///
/// The invariants every normalized stream must satisfy, stated once and run
/// over every recorded cassette and corpus fixture that drains through
/// [`fixtures::drain`] — the langchain `assert_valid_event_stream` move:
/// prose contracts scattered across N adapters become one executable
/// artifact. Panics with the violated law.
///
/// Laws (universal — they hold for truncated and errored streams too):
///
/// 1. **Terminal error.** An error is the stream's last item: nothing
///    follows it.
/// 2. **Text conservation.** The aggregated text is exactly the
///    concatenation of the yielded text deltas: accumulated delta content
///    equals the payload the aggregate delivers.
/// 3. **Completed-call conservation.** Every completed tool call yielded on
///    the stream appears in the aggregated choice exactly once, and vice
///    versa (counts match; aggregation neither drops nor duplicates).
/// 4. **Sequence.** The events read back through
///    [`Transcript::parse_prefix`](crate::streaming::Transcript::parse_prefix):
///    each part starts once, in position, grows only while open, and ends
///    once.
/// 5. **Reasoning provenance.** The response contains a reasoning part only
///    if the stream ended one.
pub fn assert_valid_event_stream(
    items: &[Result<Item<StreamEvent>, ErrorReport>],
    choice: &[AssistantContent],
) {
    use crate::message::AssistantContent;

    // Law 1: an error is the last item.
    if let Some(error_index) = items.iter().position(Result::is_err) {
        assert_eq!(
            error_index + 1,
            items.len(),
            "law 1 (terminal error): an item followed the stream's error"
        );
    }
    let events: Vec<&StreamEvent> = items
        .iter()
        .filter_map(|item| match item {
            Ok(Item::Event(event)) => Some(event),
            _ => None,
        })
        .collect();

    // Law 2: text conservation.
    let streamed_text: String = events
        .iter()
        .filter_map(|event| match event {
            StreamEvent::Text { text, .. } => Some(text.as_str()),
            _ => None,
        })
        .collect();
    let aggregated_text: String = choice
        .iter()
        .filter_map(|content| match content {
            AssistantContent::Text(text) => Some(text.text.as_str()),
            _ => None,
        })
        .collect();
    assert_eq!(
        aggregated_text, streamed_text,
        "law 2 (text conservation): aggregated text differs from the streamed fragments"
    );

    // Law 3: completed-call conservation.
    let yielded_calls = events
        .iter()
        .filter(|event| {
            matches!(
                event,
                StreamEvent::End {
                    content: AssistantContent::ToolCall(_),
                    ..
                }
            )
        })
        .count();
    let aggregated_calls = choice
        .iter()
        .filter(|content| matches!(content, AssistantContent::ToolCall(_)))
        .count();
    assert_eq!(
        aggregated_calls, yielded_calls,
        "law 3 (completed-call conservation): {yielded_calls} calls yielded, \
         {aggregated_calls} aggregated"
    );

    // Law 4: the sequence reads back.
    let serialized = serde_json::to_value(
        items
            .iter()
            .filter_map(|item| item.as_ref().ok())
            .collect::<Vec<_>>(),
    )
    .unwrap_or_default();
    let read_back = crate::streaming::Transcript::parse_prefix(serialized);
    assert!(
        read_back.is_ok(),
        "law 4 (sequence): the stream's events do not read back: {read_back:?}"
    );

    // Law 5: reasoning provenance.
    let yielded_reasoning = events.iter().any(|event| {
        matches!(
            event,
            StreamEvent::End {
                content: AssistantContent::Reasoning(_),
                ..
            }
        )
    });
    let aggregated_reasoning = choice
        .iter()
        .any(|content| matches!(content, AssistantContent::Reasoning(_)));
    assert!(
        yielded_reasoning || !aggregated_reasoning,
        "law 5 (reasoning provenance): aggregated reasoning with no reasoning yielded"
    );
}

/// Everything the consumer observed from one full pipeline run: the yielded
/// items in order, plus the response they folded into.
#[derive(Debug)]
pub struct DrainedStream {
    /// Every item the stream yielded, in order.
    pub items: Vec<Result<Item<StreamEvent>, ErrorReport>>,
    /// The parts that ended, in position: the response's choice, or what
    /// arrived before the stream stopped.
    pub choice: Vec<AssistantContent>,
    /// The response, absent on truncation or a terminal error.
    pub response: Option<CompletionResponse>,
}

impl DrainedStream {
    fn events(&self) -> impl Iterator<Item = &StreamEvent> {
        self.items.iter().filter_map(|item| match item {
            Ok(Item::Event(event)) => Some(event),
            _ => None,
        })
    }

    /// Text fragments yielded to the consumer, in order.
    pub fn texts(&self) -> Vec<&str> {
        self.events()
            .filter_map(|event| match event {
                StreamEvent::Text { text, .. } => Some(text.as_str()),
                _ => None,
            })
            .collect()
    }

    /// Names of the tool calls that ended on the stream, in order.
    pub fn tool_call_names(&self) -> Vec<&str> {
        self.events()
            .filter_map(|event| match event {
                StreamEvent::End {
                    content: AssistantContent::ToolCall(tool_call),
                    ..
                } => Some(tool_call.function.name.as_str()),
                _ => None,
            })
            .collect()
    }

    /// Raw payloads of the `Unknown` items the stream yielded, in order.
    pub fn unknown_values(&self) -> Vec<&serde_json::Value> {
        self.items
            .iter()
            .filter_map(|item| match item {
                Ok(Item::Unknown(value)) => Some(value.value()),
                _ => None,
            })
            .collect()
    }

    /// Number of `Err` items the stream yielded.
    pub fn error_count(&self) -> usize {
        self.items.iter().filter(|item| item.is_err()).count()
    }

    /// Whether the provider ended the reply: it has a response.
    fn has_terminal(&self) -> bool {
        self.response.is_some()
    }

    /// Whether the stream reached the provider's end with no `Err` item —
    /// the clean-completion precondition the content-shape scenarios share.
    fn completed_cleanly(&self) -> bool {
        self.error_count() == 0 && self.response.is_some()
    }

    /// Index of the first `Err` item, if any.
    fn first_error_index(&self) -> Option<usize> {
        self.items.iter().position(std::result::Result::is_err)
    }

    /// Text parts in the aggregated choice, in order.
    pub fn choice_texts(&self) -> Vec<&str> {
        self.choice
            .iter()
            .filter_map(|content| match content {
                AssistantContent::Text(text) => Some(text.text.as_str()),
                _ => None,
            })
            .collect()
    }

    /// Reasoning parts in the aggregated choice, in order.
    pub fn choice_reasoning(&self) -> Vec<&crate::message::Reasoning> {
        self.choice
            .iter()
            .filter_map(|content| match content {
                AssistantContent::Reasoning(reasoning) => Some(reasoning),
                _ => None,
            })
            .collect()
    }
}

type DriveFn = Box<
    dyn Fn(WireChunks) -> BoxFuture<'static, Result<DrainedStream, ProviderError>> + Send + Sync,
>;

/// One provider's full streaming pipeline over scripted wire chunks.
///
/// The closure builds a fresh provider client over a scripted HTTP double
/// (`SequencedStreamingHttpClient`), opens `Model::stream`, drains
/// it, and returns everything the consumer observed.
pub struct WireDriver {
    /// Stable descriptor name of the provider under test.
    pub provider: &'static str,
    drive: DriveFn,
}

impl WireDriver {
    /// Wrap a provider pipeline closure.
    pub fn new(
        provider: &'static str,
        drive: impl Fn(WireChunks) -> BoxFuture<'static, Result<DrainedStream, ProviderError>>
        + Send
        + Sync
        + 'static,
    ) -> Self {
        Self {
            provider,
            drive: Box::new(drive),
        }
    }

    /// Run the provider's full pipeline over `chunks` and drain it.
    pub async fn drive(&self, chunks: WireChunks) -> Result<DrainedStream, ProviderError> {
        (self.drive)(chunks).await
    }
}

/// Refusal frames and the text the pipeline must deliver for them.
pub struct RefusalFixture {
    /// Frames carrying the refusal content.
    pub frames: Vec<WireInput>,
    /// Text the consumer must observe.
    pub expected_text: &'static str,
}

/// The interleaving-boundary shape for a wire whose reasoning identity is
/// a constant per-stream minted key: reasoning, an interleaved tool call,
/// then more reasoning, which must aggregate as three ordered parts —
/// never one merged item that misorders history on replay.
pub struct InterleavedReasoningFixture {
    /// Reasoning → tool call → reasoning frames, terminal included.
    pub frames: Vec<WireInput>,
    /// The reasoning content streamed before the boundary.
    pub first_reasoning: &'static str,
    /// The interleaved call's tool name.
    pub tool_name: &'static str,
    /// The reasoning content streamed after the boundary.
    pub second_reasoning: &'static str,
}

type BufferedDriveFn = Box<
    dyn Fn(String) -> BoxFuture<'static, Result<Vec<AssistantContent>, ProviderError>>
        + Send
        + Sync,
>;

/// A buffered-body pipeline (the ChatGPT backend shape): the full SSE body is
/// re-parsed after the fact and merged with the terminal response body.
pub struct BufferedBodyDriver {
    /// Stable descriptor name of the provider under test.
    pub provider: &'static str,
    drive: BufferedDriveFn,
}

impl BufferedBodyDriver {
    /// Wrap a buffered pipeline closure.
    pub fn new(
        provider: &'static str,
        drive: impl Fn(String) -> BoxFuture<'static, Result<Vec<AssistantContent>, ProviderError>>
        + Send
        + Sync
        + 'static,
    ) -> Self {
        Self {
            provider,
            drive: Box::new(drive),
        }
    }

    /// Run the buffered pipeline over a complete SSE body.
    pub async fn drive(&self, body: String) -> Result<Vec<AssistantContent>, ProviderError> {
        (self.drive)(body).await
    }
}

/// Per-provider wire frames for the shared scenario set.
///
/// `Option` fields cover sequence shapes a wire family cannot spell (e.g.
/// an NDJSON wire has no event types, so no "unknown event type" frame).
pub struct ProviderWireFixture {
    /// The provider's full pipeline.
    pub driver: WireDriver,
    /// Frames that deliver exactly the text deltas in `expected_texts`.
    pub text_frames: Vec<WireInput>,
    /// The text deltas `text_frames` delivers, in order.
    pub expected_texts: Vec<&'static str>,
    /// Frames that fully deliver one tool call (including any completion
    /// signal the wire needs, but no stream terminal).
    pub tool_call_frames: Vec<WireInput>,
    /// Name of the tool call `tool_call_frames` delivers.
    pub expected_tool_name: &'static str,
    /// Frames that leave a tool call mid-arguments, where the wire streams
    /// arguments incrementally.
    pub partial_tool_call_frames: Option<Vec<WireInput>>,
    /// The provider's genuine stream terminal, carrying usage.
    pub terminal_frames: Vec<WireInput>,
    /// Total tokens `terminal_frames` reports.
    pub expected_usage_total: u64,
    /// Finish reason `terminal_frames` reports.
    pub expected_finish_reason: Option<FinishReason>,
    /// A genuine terminal that reports no usage metrics at all.
    pub zero_usage_terminal_frames: Option<Vec<WireInput>>,
    /// A terminal signal that carries no data of its own (e.g. a bare
    /// `[DONE]`), for wires that have one.
    pub bare_terminal_frames: Option<Vec<WireInput>>,
    /// A frame that fails the wire decode entirely. `None` only for
    /// typed-event wires, whose SDK surfaces decode failures as transport
    /// errors — a frame-level corrupt input cannot be spelled there.
    pub malformed_frame: Option<WireInput>,
    /// An event type this client does not know, for typed-event wires.
    pub unknown_event_frame: Option<WireInput>,
    /// A known event whose payload is schema-defective.
    pub defective_known_frame: Option<WireInput>,
    /// A delta-less choice prelude (the Azure `prompt_filter_results` shape).
    pub delta_less_prelude_frame: Option<WireInput>,
    /// Refusal content frames, where the wire has a refusal channel.
    pub refusal: Option<RefusalFixture>,
    /// The interleaving-boundary shape, where the wire's reasoning identity
    /// is a constant per-stream minted key (its adapter synthesizes the
    /// reasoning ends other output implies).
    pub interleaved_reasoning: Option<InterleavedReasoningFixture>,
}

impl ProviderWireFixture {
    /// The capability set this fixture's populated optional fields spell —
    /// the descriptor the suite macro gates scenarios on.
    ///
    /// Deriving flags here (instead of hand-writing them per suite
    /// invocation) makes flag/fixture drift structurally impossible: a shape
    /// the fixture supplies is a declared capability, a shape it lacks is a
    /// visible named skip, and there is nothing else to keep in sync.
    pub fn capabilities(&self) -> SuiteCapabilities {
        SuiteCapabilities {
            partial_tool_args: self.partial_tool_call_frames.is_some(),
            zero_usage_terminal: self.zero_usage_terminal_frames.is_some(),
            bare_terminal: self.bare_terminal_frames.is_some(),
            malformed_frame: self.malformed_frame.is_some(),
            unknown_event_frame: self.unknown_event_frame.is_some(),
            defective_known_frame: self.defective_known_frame.is_some(),
            delta_less_prelude: self.delta_less_prelude_frame.is_some(),
            refusal: self.refusal.is_some(),
            interleaved_reasoning: self.interleaved_reasoning.is_some(),
        }
    }
}

fn concat_frames(parts: &[&[WireInput]]) -> Vec<WireInput> {
    parts
        .iter()
        .flat_map(|frames| frames.iter().cloned())
        .collect()
}

/// A scenario's identity and running notes, bound once.
///
/// The scenario name and the provider under test are facts of the *scenario*,
/// not of each check inside it: every contract violation in a body reports
/// the same pair. `Checks` holds them, so a check states only what must hold
/// and a report states only what was observed.
struct Checks {
    name: &'static str,
    provider: &'static str,
    observations: Vec<String>,
}

impl Checks {
    fn new(name: &'static str, provider: &'static str) -> Self {
        Self {
            name,
            provider,
            observations: Vec::new(),
        }
    }

    /// This scenario's contract violation, for the `ok_or_else` shape.
    fn fail(&self, details: impl Into<String>) -> ConformanceError {
        ConformanceError::contract(self.name, self.provider, details)
    }

    /// Require a contract to hold. `details` names the violation and is
    /// built only when the check fails.
    fn require<D: Into<String>>(
        &self,
        held: bool,
        details: impl FnOnce() -> D,
    ) -> Result<(), ConformanceError> {
        if held {
            return Ok(());
        }
        Err(self.fail(details()))
    }

    /// Record a verified sub-case.
    fn note(&mut self, observation: impl Into<String>) {
        self.observations.push(observation.into());
    }

    /// The wire family cannot spell this sequence shape: an explicit, named
    /// skip that [`check_gated_outcome`] cross-checks against the suite's
    /// declared capabilities.
    fn skip(&self, reason: &'static str) -> ScenarioOutcome {
        ScenarioOutcome::Skipped {
            name: self.name,
            provider: self.provider,
            reason,
        }
    }

    fn report(self) -> ScenarioReport {
        ScenarioReport {
            name: self.name,
            provider: self.provider,
            observations: self.observations,
        }
    }

    fn ran(self) -> ScenarioOutcome {
        ScenarioOutcome::Ran(self.report())
    }
}

/// Truncation at every position — EOF before content, mid-text, mid-tool-args,
/// after a tool call's frames — must preserve the content that ended and
/// never produce a terminal record.
///
/// Pins the truncation family from round one (`rig-2257-code-review-findings-ec9f2625.md`):
/// EOF without the provider's end event must not synthesize a successful
/// usage-less terminal.
pub async fn truncation_preserves_content_without_terminal(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new(
        "truncation_preserves_content_without_terminal",
        fixture.driver.provider,
    );

    // EOF before any content.
    let drained = fixture.driver.drive(Vec::new()).await?;
    checks.require(
        !drained.has_terminal(),
        || "an empty stream must not synthesize a terminal record",
    )?;
    checks.note("EOF before content: no terminal");

    // EOF after text deltas.
    let drained = fixture
        .driver
        .drive(ok_chunks(fixture.text_frames.clone()))
        .await?;
    checks.require(drained.texts() == fixture.expected_texts, || {
        format!(
            "text delivered before truncation must be preserved: expected {:?}, observed {:?}",
            fixture.expected_texts,
            drained.texts()
        )
    })?;
    checks.require(
        !drained.has_terminal(),
        || "EOF after text deltas must not synthesize a terminal record",
    )?;
    checks.note("EOF mid-text: content preserved, no terminal");

    // EOF mid-tool-arguments, where the wire streams arguments.
    if let Some(partial) = &fixture.partial_tool_call_frames {
        let drained = fixture.driver.drive(ok_chunks(partial.clone())).await?;
        checks.require(
            !drained.has_terminal(),
            || "EOF mid-tool-arguments must not synthesize a terminal record",
        )?;
        checks.note("EOF mid-tool-args: no terminal");
    }

    // EOF after a tool call's frames, before the provider's end. A call is
    // visible once it closes: on a wire that closes calls at the end, the
    // cut call never surfaces.
    let drained = fixture
        .driver
        .drive(ok_chunks(fixture.tool_call_frames.clone()))
        .await?;
    checks.require(
        drained
            .tool_call_names()
            .iter()
            .all(|name| *name == fixture.expected_tool_name),
        || {
            format!(
                "only the delivered call may surface: observed {:?}",
                drained.tool_call_names()
            )
        },
    )?;
    checks.require(
        !drained.has_terminal(),
        || "EOF after a tool call must not synthesize a terminal record",
    )?;
    checks.note("EOF after a tool call: no terminal");

    Ok(checks.report())
}

/// A transport failure after a tool call's frames ends the reply: the `Err`
/// is the last item, and there is no terminal record. A call the wire closed
/// before the failure precedes it; one still open never surfaces.
pub async fn transport_error_after_tool_call_yields_err_then_end(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new(
        "transport_error_after_tool_call_yields_err_then_end",
        fixture.driver.provider,
    );

    let mut chunks = ok_chunks(fixture.tool_call_frames.clone());
    chunks.push(transport_error_chunk());
    let drained = fixture.driver.drive(chunks).await?;

    checks.require(
        drained
            .tool_call_names()
            .iter()
            .all(|name| *name == fixture.expected_tool_name),
        || {
            format!(
                "only the delivered call may precede the transport error: observed {:?}",
                drained.tool_call_names()
            )
        },
    )?;
    let error_index = drained
        .first_error_index()
        .ok_or_else(|| checks.fail("the transport failure must reach the consumer"))?;
    checks.require(
        error_index + 1 == drained.items.len(),
        || "nothing may follow the terminal transport error",
    )?;
    checks.require(
        !drained.has_terminal(),
        || "a transport failure must not be papered over with a terminal record",
    )?;

    checks.note("Err, then end; no terminal");
    Ok(checks.report())
}

/// A malformed frame between valid content and the provider's end ends the
/// reply: it surfaces as the stream's last item, content before it is kept,
/// and there is no response.
pub async fn malformed_frame_ends_the_reply(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioOutcome, ConformanceError> {
    let mut checks = Checks::new("malformed_frame_ends_the_reply", fixture.driver.provider);
    let Some(malformed) = &fixture.malformed_frame else {
        return Ok(checks.skip("wire family cannot spell a frame-level decode failure"));
    };

    let frames = concat_frames(&[
        &fixture.text_frames,
        std::slice::from_ref(malformed),
        &fixture.terminal_frames,
    ]);
    let drained = fixture.driver.drive(ok_chunks(frames)).await?;

    checks.require(drained.error_count() == 1, || {
        format!(
            "the malformed frame must surface as exactly one Err item, observed {}",
            drained.error_count()
        )
    })?;
    checks.require(
        drained.first_error_index() == Some(drained.items.len() - 1),
        || "the malformed frame's error must be the stream's last item",
    )?;
    checks.require(
        drained.texts() == fixture.expected_texts,
        || "content before the malformed frame must be preserved",
    )?;
    checks.require(
        !drained.has_terminal(),
        || "a reply cut by a corrupt frame has no response",
    )?;

    checks.note("Err surfaced last; content kept; no response");
    Ok(checks.ran())
}

/// An event type the client does not know must be skipped without an error,
/// and the stream must still complete.
///
/// Pins the unknown-event forward-compatibility policy (round three,
/// `rig-2257-code-review-findings-8a2f41c7.md`).
pub async fn unknown_event_is_skipped(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioOutcome, ConformanceError> {
    let mut checks = Checks::new("unknown_event_is_skipped", fixture.driver.provider);
    let Some(unknown) = &fixture.unknown_event_frame else {
        return Ok(checks.skip("wire family cannot spell an unknown event type"));
    };

    let frames = concat_frames(&[
        &fixture.text_frames,
        std::slice::from_ref(unknown),
        &fixture.terminal_frames,
    ]);
    let drained = fixture.driver.drive(ok_chunks(frames)).await?;

    checks.require(
        drained.error_count() == 0,
        || "an unknown event type must be skipped, not surfaced as an error",
    )?;
    checks.require(
        drained.texts() == fixture.expected_texts && drained.response.is_some(),
        || "the stream must deliver its content and complete around the skipped event",
    )?;
    // The frame is skipped semantically but observable verbatim on the raw
    // passthrough channel (openai-agents' raw-event precedent, #2258 item 5).
    checks.require(drained.unknown_values().len() == 1, || {
        format!(
            "exactly one Unknown passthrough item must surface for the unknown frame, \
             observed {}",
            drained.unknown_values().len()
        )
    })?;

    // Control run without the unknown frame: the aggregated assistant choice
    // must be byte-identical — the passthrough item is never folded in.
    let control_frames = concat_frames(&[&fixture.text_frames, &fixture.terminal_frames]);
    let control = fixture.driver.drive(ok_chunks(control_frames)).await?;
    checks.require(
        drained.choice == control.choice,
        || "the unknown frame must not perturb the aggregated assistant choice",
    )?;

    checks.note(
        "unknown event skipped semantically, surfaced on the raw channel, \
         choice unchanged, stream completed",
    );
    Ok(checks.ran())
}

/// A *known* event whose payload is schema-defective ends the reply: it
/// surfaces as the stream's last item, and there is no response.
///
/// Pins the round-5 known-type strictness policy (a defective known event is
/// never demoted to an unknown one).
pub async fn defective_known_event_ends_the_reply(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioOutcome, ConformanceError> {
    let mut checks = Checks::new(
        "defective_known_event_ends_the_reply",
        fixture.driver.provider,
    );
    let Some(defective) = &fixture.defective_known_frame else {
        return Ok(
            checks.skip("wire family cannot spell a known event with a schema-defective payload")
        );
    };

    let frames = concat_frames(&[
        &fixture.text_frames,
        std::slice::from_ref(defective),
        &fixture.terminal_frames,
    ]);
    let drained = fixture.driver.drive(ok_chunks(frames)).await?;

    checks.require(drained.error_count() == 1, || {
        format!(
            "a known event with a schema defect must surface exactly one Err item, observed {}",
            drained.error_count()
        )
    })?;
    checks.require(
        !drained.has_terminal(),
        || "a reply cut by a defective event has no response",
    )?;

    checks.note("defective known event ended the reply with its Err");
    Ok(checks.ran())
}

/// A delta-less choice (the Azure `prompt_filter_results` prelude) must be a
/// no-op — no error, no content, and the rest of the stream unaffected.
///
/// Pins the Azure prelude no-op from round two
/// (`rig-2257-code-review-findings-b91d03aa.md`).
pub async fn delta_less_choice_prelude_is_a_noop(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioOutcome, ConformanceError> {
    let mut checks = Checks::new(
        "delta_less_choice_prelude_is_a_noop",
        fixture.driver.provider,
    );
    let Some(prelude) = &fixture.delta_less_prelude_frame else {
        return Ok(checks.skip("wire family has no delta-less prelude shape"));
    };

    let frames = concat_frames(&[
        std::slice::from_ref(prelude),
        &fixture.text_frames,
        &fixture.terminal_frames,
    ]);
    let drained = fixture.driver.drive(ok_chunks(frames)).await?;

    checks.require(
        drained.error_count() == 0,
        || "the delta-less prelude must not surface an error",
    )?;
    checks.require(
        drained.texts() == fixture.expected_texts && drained.response.is_some(),
        || "the prelude must not perturb content delivery or the terminal",
    )?;

    checks.note("delta-less prelude ignored; stream unaffected");
    Ok(checks.ran())
}

/// Refusal frames must deliver their text to the consumer without an error.
///
/// Pins the refusal-delta handling from round three
/// (`rig-2257-code-review-findings-8a2f41c7.md`).
pub async fn refusal_frames_deliver_text_without_error(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioOutcome, ConformanceError> {
    let mut checks = Checks::new(
        "refusal_frames_deliver_text_without_error",
        fixture.driver.provider,
    );
    let Some(refusal) = &fixture.refusal else {
        return Ok(checks.skip("wire family has no refusal channel"));
    };

    let frames = concat_frames(&[&refusal.frames, &fixture.terminal_frames]);
    let drained = fixture.driver.drive(ok_chunks(frames)).await?;

    checks.require(
        drained.error_count() == 0,
        || "refusal content must not surface as an error",
    )?;
    let delivered = drained.texts().concat();
    checks.require(delivered == refusal.expected_text, || {
        format!(
            "refusal text must be delivered: expected {:?}, observed {delivered:?}",
            refusal.expected_text
        )
    })?;
    checks.require(
        drained.response.is_some(),
        || "a refused turn still ends with the provider's genuine terminal",
    )?;

    checks.note("refusal text delivered without error");
    Ok(checks.ran())
}

/// On the buffered-body pipeline (the ChatGPT backend), a terminal whose body
/// carries text never seen as a delta must merge that text into the choice
/// exactly once, and a body restating streamed deltas must not duplicate them.
///
/// Pins the terminal-body/delta per-kind merge from round five
/// (`rig-2257-code-review-findings-5c73639c.md`) and the empty-delta merge
/// direction verified in round six (`rig-2257-code-review-findings-34ee8ba5.md`
/// P3-2).
pub async fn terminal_body_content_merges_per_kind(
    driver: &BufferedBodyDriver,
    cases: Vec<(&'static str, String)>,
    expected_text: &str,
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new("terminal_body_content_merges_per_kind", driver.provider);

    for (label, body) in cases {
        let choice = driver.drive(body).await?;
        let choice_text: String = choice
            .iter()
            .filter_map(|content| match content {
                AssistantContent::Text(text) => Some(text.text.as_str()),
                _ => None,
            })
            .collect();
        let occurrences = choice_text.matches(expected_text).count();
        checks.require(occurrences == 1, || {
            format!(
                "{label}: terminal-body text must appear exactly once in the choice, observed {occurrences} in {choice_text:?}"
            )
        })?;
        checks.note(format!("{label}: text merged exactly once"));
    }

    Ok(checks.report())
}

/// A bare terminal signal after only-unparseable frames must not fabricate a
/// successful terminal record: the parse errors were already surfaced, and a
/// default-usage terminal would dress the failure up as success.
///
/// Pins the bare-`[DONE]` guard from round six
/// (`rig-2257-code-review-findings-5c73639c.md`, carried into `34ee8ba5`).
pub async fn bare_terminal_after_only_unparseable_frames_fabricates_nothing(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioOutcome, ConformanceError> {
    let mut checks = Checks::new(
        "bare_terminal_after_only_unparseable_frames_fabricates_nothing",
        fixture.driver.provider,
    );
    let Some(bare_terminal) = &fixture.bare_terminal_frames else {
        return Ok(checks.skip("wire family has no data-less terminal signal"));
    };
    let Some(malformed) = &fixture.malformed_frame else {
        return Ok(checks.skip("wire family cannot spell a frame-level decode failure"));
    };

    let frames = concat_frames(&[std::slice::from_ref(malformed), bare_terminal]);
    let drained = fixture.driver.drive(ok_chunks(frames)).await?;

    checks.require(
        drained.error_count() != 0,
        || "the unparseable frame must surface as an Err item",
    )?;
    checks.require(
        !drained.has_terminal(),
        || "a bare terminal with no decoded frame must not fabricate a terminal record",
    )?;

    checks.note("no fabricated terminal after only-unparseable frames");
    Ok(checks.ran())
}

/// The genuine terminal must report the provider's usage; a terminal without
/// usage metrics must complete with no counter reported rather than being
/// suppressed or invented.
///
/// Pins the absent-usage contract on [`CompletionResponse::usage`](crate::completion::CompletionResponse::usage)
/// (round one, `rig-2257-code-review-findings-ec9f2625.md`).
pub async fn usage_variants_are_reported_or_absent(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new(
        "usage_variants_are_reported_or_absent",
        fixture.driver.provider,
    );

    let frames = concat_frames(&[&fixture.text_frames, &fixture.terminal_frames]);
    let drained = fixture.driver.drive(ok_chunks(frames)).await?;
    let response = drained
        .response
        .as_ref()
        .ok_or_else(|| checks.fail("the genuine terminal must produce a record"))?;
    checks.require(
        response.usage.total_tokens == Some(fixture.expected_usage_total),
        || {
            format!(
                "terminal usage must be preserved: expected total {}, observed {:?}",
                fixture.expected_usage_total, response.usage.total_tokens
            )
        },
    )?;
    checks.require(
        response.finish_reason() == fixture.expected_finish_reason,
        || {
            format!(
                "terminal finish reason must be normalized: expected {:?}, observed {:?}",
                fixture.expected_finish_reason,
                response.finish_reason()
            )
        },
    )?;
    checks.note(format!(
        "usage total {} and finish reason {:?} preserved",
        fixture.expected_usage_total, fixture.expected_finish_reason
    ));

    if let Some(zero_usage) = &fixture.zero_usage_terminal_frames {
        let frames = concat_frames(&[&fixture.text_frames, zero_usage]);
        let drained = fixture.driver.drive(ok_chunks(frames)).await?;
        let response = drained.response.as_ref().ok_or_else(|| {
            checks.fail("a usage-less genuine terminal must still complete the stream")
        })?;
        checks.require(!response.usage.is_reported(), || {
            format!(
                "missing usage metrics must leave every counter unreported, not invented: {:?}",
                response.usage
            )
        })?;
        checks.note("usage-less terminal completed with no counter reported");
    }

    Ok(checks.report())
}

/// Reasoning-summary deltas followed by the item's full `output_item.done`
/// block must aggregate to the summary exactly once — the full block
/// supersedes its own deltas, never duplicates them.
///
/// Pins the open P1 in `rig-2257-code-review-findings-34ee8ba5.md` ("OpenAI
/// Responses reasoning-summary streams duplicate reasoning content"):
/// `reasoning_summary_text.delta` drops `item_id`, so the strict same-item
/// table appends the full block beside the delta-built item.
pub async fn reasoning_summary_deltas_are_superseded_without_duplication(
    driver: &WireDriver,
    frames: Vec<WireInput>,
    summary_text: &str,
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new(
        "reasoning_summary_deltas_are_superseded_without_duplication",
        driver.provider,
    );

    let drained = driver.drive(ok_chunks(frames)).await?;
    checks.require(
        drained.completed_cleanly(),
        || "the reasoning stream must complete without errors",
    )?;
    let reasoning = drained.choice_reasoning();
    let occurrences: usize = reasoning
        .iter()
        .map(|item| item.text.matches(summary_text).count())
        .sum();
    checks.require(occurrences == 1, || {
        format!(
            "the summary must appear exactly once in the aggregated choice, observed {occurrences} across {reasoning:?}"
        )
    })?;
    checks.require(reasoning.len() == 1, || {
        format!(
            "deltas and their full block must collapse to one reasoning item, observed {}",
            reasoning.len()
        )
    })?;

    checks.note("summary aggregated exactly once");
    Ok(checks.report())
}

/// A reasoning item whose `output_item.done` carries several parts under one
/// item id (summary parts, text, encrypted) must keep every part, in order:
/// the item is one block whose provider item holds them all.
pub async fn multi_part_same_id_reasoning_keeps_every_part(
    driver: &WireDriver,
    frames: Vec<WireInput>,
    expected_parts: &[&str],
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new(
        "multi_part_same_id_reasoning_keeps_every_part",
        driver.provider,
    );

    let drained = driver.drive(ok_chunks(frames)).await?;
    checks.require(
        drained.completed_cleanly(),
        || "the reasoning stream must complete without errors",
    )?;
    let items: Vec<String> = drained
        .choice_reasoning()
        .iter()
        .filter_map(|item| item.native.as_ref())
        .map(|native| native.item.to_string())
        .collect();
    let in_order = |item: &String| {
        let mut rest = item.as_str();
        expected_parts.iter().all(|part| match rest.find(part) {
            Some(at) => {
                rest = &rest[at + part.len()..];
                true
            }
            None => false,
        })
    };
    let observed = items;
    checks.require(observed.len() == 1 && observed.iter().all(in_order), || {
        format!(
            "every same-id reasoning part must survive in order: expected {expected_parts:?}, observed {observed:?}"
        )
    })?;

    checks.note(format!(
        "all {} reasoning parts survived",
        expected_parts.len()
    ));
    Ok(checks.report())
}

/// Reasoning deltas interleaved with a tool call, then the item's completed
/// block, must aggregate to exactly one reasoning item carrying the block's
/// content.
///
/// Pins the interleaved-reasoning replacement contract on
/// completed reasoning block (round six,
/// `rig-2257-code-review-findings-34ee8ba5.md`, "Verified sound" section).
pub async fn interleaved_reasoning_aggregates_to_one_item(
    driver: &WireDriver,
    frames: Vec<WireInput>,
    expected_text: &str,
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new(
        "interleaved_reasoning_aggregates_to_one_item",
        driver.provider,
    );

    let drained = driver.drive(ok_chunks(frames)).await?;
    checks.require(
        drained.completed_cleanly(),
        || "the interleaved stream must complete without errors",
    )?;
    let reasoning = drained.choice_reasoning();
    checks.require(reasoning.len() == 1, || {
        format!(
            "interleaved deltas and their completed block must collapse to one reasoning item, observed {}",
            reasoning.len()
        )
    })?;
    let carries_text = reasoning.iter().any(|item| item.text == expected_text);
    checks.require(carries_text, || {
        format!("the reasoning item must carry the completed block's text {expected_text:?}")
    })?;

    checks.note("exactly one reasoning item with the completed content");
    Ok(checks.report())
}

/// On a constant-id wire (a boundary-minted per-stream reasoning id), other
/// output closes the open reasoning item: thought → tool call → thought must
/// aggregate as `[Reasoning(first), ToolCall, Reasoning(second)]` — two items
/// in arrival order, never one merged item that misorders history on replay.
///
/// Pins the F1b ordering dimension of the #2258 review (main's
/// "other output closes the reasoning item" boundary, lost when identity
/// became the per-stream constant).
pub async fn interleaved_constant_id_reasoning_preserves_order(
    fixture: &ProviderWireFixture,
) -> Result<ScenarioOutcome, ConformanceError> {
    let mut checks = Checks::new(
        "interleaved_constant_id_reasoning_preserves_order",
        fixture.driver.provider,
    );
    let Some(interleaved) = &fixture.interleaved_reasoning else {
        return Ok(checks.skip("wire fixture supplies no interleaved reasoning frames"));
    };

    let drained = fixture
        .driver
        .drive(ok_chunks(interleaved.frames.clone()))
        .await?;
    checks.require(
        drained.completed_cleanly(),
        || "the interleaved stream must complete without errors",
    )?;
    assert_reasoning_tool_reasoning(
        &checks,
        &drained,
        interleaved.first_reasoning,
        interleaved.tool_name,
        interleaved.second_reasoning,
    )?;

    checks.note("boundary kept: reasoning, tool call, reasoning in order");
    Ok(checks.ran())
}

/// On a constant-id wire whose completed reasoning block arrives as a signed
/// full restatement (gemini `thoughtSignature`), a full block *after*
/// interleaved output must not replace-and-discard the thought accumulated
/// before the boundary: the choice keeps `[Reasoning(first), ToolCall,
/// Reasoning(second, signed)]`.
///
/// Pins the F1b erasure dimension of the #2258 review, on top of the F1
/// adapter fix (the signed chunk restates only post-boundary fragments).
pub async fn interleaved_signed_full_reasoning_does_not_erase_prior_thought(
    driver: &WireDriver,
    frames: Vec<WireInput>,
    first: &str,
    tool_name: &str,
    second: &str,
) -> Result<ScenarioReport, ConformanceError> {
    let mut checks = Checks::new(
        "interleaved_signed_full_reasoning_does_not_erase_prior_thought",
        driver.provider,
    );

    let drained = driver.drive(ok_chunks(frames)).await?;
    checks.require(
        drained.completed_cleanly(),
        || "the interleaved stream must complete without errors",
    )?;
    assert_reasoning_tool_reasoning(&checks, &drained, first, tool_name, second)?;
    let signed = drained
        .choice_reasoning()
        .last()
        .is_some_and(|reasoning| reasoning.native.is_some());
    checks.require(signed, || "the post-boundary block must keep its signature")?;

    checks.note("pre-boundary thought survived; signed block completed the post-boundary part");
    Ok(checks.report())
}

/// Shared assertion: the aggregated choice is exactly
/// `[Reasoning(first), ToolCall(tool_name), Reasoning(second…)]`.
fn assert_reasoning_tool_reasoning(
    checks: &Checks,
    drained: &DrainedStream,
    first: &str,
    tool_name: &str,
    second: &str,
) -> Result<(), ConformanceError> {
    let shape: Vec<String> = drained
        .choice
        .iter()
        .map(|content| match content {
            AssistantContent::Reasoning(reasoning) => format!("reasoning:{}", reasoning.text),
            AssistantContent::ToolCall(tool_call) => {
                format!("tool:{}", tool_call.function.name)
            }
            AssistantContent::Text(text) => format!("text:{}", text.text),
            AssistantContent::Image(_) => "image".to_string(),
            AssistantContent::Opaque(opaque) => format!("opaque:{}", opaque.kind().unwrap_or("")),
        })
        .collect();
    let expected = vec![
        format!("reasoning:{first}"),
        format!("tool:{tool_name}"),
        format!("reasoning:{second}"),
    ];
    checks.require(shape == expected, || {
        format!("the boundary must survive aggregation: expected {expected:?}, observed {shape:?}")
    })
}

/// Per-provider wire fixtures for the shared scenario set.
pub mod fixtures {
    use super::*;
    use crate::driver::{Model, Transport};
    use crate::operation::Completion;
    use crate::test_utils::SequencedStreamingHttpClient;
    use crate::wire::Wire;
    use serde_json::json;

    /// Drain a full normalized stream into everything the consumer observed.
    /// Public so provider-crate conformance suites (the typed-event wires)
    /// can reuse it in their drivers.
    pub async fn drain(mut stream: crate::streaming::CompletionStream) -> DrainedStream {
        let mut items = Vec::new();
        while let Some(item) = stream.next().await {
            items.push(item.map_err(|error| ErrorReport::from(&error)));
        }
        let partial = stream.partial().choice;
        let response = stream.finish().await.ok();
        let drained = DrainedStream {
            items,
            choice: response
                .as_ref()
                .map_or(partial, |response| response.choice.clone()),
            response,
        };
        // Every fixture and cassette that drains through this helper runs
        // the lifecycle validator — the prose invariants as one executable
        // artifact (#2258 C1).
        super::assert_valid_event_stream(&drained.items, &drained.choice);
        drained
    }

    /// Drive `model` through the observed stream entry and drain it. Every
    /// wire threads the context it is handed: the trace must show the
    /// request it sent and how the attempt closed, or the wire has silently
    /// taken the context-discarding default.
    pub async fn drain_observed<W, T>(
        model: &Model<W, T>,
        request: crate::completion::CompletionRequest,
    ) -> Result<DrainedStream, ProviderError>
    where
        W: Wire<Op = Completion>,
        T: Transport<W>,
    {
        let log = std::sync::Arc::new(crate::observe::ObservationLog::default());
        let context = crate::observe::AdapterContext::new(
            log.clone(),
            crate::observe::Subject::default(),
            "conformance",
        );
        // A stream that fails to open still sent (or failed to send) a
        // request: the facts are asserted before the error propagates.
        let drained = match model.stream_observed(request, context) {
            Ok(stream) => Ok(drain(stream).await),
            Err(error) => Err(error),
        };
        let events: Vec<_> = log
            .trace()
            .observations
            .iter()
            .filter_map(|o| match &o.action {
                crate::observe::Action::Adapter { observation } => Some(observation.event.clone()),
                _ => None,
            })
            .collect();
        assert!(
            matches!(
                events.first(),
                Some(crate::observe::AdapterEvent::Started { .. })
            ),
            "the wire must attach the observation context it was handed: {events:?}"
        );
        assert!(
            matches!(
                events.last(),
                Some(crate::observe::AdapterEvent::Finished { .. })
            ),
            "the attempt must close: {events:?}"
        );
        drained
    }

    /// Lower fixture frames onto the byte transport a `SequencedStreamingHttpClient`
    /// replays. Only byte frames are valid here — an event frame in a
    /// byte-driver fixture is a fixture authoring error.
    fn byte_chunks(chunks: WireChunks) -> Result<Vec<http_client::Result<Bytes>>, ProviderError> {
        chunks
            .into_iter()
            .map(|chunk| match chunk {
                Ok(WireInput::Bytes(bytes)) => Ok(Ok(bytes)),
                Ok(WireInput::Event(_)) => Err(ProviderError::Provider(
                    "typed-event frame fed to a byte-transport driver".to_string(),
                )),
                Err(error) => Ok(Err(error)),
            })
            .collect()
    }

    /// Drive a byte-transport wire over the scripted chunks: bind the
    /// provider's wire to the replaying transport, open the model `bind`
    /// names, and drain the observed stream.
    ///
    /// Every byte-wire conformance driver is this walk; since the wire
    /// unification they differ only in which wire they name, so the walk is
    /// written once and each family supplies its own `bind`.
    fn byte_driver<W>(
        provider: &'static str,
        bind: fn(SequencedStreamingHttpClient) -> Model<W, SequencedStreamingHttpClient>,
    ) -> WireDriver
    where
        W: Wire<Op = Completion, Payload = crate::wire::Encoded, Frame = crate::wire::WireFrame>,
    {
        WireDriver::new(provider, move |chunks| {
            Box::pin(async move {
                let model = bind(SequencedStreamingHttpClient::new(byte_chunks(chunks)?));
                let request = CompletionRequest::new("hello");
                drain_observed(&model, request).await
            })
        })
    }

    fn sse(frame: &serde_json::Value) -> WireInput {
        WireInput::Bytes(Bytes::from(format!("data: {frame}\n\n")))
    }

    fn sse_raw(data: &str) -> WireInput {
        WireInput::Bytes(Bytes::from(format!("data: {data}\n\n")))
    }

    /// The frame's SSE text, for buffered-body pipelines that re-parse a
    /// whole body string.
    fn frame_text(frame: &WireInput) -> String {
        frame
            .as_bytes()
            .map(|bytes| String::from_utf8_lossy(bytes).into_owned())
            .unwrap_or_default()
    }

    /// OpenAI chat-completions wire (the shared OpenAI-compatible SSE path).
    pub mod openai_chat {
        use super::*;

        fn driver() -> WireDriver {
            byte_driver("openai", |transport| {
                crate::driver::Model::new(
                    crate::providers::openai::wire::OpenAIConfig::with_key(
                        &crate::providers::openai::wire::OPENAI,
                        "test-key",
                    )
                    .chat("gpt-4o"),
                    transport,
                )
            })
        }

        /// The chat-completions fixture.
        pub fn fixture() -> ProviderWireFixture {
            ProviderWireFixture {
                driver: driver(),
                text_frames: vec![sse(&json!({
                    "id": "chatcmpl-1",
                    "model": "gpt-4o-2024-08-06",
                    "choices": [{"index": 0, "delta": {"content": "hi"}, "finish_reason": null}],
                    "usage": null,
                }))],
                expected_texts: vec!["hi"],
                tool_call_frames: vec![
                    sse(&json!({
                        "choices": [{"index": 0, "delta": {"tool_calls": [{
                            "index": 0,
                            "id": "call_1",
                            "type": "function",
                            "function": {"name": "get_weather", "arguments": ""},
                        }]}, "finish_reason": null}],
                    })),
                    sse(&json!({
                        "choices": [{"index": 0, "delta": {"tool_calls": [{
                            "index": 0,
                            "function": {"arguments": "{\"city\":\"Tokyo\"}"},
                        }]}, "finish_reason": null}],
                    })),
                    // No `finish_reason` chunk: on the chat wire that IS the
                    // terminal signal, and these frames must stop short of it.
                    // EOF/error cleanup still flushes the completed call.
                ],
                expected_tool_name: "get_weather",
                partial_tool_call_frames: Some(vec![sse(&json!({
                    "choices": [{"index": 0, "delta": {"tool_calls": [{
                        "index": 0,
                        "id": "call_1",
                        "type": "function",
                        "function": {"name": "get_weather", "arguments": "{\"cit"},
                    }]}, "finish_reason": null}],
                }))]),
                terminal_frames: vec![
                    sse(&json!({
                        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
                        "usage": null,
                    })),
                    sse(&json!({
                        "choices": [],
                        "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
                    })),
                    sse_raw("[DONE]"),
                ],
                expected_usage_total: 15,
                expected_finish_reason: Some(FinishReason::Stop),
                zero_usage_terminal_frames: Some(vec![
                    sse(&json!({
                        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
                        "usage": null,
                    })),
                    sse_raw("[DONE]"),
                ]),
                bare_terminal_frames: Some(vec![sse_raw("[DONE]")]),
                malformed_frame: Some(sse_raw("{not json")),
                unknown_event_frame: None,
                // A wrongly-typed `content` is tolerated by the lenient delta
                // decode; a wrongly-typed `choices` is a genuine schema defect
                // of the known chunk shape.
                // The Chat decoder reads leniently, as pi does: `choices`
                // of another type is a chunk with no choices, not a defect.
                defective_known_frame: None,
                // The Azure `prompt_filter_results` prelude: a choice with no
                // `delta` at all.
                delta_less_prelude_frame: Some(sse_raw(
                    r#"{"id":"","object":"","choices":[{"prompt_index":0,"content_filter_results":{"hate":{"filtered":false,"severity":"safe"}}}]}"#,
                )),
                refusal: None,
                // Deliberately absent — a documented named skip, not a gap.
                // The chat wire streams tool calls as fragments that only
                // finalize at a boundary the wire itself signals (next slot,
                // finish_reason, terminal), so the AGGREGATED part order
                // cannot pin reasoning→tool→reasoning without risky early
                // finalization; and the chat request format erases part
                // order on replay regardless (`tool_calls` is a flat array
                // beside `content`). The boundary the adapter does own —
                // closing the open reasoning block before emitting tool
                // content — is pinned at emission level by the adapter's
                // unit tests and by the driver's debug-mode sequence laws.
                interleaved_reasoning: None,
            }
        }
    }

    /// OpenAI Responses API wire.
    pub mod openai_responses {
        use super::*;

        /// The driver alone, for the reasoning-specific scenarios.
        pub fn driver() -> WireDriver {
            byte_driver("openai", |transport| {
                crate::driver::Model::new(
                    crate::providers::openai::OpenAIConfig::new("test-key").responses("gpt-5.4"),
                    transport,
                )
            })
        }

        fn completed_response(
            usage: Option<&serde_json::Value>,
            output: &serde_json::Value,
        ) -> serde_json::Value {
            json!({
                "id": "resp_1",
                "object": "response",
                "created_at": 0,
                "status": "completed",
                "model": "gpt-5.4",
                "output": output,
                "tools": [],
                "usage": usage,
            })
        }

        fn terminal(usage: Option<&serde_json::Value>, output: &serde_json::Value) -> WireInput {
            sse(&json!({
                "type": "response.completed",
                "sequence_number": 99,
                "response": completed_response(usage, output),
            }))
        }

        fn usage_json() -> serde_json::Value {
            json!({
                "input_tokens": 10,
                "output_tokens": 5,
                "output_tokens_details": {"reasoning_tokens": 0},
                "total_tokens": 15,
            })
        }

        fn text_delta(text: &str) -> WireInput {
            sse(&json!({
                "type": "response.output_text.delta",
                "content_index": 0,
                "delta": text,
                "item_id": "msg_1",
                "output_index": 0,
                "sequence_number": 1,
            }))
        }

        fn tool_call_done() -> WireInput {
            sse(&json!({
                "type": "response.output_item.done",
                "output_index": 0,
                "sequence_number": 2,
                "item": {
                    "type": "function_call",
                    "id": "fc_1",
                    "arguments": "{\"city\":\"Tokyo\"}",
                    "call_id": "call_1",
                    "name": "get_weather",
                    "status": "completed",
                },
            }))
        }

        /// Synthetic twin of the recorded
        /// `openai/streaming_grammar/incomplete_mid_tool_call` cassette: a
        /// forced tool call cut by `max_output_tokens` mid-arguments. The wire
        /// restates the call on `response.output_item.done` with the
        /// arguments truncated mid-JSON and item status `incomplete`, then
        /// ends with a genuine `response.incomplete` terminal.
        pub fn incomplete_mid_tool_call_frames() -> Vec<WireInput> {
            vec![
                sse(&json!({
                    "type": "response.output_item.added",
                    "output_index": 0,
                    "sequence_number": 1,
                    "item": {
                        "type": "function_call",
                        "id": "fc_1",
                        "arguments": "",
                        "call_id": "call_1",
                        "name": "add",
                        "status": "in_progress",
                    },
                })),
                sse(&json!({
                    "type": "response.function_call_arguments.delta",
                    "item_id": "fc_1",
                    "output_index": 0,
                    "sequence_number": 2,
                    "delta": "{\"x",
                })),
                sse(&json!({
                    "type": "response.function_call_arguments.delta",
                    "item_id": "fc_1",
                    "output_index": 0,
                    "sequence_number": 3,
                    "delta": "\":48151",
                })),
                sse(&json!({
                    "type": "response.function_call_arguments.done",
                    "item_id": "fc_1",
                    "output_index": 0,
                    "sequence_number": 4,
                    "arguments": "{\"x\":48151",
                })),
                sse(&json!({
                    "type": "response.output_item.done",
                    "output_index": 0,
                    "sequence_number": 5,
                    "item": {
                        "type": "function_call",
                        "id": "fc_1",
                        "arguments": "{\"x\":48151",
                        "call_id": "call_1",
                        "name": "add",
                        "status": "incomplete",
                    },
                })),
                sse(&json!({
                    "type": "response.incomplete",
                    "sequence_number": 6,
                    "response": {
                        "id": "resp_1",
                        "object": "response",
                        "created_at": 0,
                        "status": "incomplete",
                        "incomplete_details": {"reason": "max_output_tokens"},
                        "model": "gpt-5.4",
                        "output": [{
                            "type": "function_call",
                            "id": "fc_1",
                            "arguments": "{\"x\":48151",
                            "call_id": "call_1",
                            "name": "add",
                            "status": "incomplete",
                        }],
                        "tools": [],
                        "usage": usage_json(),
                    },
                })),
            ]
        }

        fn reasoning_done_item(
            id: &str,
            summary: &serde_json::Value,
            content: &serde_json::Value,
            encrypted: Option<&str>,
        ) -> WireInput {
            let mut item = json!({
                "type": "reasoning",
                "id": id,
                "summary": summary,
                "content": content,
                "status": "completed",
            });
            if let (Some(encrypted), Some(object)) = (encrypted, item.as_object_mut()) {
                object.insert("encrypted_content".to_string(), json!(encrypted));
            }
            sse(&json!({
                "type": "response.output_item.done",
                "output_index": 0,
                "sequence_number": 3,
                "item": item,
            }))
        }

        /// The Responses-API fixture.
        pub fn fixture() -> ProviderWireFixture {
            ProviderWireFixture {
                driver: driver(),
                text_frames: vec![text_delta("hi")],
                expected_texts: vec!["hi"],
                tool_call_frames: vec![tool_call_done()],
                expected_tool_name: "get_weather",
                partial_tool_call_frames: Some(vec![
                    sse(&json!({
                        "type": "response.output_item.added",
                        "output_index": 0,
                        "sequence_number": 1,
                        "item": {
                            "type": "function_call",
                            "id": "fc_1",
                            "arguments": "",
                            "call_id": "call_1",
                            "name": "get_weather",
                            "status": "in_progress",
                        },
                    })),
                    sse(&json!({
                        "type": "response.function_call_arguments.delta",
                        "item_id": "fc_1",
                        "output_index": 0,
                        "sequence_number": 2,
                        "delta": "{\"cit",
                    })),
                ]),
                terminal_frames: vec![terminal(Some(&usage_json()), &json!([]))],
                expected_usage_total: 15,
                expected_finish_reason: Some(FinishReason::Stop),
                zero_usage_terminal_frames: Some(vec![terminal(None, &json!([]))]),
                bare_terminal_frames: None,
                malformed_frame: Some(sse_raw("{not json")),
                unknown_event_frame: Some(sse(&json!({
                    "type": "response.web_search_call.searching",
                    "output_index": 0,
                    "sequence_number": 4,
                    "item_id": "ws_1",
                }))),
                // The decoder reads only the fields it builds blocks from, and
                // reads those leniently, so no known event fails a reply (#2668).
                defective_known_frame: None,
                delta_less_prelude_frame: None,
                refusal: Some(RefusalFixture {
                    frames: vec![sse(&json!({
                        "type": "response.refusal.delta",
                        "content_index": 0,
                        "delta": "I cannot help with that.",
                        "item_id": "msg_1",
                        "output_index": 0,
                        "sequence_number": 1,
                    }))],
                    expected_text: "I cannot help with that.",
                }),
                interleaved_reasoning: None,
            }
        }

        /// The buffered-body pipeline the ChatGPT backend uses: the SSE body
        /// is re-parsed after the fact and merged with the terminal response
        /// body, per content kind.
        ///
        /// Drives the *real* entry — `Model::call` on the
        /// ChatGPT wire whose HTTP double answers the `/responses` POST with
        /// the scripted SSE body — so the scenario exercises the buffered
        /// fold itself rather than a mirrored copy of it (#2258 review, F8
        /// drift risk).
        pub fn buffered_driver() -> BufferedBodyDriver {
            BufferedBodyDriver::new("chatgpt", |body| {
                Box::pin(async move {
                    let model = crate::driver::Model::new(
                        crate::providers::openai::OpenAIConfig::with_key(
                            &crate::providers::chatgpt::DIALECT,
                            "test-token",
                        )
                        .with_account_id("account-id")
                        .responses("gpt-5.4"),
                        crate::test_utils::RecordingHttpClient::new(body),
                    );
                    let request = CompletionRequest::new("hello");
                    let response = model.call(request).await?;
                    Ok(response.choice)
                })
            })
        }

        fn message_output(text: &str) -> serde_json::Value {
            json!([{
                "type": "message",
                "id": "msg_1",
                "role": "assistant",
                "status": "completed",
                "content": [{"type": "output_text", "text": text, "annotations": []}],
            }])
        }

        /// A terminal whose body carries text never seen as a delta.
        pub fn terminal_body_only_sse_body(text: &str) -> String {
            frame_text(&terminal(Some(&usage_json()), &message_output(text)))
        }

        /// A streamed delta plus a terminal body restating the same text.
        pub fn terminal_body_and_delta_sse_body(text: &str) -> String {
            let frames = [
                text_delta(text),
                terminal(Some(&usage_json()), &message_output(text)),
            ];
            frames.iter().map(frame_text).collect()
        }

        /// A streamed delta whose terminal body carries no output items — the
        /// gpt-5.x shape the buffered fallback exists for.
        pub fn delta_only_sse_body(text: &str) -> String {
            let frames = [text_delta(text), terminal(Some(&usage_json()), &json!([]))];
            frames.iter().map(frame_text).collect()
        }

        /// The ChatGPT envelope-less replay shape (#2258 F3): a summary delta
        /// with NO envelope bookkeeping at all (repair injects
        /// `output_index: 0`, minting the `output-0` identity), then the
        /// item's envelope-full `output_item.done` restating the summary
        /// under its real `rs_*` id, then the terminal. The done item must
        /// adopt the minted per-slot identity and supersede the delta build.
        pub fn envelope_less_reasoning_supersede_sse_body() -> (String, &'static str) {
            let delta = json!({
                "type": "response.reasoning_summary_text.delta",
                "delta": "step 1",
            });
            let frames = [
                sse(&delta),
                reasoning_done_item(
                    "rs_1",
                    &json!([{"type": "summary_text", "text": "step 1"}]),
                    &json!([]),
                    None,
                ),
                terminal(Some(&usage_json()), &json!([])),
            ];
            (frames.iter().map(frame_text).collect(), "step 1")
        }

        /// Summary deltas followed by their item's full `output_item.done`
        /// block, then the terminal. The deltas carry `item_id` on the wire;
        /// the full block restates the summary.
        pub fn reasoning_summary_supersede_frames() -> (Vec<WireInput>, &'static str) {
            let frames = vec![
                sse(&json!({
                    "type": "response.reasoning_summary_text.delta",
                    "item_id": "rs_1",
                    "output_index": 0,
                    "summary_index": 0,
                    "sequence_number": 1,
                    "delta": "step 1",
                })),
                reasoning_done_item(
                    "rs_1",
                    &json!([{"type": "summary_text", "text": "step 1"}]),
                    &json!([]),
                    None,
                ),
                terminal(Some(&usage_json()), &json!([])),
            ];
            (frames, "step 1")
        }

        /// One reasoning item done-block carrying two summary parts, visible
        /// text, and encrypted content under a single item id.
        pub fn multi_part_reasoning_frames() -> (Vec<WireInput>, Vec<&'static str>) {
            let frames = vec![
                reasoning_done_item(
                    "rs_1",
                    &json!([
                        {"type": "summary_text", "text": "s1"},
                        {"type": "summary_text", "text": "s2"},
                    ]),
                    &json!([{"type": "reasoning_text", "text": "visible"}]),
                    Some("enc_blob"),
                ),
                terminal(Some(&usage_json()), &json!([])),
            ];
            (frames, vec!["s1", "s2", "visible", "enc_blob"])
        }

        /// A reasoning delta, an interleaved tool call, then the reasoning
        /// item's completed block and the terminal.
        pub fn interleaved_reasoning_frames() -> (Vec<WireInput>, &'static str) {
            let frames = vec![
                sse(&json!({
                    "type": "response.reasoning_text.delta",
                    "item_id": "rs_2",
                    "output_index": 0,
                    "content_index": 0,
                    "sequence_number": 1,
                    "delta": "full ",
                })),
                // The call is the reply's next output item.
                sse(&json!({
                    "type": "response.output_item.done",
                    "output_index": 1,
                    "sequence_number": 2,
                    "item": {
                        "type": "function_call",
                        "id": "fc_1",
                        "arguments": "{\"city\":\"Tokyo\"}",
                        "call_id": "call_1",
                        "name": "get_weather",
                        "status": "completed",
                    },
                })),
                reasoning_done_item(
                    "rs_2",
                    &json!([]),
                    &json!([{"type": "reasoning_text", "text": "full reasoning"}]),
                    None,
                ),
                terminal(Some(&usage_json()), &json!([])),
            ];
            (frames, "full reasoning")
        }
    }

    /// Gemini REST (`streamGenerateContent`) SSE wire.
    pub mod gemini_rest {
        use super::*;

        fn driver() -> WireDriver {
            byte_driver("gemini", |transport| {
                crate::driver::Model::new(
                    crate::providers::gemini::GeminiConfig::new("test-key").completion(
                        crate::providers::gemini::completion::GEMINI_2_5_PRO_PREVIEW_06_05,
                    ),
                    transport,
                )
            })
        }

        /// The Gemini REST fixture.
        pub fn fixture() -> ProviderWireFixture {
            ProviderWireFixture {
                driver: driver(),
                text_frames: vec![sse(&json!({
                    "candidates": [{"content": {"parts": [{"text": "hi"}], "role": "model"}}],
                    "responseId": "resp-1",
                    "modelVersion": "gemini-2.5-pro",
                }))],
                expected_texts: vec!["hi"],
                tool_call_frames: vec![sse(&json!({
                    "candidates": [{"content": {"parts": [{
                        "functionCall": {"name": "get_weather", "args": {"city": "Tokyo"}},
                    }], "role": "model"}}],
                    "responseId": "resp-1",
                    "modelVersion": "gemini-2.5-pro",
                }))],
                expected_tool_name: "get_weather",
                // Gemini delivers tool calls whole; arguments never stream.
                partial_tool_call_frames: None,
                terminal_frames: vec![sse(&json!({
                    "candidates": [{
                        "content": {"parts": [], "role": "model"},
                        "finishReason": "STOP",
                    }],
                    "usageMetadata": {
                        "promptTokenCount": 5,
                        "candidatesTokenCount": 2,
                        "totalTokenCount": 7,
                    },
                    "responseId": "resp-1",
                    "modelVersion": "gemini-2.5-pro",
                }))],
                expected_usage_total: 7,
                expected_finish_reason: Some(FinishReason::Stop),
                zero_usage_terminal_frames: Some(vec![sse(&json!({
                    "candidates": [{
                        "content": {"parts": [], "role": "model"},
                        "finishReason": "STOP",
                    }],
                    "responseId": "resp-1",
                    "modelVersion": "gemini-2.5-pro",
                }))]),
                bare_terminal_frames: None,
                malformed_frame: Some(sse_raw("{not json")),
                // The wire has no event tag; valid JSON carrying neither
                // `candidates` nor `usageMetadata` is unrecognizable and must
                // be warn-skipped, not silently decoded as an empty chunk.
                unknown_event_frame: Some(sse_raw(r#"{"noise":true}"#)),
                defective_known_frame: Some(sse_raw(r#"{"candidates": 42}"#)),
                delta_less_prelude_frame: None,
                refusal: None,
                interleaved_reasoning: Some(interleaved_thought_fixture()),
            }
        }

        fn chunk(parts: &serde_json::Value) -> WireInput {
            sse(&json!({
                "candidates": [{"content": {"parts": parts, "role": "model"}}],
                "responseId": "resp-1",
                "modelVersion": "gemini-2.5-pro",
            }))
        }

        fn terminal_frame() -> WireInput {
            sse(&json!({
                "candidates": [{
                    "content": {"parts": [], "role": "model"},
                    "finishReason": "STOP",
                }],
                "usageMetadata": {
                    "promptTokenCount": 5,
                    "candidatesTokenCount": 2,
                    "totalTokenCount": 7,
                },
                "responseId": "resp-1",
                "modelVersion": "gemini-2.5-pro",
            }))
        }

        /// Thought delta, interleaved tool call, thought delta, terminal —
        /// the constant-id (`reasoning-0`) interleaving shape.
        fn interleaved_thought_fixture() -> InterleavedReasoningFixture {
            InterleavedReasoningFixture {
                frames: vec![
                    chunk(&json!([{"text": "before tool", "thought": true}])),
                    chunk(&json!([{
                        "functionCall": {"name": "get_weather", "args": {"city": "Tokyo"}},
                    }])),
                    chunk(&json!([{"text": "after tool", "thought": true}])),
                    terminal_frame(),
                ],
                first_reasoning: "before tool",
                tool_name: "get_weather",
                second_reasoning: "after tool",
            }
        }

        /// Thought delta, interleaved tool call, then a signed full thought
        /// chunk carrying non-empty text — the F1 erasure shape.
        pub fn interleaved_signed_thought_frames()
        -> (Vec<WireInput>, &'static str, &'static str, &'static str) {
            let frames = vec![
                chunk(&json!([{"text": "before tool", "thought": true}])),
                chunk(&json!([{
                    "functionCall": {"name": "get_weather", "args": {"city": "Tokyo"}},
                }])),
                chunk(&json!([{
                    "text": "signed conclusion",
                    "thought": true,
                    "thoughtSignature": "sig-1",
                }])),
                terminal_frame(),
            ];
            (frames, "before tool", "get_weather", "signed conclusion")
        }
    }

    /// Gemini Interactions SSE wire (`event_type`-tagged events).
    pub mod interactions {
        use super::*;

        fn driver() -> WireDriver {
            byte_driver("gemini", |transport| {
                crate::driver::Model::new(
                    crate::providers::gemini::GeminiConfig::new("test-key")
                        .interactions("gemini-2.5-pro"),
                    transport,
                )
            })
        }

        fn completed(usage: Option<serde_json::Value>) -> WireInput {
            let mut interaction = json!({
                "id": "int-1",
                "model": "gemini-2.5-pro",
                "status": "completed",
            });
            if let (Some(usage), Some(object)) = (usage, interaction.as_object_mut()) {
                object.insert("usage".to_string(), usage);
            }
            sse(&json!({
                "event_type": "interaction.completed",
                "interaction": interaction,
            }))
        }

        /// The Interactions fixture.
        pub fn fixture() -> ProviderWireFixture {
            ProviderWireFixture {
                driver: driver(),
                text_frames: vec![sse(&json!({
                    "event_type": "step.delta",
                    "index": 0,
                    "delta": {"type": "text", "text": "hi"},
                }))],
                expected_texts: vec!["hi"],
                tool_call_frames: vec![sse(&json!({
                    "event_type": "step.delta",
                    "index": 0,
                    "delta": {
                        "type": "function_call",
                        "name": "get_weather",
                        "arguments": {"city": "Tokyo"},
                        "id": "call-1",
                    },
                }))],
                expected_tool_name: "get_weather",
                // The Interactions wire delivers function calls whole;
                // arguments never stream.
                partial_tool_call_frames: None,
                terminal_frames: vec![completed(Some(json!({
                    "total_input_tokens": 5,
                    "total_output_tokens": 2,
                    "total_tokens": 7,
                })))],
                expected_usage_total: 7,
                expected_finish_reason: Some(FinishReason::Stop),
                zero_usage_terminal_frames: Some(vec![completed(None)]),
                bare_terminal_frames: None,
                malformed_frame: Some(sse_raw("{not json")),
                unknown_event_frame: Some(sse(&json!({
                    "event_type": "future.event",
                    "index": 0,
                }))),
                // A known tag (`step.delta`) whose delta, the content of its
                // block, is not an object fails the reply, never skipped.
                defective_known_frame: Some(sse_raw(
                    r#"{"event_type":"step.delta","index":0,"delta":42}"#,
                )),
                delta_less_prelude_frame: None,
                refusal: None,
                interleaved_reasoning: Some(interleaved_thought_fixture()),
            }
        }

        /// A thought step, a function-call step, a second thought step,
        /// terminal: each step is its own block on the Interactions wire.
        fn interleaved_thought_fixture() -> InterleavedReasoningFixture {
            let frames = vec![
                sse(&json!({
                    "event_type": "step.delta",
                    "index": 0,
                    "delta": {
                        "type": "thought_summary",
                        "content": {"type": "text", "text": "before tool"},
                    },
                })),
                sse(&json!({
                    "event_type": "step.delta",
                    "index": 1,
                    "delta": {
                        "type": "function_call",
                        "name": "get_weather",
                        "arguments": {"city": "Tokyo"},
                        "id": "call-1",
                    },
                })),
                sse(&json!({
                    "event_type": "step.delta",
                    "index": 2,
                    "delta": {
                        "type": "thought_summary",
                        "content": {"type": "text", "text": "after tool"},
                    },
                })),
                completed(Some(json!({
                    "total_input_tokens": 5,
                    "total_output_tokens": 2,
                    "total_tokens": 7,
                }))),
            ];
            InterleavedReasoningFixture {
                frames,
                first_reasoning: "before tool",
                tool_name: "get_weather",
                second_reasoning: "after tool",
            }
        }
    }

    /// Anthropic Messages SSE wire (`type`-tagged events, index-as-id blocks).
    pub mod anthropic {
        use super::*;

        fn driver() -> WireDriver {
            byte_driver("anthropic", |transport| {
                crate::driver::Model::new(
                    crate::providers::anthropic::wire::AnthropicConfig::new("test-key")
                        .completion(crate::providers::anthropic::completion::CLAUDE_SONNET_4_6),
                    transport,
                )
            })
        }

        fn message_start() -> WireInput {
            sse(&json!({
                "type": "message_start",
                "message": {
                    "id": "msg_1",
                    "role": "assistant",
                    "content": [],
                    "model": "claude-sonnet-4-6",
                    "stop_reason": null,
                    "stop_sequence": null,
                    "usage": {"input_tokens": 5, "output_tokens": 0},
                },
            }))
        }

        /// The Anthropic fixture.
        pub fn fixture() -> ProviderWireFixture {
            ProviderWireFixture {
                driver: driver(),
                text_frames: vec![
                    message_start(),
                    sse(&json!({
                        "type": "content_block_start",
                        "index": 0,
                        "content_block": {"type": "text", "text": ""},
                    })),
                    sse(&json!({
                        "type": "content_block_delta",
                        "index": 0,
                        "delta": {"type": "text_delta", "text": "hi"},
                    })),
                ],
                expected_texts: vec!["hi"],
                tool_call_frames: vec![
                    sse(&json!({
                        "type": "content_block_start",
                        "index": 0,
                        "content_block": {
                            "type": "tool_use",
                            "id": "toolu_1",
                            "name": "get_weather",
                            "input": {},
                        },
                    })),
                    sse(&json!({
                        "type": "content_block_delta",
                        "index": 0,
                        "delta": {"type": "input_json_delta", "partial_json": "{\"city\":\"Tokyo\"}"},
                    })),
                    // `content_block_stop` completes the call; the stream
                    // terminal (`message_delta`) is deliberately absent.
                    sse(&json!({"type": "content_block_stop", "index": 0})),
                ],
                expected_tool_name: "get_weather",
                partial_tool_call_frames: Some(vec![
                    sse(&json!({
                        "type": "content_block_start",
                        "index": 0,
                        "content_block": {
                            "type": "tool_use",
                            "id": "toolu_1",
                            "name": "get_weather",
                            "input": {},
                        },
                    })),
                    sse(&json!({
                        "type": "content_block_delta",
                        "index": 0,
                        "delta": {"type": "input_json_delta", "partial_json": "{\"cit"},
                    })),
                ]),
                terminal_frames: vec![sse(&json!({
                    "type": "message_delta",
                    "delta": {"stop_reason": "end_turn", "stop_sequence": null},
                    "usage": {"output_tokens": 4},
                }))],
                // input 5 (from message_start) + output 4.
                expected_usage_total: 9,
                expected_finish_reason: Some(FinishReason::Stop),
                // The Anthropic terminal always carries `usage`; there is no
                // usage-less genuine terminal to spell on this wire.
                zero_usage_terminal_frames: None,
                // `message_stop` carries no data of its own and must not
                // fabricate a terminal record.
                bare_terminal_frames: Some(vec![sse(&json!({"type": "message_stop"}))]),
                malformed_frame: Some(sse_raw("{not json")),
                unknown_event_frame: Some(sse(&json!({
                    "type": "content_block_heartbeat",
                    "index": 0,
                }))),
                // A known tag (`content_block_delta`) with a schema-defective
                // payload must classify `Corrupt`, never `Unknown`.
                defective_known_frame: Some(sse_raw(
                    r#"{"type":"content_block_delta","index":0,"delta":42}"#,
                )),
                delta_less_prelude_frame: None,
                refusal: None,
                interleaved_reasoning: None,
            }
        }
    }
}