rig_core/

wire.rs

//! Operations, provider wires and their decoders. An [`Operation`] names what
//! goes in, what comes out event by event and what ends a reply, and its
//! [`Fold`] turns one reply's events and end into the response. A [`Wire`]
//! describes a provider endpoint as plain data, encodes requests without
//! transport access, and names a fresh [`Decoder`] for each reply. A decoder
//! is told nothing about how the reply arrives, so buffered and streamed
//! replies go through the same decoder and fold.
//!
//! ```
//! use rig_core::wire::WireFrame;
//!
//! let frame = WireFrame::Bytes(b"response".to_vec());
//! assert_eq!(frame.as_str(), "response");
//! ```

use std::sync::{Arc, Mutex};

use crate::error::{EncodeError, ProviderError};
use crate::http_client::MultipartForm;
use crate::wasm_compat::{WasmCompatSend, WasmCompatSync};

pub use crate::http_client::framing::{Framing, WireFrame};
pub use crate::observe::{
    AdapterErrorEnvelope, AdapterEvent, AdapterUsage, AdapterVerdict, ObservationSink,
};

mod citation;
pub mod document;
pub(crate) mod secret;

pub use citation::{SpanUnit, WireCitation, WireSpan};
pub use secret::Secret;

/// The request a wire sends, and how its reply is framed.
///
/// Data only: built by [`Wire::encode`] from the wire and the request, and
/// read by the driver. A wire never touches a socket or a request extension.
///
/// `Debug` shows request methods and URI paths, not schemes, authorities,
/// queries, header values or bodies. Paths are not scrubbed: callers must
/// still avoid placing sensitive data in them.
pub struct Encoded {
    /// The HTTP request.
    pub request: http::Request<Body>,
    /// How the reply's bytes split into frames.
    pub framing: Framing,
    /// The reply header carrying the provider's transport request id
    /// (Anthropic `request-id`, OpenAI `x-request-id`), when the provider
    /// reports one. `None` is "does not report one", never an error.
    pub request_id_header: Option<&'static str>,
    /// Whether a streamed reply may omit `Content-Type` (one gateway
    /// replays Responses bodies without it). A *wrong* content type is
    /// still rejected.
    pub relaxed_content_type: bool,
    /// Stable endpoint template for observation grouping, without base-URL
    /// prefixes or interpolated values. `None` uses the concrete request path.
    pub route: Option<&'static str>,
    /// Reads observation facts off each raw reply payload before
    /// normalization discards them. `None` projects nothing.
    pub project: Option<Projector>,
    /// Whether a frame carries only analysis metadata: it still decodes, but
    /// does not advance observation's EOF and corruption positions.
    pub analysis_only: Option<fn(&WireFrame) -> bool>,
}

/// Reads one raw payload's observation facts (verdicts, usage, ids, error
/// envelopes) through the sink. Text it forwards goes through
/// [`ObservationSink::scrub`]: a payload can echo credentials.
pub type Projector = fn(&[u8], &mut ObservationSink<'_>);

impl Encoded {
    /// One request whose provider reports no transport request id.
    pub fn new(request: http::Request<Body>, framing: Framing) -> Self {
        Self {
            request,
            framing,
            request_id_header: None,
            relaxed_content_type: false,
            route: None,
            project: None,
            analysis_only: None,
        }
    }

    /// Name the reply header carrying the provider's transport request id.
    pub fn with_request_id_header(mut self, header: Option<&'static str>) -> Self {
        self.request_id_header = header;
        self
    }

    /// Accept a streamed reply that names no content type.
    pub fn with_relaxed_content_type(mut self) -> Self {
        self.relaxed_content_type = true;
        self
    }

    /// Name the endpoint template observation groups attempts under.
    pub fn with_route(mut self, route: Option<&'static str>) -> Self {
        self.route = route;
        self
    }

    /// Read observation facts off each reply payload through `project`.
    pub fn with_projection(mut self, project: Projector) -> Self {
        self.project = Some(project);
        self
    }

    /// Exempt frames `analysis_only` accepts from observation's frame
    /// positions.
    pub fn with_analysis_only(mut self, analysis_only: fn(&WireFrame) -> bool) -> Self {
        self.analysis_only = Some(analysis_only);
        self
    }
}

/// A request body: bytes, or a multipart form for the upload endpoints.
///
/// `Debug` prints the body's shape and size, never its bytes: a request
/// body carries prompts, documents and uploaded files.
pub enum Body {
    /// A serialized body (JSON for every wire in this crate, or empty).
    Bytes(Vec<u8>),
    /// A multipart form (audio transcription, image edits).
    Multipart(MultipartForm),
}

impl Body {
    /// An empty body, for a `GET`.
    pub fn empty() -> Self {
        Self::Bytes(Vec::new())
    }
}

impl std::fmt::Debug for Body {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Bytes(bytes) => write!(f, "Bytes({} bytes)", bytes.len()),
            Self::Multipart(_) => f.write_str("Multipart"),
        }
    }
}

impl std::fmt::Debug for Encoded {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Diagnostics omit query, authority, headers and bodies. A caller's
        // path can still contain sensitive data; it is not scrubbed here.
        f.debug_struct("Encoded")
            .field(
                "request",
                &(self.request.method(), self.request.uri().path()),
            )
            .field("framing", &self.framing)
            .field("request_id_header", &self.request_id_header)
            .field("relaxed_content_type", &self.relaxed_content_type)
            .field("route", &self.route)
            .field("project", &self.project.is_some())
            .finish()
    }
}

/// Reply mode used by the wire to select request encoding and framing.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Mode {
    /// One whole reply ([`Model::call`](crate::driver::Model::call)).
    Unary,
    /// A streamed reply ([`Model::stream`](crate::driver::Model::stream)).
    Streaming,
}

/// An operation: what goes in, what comes out event by event, what ends a
/// reply, and how those fold into one response.
///
/// Implemented once per operation, never per provider, so every provider of
/// an operation is interchangeable behind a [`DynModel`](crate::DynModel).
pub trait Operation: Sized + 'static {
    /// The normalized request this operation accepts.
    type Request: WasmCompatSend + 'static;
    /// One decoded step of a reply.
    type Event: WasmCompatSend + 'static;
    /// What the provider sends when it ends the reply. A fold needs one to
    /// produce a response, so a reply that stopped early has none.
    type End: WasmCompatSend + 'static;
    /// The normalized response the events and the end fold into.
    type Response: WasmCompatSend + 'static;
    /// One reply's state: it absorbs the events the consumer sees, then
    /// folds them with the end into the response.
    type Fold: Fold<Self> + WasmCompatSend + 'static;
    /// Who builds the events: [`Free`] for events a decoder writes itself,
    /// [`Assembled`] for the completion events only the writer's part
    /// handles build.
    type Emit: Emit<Self>;

    /// The state for one reply to `request`, before its first frame. The
    /// call says who answers and in which mode; an operation that records
    /// telemetry opens its span here and hands it to
    /// [`Call::instrument`].
    fn fold(request: &Self::Request, call: &mut Call<'_>) -> Self::Fold;

    /// Shape `request` for the wire `wire` describes, or reject a request
    /// no provider of this operation can answer. The driver calls it before
    /// the request is encoded, so a rejected request reaches no wire or
    /// transport. Passes every request through by default.
    fn prepare(
        request: Self::Request,
        wire: &Descriptor<'_>,
    ) -> Result<Self::Request, ProviderError> {
        let _ = wire;
        Ok(request)
    }

    /// The normalized usage one reply's `end` carries. The driver hands it
    /// to the attempt's one closing call when the reply ends, so the
    /// observation is the usage the decoder already normalized and never a
    /// second parse of the payload. Observes nothing by default.
    fn observed_usage(end: &Self::End) -> Option<&crate::completion::Usage> {
        let _ = end;
        None
    }
}

/// What a model keeps from preparing one request for the next
/// ([`Descriptor::prepared`]): a completion's history as it was last
/// adapted, so a history that grew is adapted only from its last turn on. A
/// model's clones share it.
#[derive(Clone, Debug, Default)]
pub struct Prepared(pub(crate) Arc<Mutex<crate::completion::history::Memo>>);

/// What a model kept never makes two models differ.
impl PartialEq for Prepared {
    fn eq(&self, _: &Self) -> bool {
        true
    }
}

/// Events a decoder builds and writes with [`Out::event`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Free {}

/// Events only the completion writer's part handles build.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Assembled {}

/// Who builds an operation's events: [`Free`] or [`Assembled`]. Sealed.
pub trait Emit<Op: Operation>: reply::Closing<Op> {}

impl<Op: Operation> reply::Closing<Op> for Free {
    fn close(_shared: &mut Shared<Op>) {}
}

impl<Op: Operation> Emit<Op> for Free {}

/// What the driver knows about a call before its first frame.
pub struct Call<'a> {
    /// What the wire says about itself.
    pub wire: &'a Descriptor<'a>,
    /// How the reply arrives.
    pub mode: Mode,
    pub(crate) span: tracing::Span,
}

impl<'a> Call<'a> {
    /// A call to the wire `wire` describes, in `mode`, under no span.
    pub(crate) fn new(wire: &'a Descriptor<'a>, mode: Mode) -> Self {
        Self {
            wire,
            mode,
            span: tracing::Span::none(),
        }
    }

    /// Run the call under `span`: a unary call sends under it and a stream
    /// decodes under it.
    pub fn instrument(&mut self, span: tracing::Span) {
        self.span = span;
    }
}

/// One reply's state for an operation.
///
/// Each event is [`absorb`](Self::absorb)ed as the consumer takes it, so
/// the fold never runs ahead of what was delivered. The reply's end is what
/// [`finish`](Self::finish) needs: without the provider's end there is no
/// response.
pub trait Fold<Op: Operation> {
    /// Absorb one event the consumer is about to see. An error fails the
    /// reply.
    fn absorb(&mut self, event: &Op::Event) -> Result<(), ProviderError>;

    /// The response, from the events absorbed, the provider's end and what
    /// the driver learned about the reply.
    fn finish(self, end: Op::End, reply: Reply) -> Result<Op::Response, ProviderError>;
}

/// What the driver learned about a reply beyond its events.
#[derive(Debug, Clone, PartialEq)]
pub struct Reply {
    /// The provider descriptor name, for the response's `provider` field.
    pub provider: String,
    /// The reply's provider document (a completion's `raw`): the whole body
    /// when it is one JSON document, else what the wire's reassembler
    /// rebuilt or a free-event decoder recorded, else `Null`.
    pub raw: serde_json::Value,
    /// The provider's transport request id from the reply headers.
    pub provider_request_id: Option<String>,
}

/// Whether a decoder step left the reply open. Only [`Out::end`] makes an
/// [`Ended`].
#[derive(Debug)]
#[must_use]
pub enum Flow {
    /// The reply continues.
    More,
    /// The provider ended the reply; nothing is read after it.
    Ended(Ended),
}

/// Proof that a decoder ended its reply with [`Out::end`].
#[derive(Debug)]
pub struct Ended(());

pub(crate) use reply::Shared;

pub(crate) mod reply {
    use std::collections::VecDeque;

    use super::{Fold, Operation, ProviderError, Reply};
    use crate::streaming::Item;

    /// One reply's items and end, shared by the driver that writes them and
    /// the stream that takes them.
    pub struct Shared<Op: Operation> {
        pub(crate) fold: Op::Fold,
        pub(crate) items: VecDeque<Result<Item<Op::Event>, ProviderError>>,
        pub(crate) end: Option<Op::End>,
        pub(crate) raw: Option<serde_json::Value>,
        /// The response a relayed reply's origin already folded.
        pub(crate) response: Option<Op::Response>,
        /// What the transport reported: the reply's request id, its whole
        /// document and its request path.
        pub(crate) request_id: Option<String>,
        pub(crate) document: Option<serde_json::Value>,
        pub(crate) route: String,
    }

    impl<Op: Operation> Shared<Op> {
        pub(crate) fn new(fold: Op::Fold) -> Self {
            Self {
                fold,
                items: VecDeque::new(),
                end: None,
                raw: None,
                response: None,
                request_id: None,
                document: None,
                route: String::new(),
            }
        }

        /// Take the next item: an event is absorbed by the fold before it
        /// leaves, and an error is stamped with the reply's request id.
        pub(crate) fn take(&mut self) -> Option<Result<Item<Op::Event>, ProviderError>> {
            Some(match self.items.pop_front()? {
                Ok(Item::Event(event)) => self.fold.absorb(&event).map(|()| Item::Event(event)),
                Ok(unknown) => Ok(unknown),
                // An id an upstream constructor already attached wins: it
                // saw the reply.
                Err(error) => Err(error.with_provider_request_id(self.request_id.clone())),
            })
        }

        /// What the driver learned about the reply so far, from `provider`.
        pub(crate) fn reply(&self, provider: &str) -> Reply {
            reply(
                provider,
                self.document.clone().or_else(|| self.raw.clone()),
                self.request_id.clone(),
            )
        }

        /// Fold the absorbed reply with the provider's end into the
        /// response, or return the response a relay's origin folded. A
        /// reply the provider did not end is [`ProviderError::Truncated`].
        pub(crate) fn conclude(self, provider: &str) -> Result<Op::Response, ProviderError> {
            if let Some(response) = self.response {
                return Ok(response);
            }
            let reply = reply(provider, self.document.or(self.raw), self.request_id);
            let end = self.end.ok_or(ProviderError::Truncated)?;
            self.fold.finish(end, reply)
        }
    }

    /// The reply from `provider` whose document is `raw`: a whole body, or
    /// what the wire's reassembler rebuilt of a stream, or what the decoder
    /// of an operation with free events recorded.
    fn reply(provider: &str, raw: Option<serde_json::Value>, request_id: Option<String>) -> Reply {
        Reply {
            provider: provider.to_owned(),
            raw: raw.unwrap_or(serde_json::Value::Null),
            provider_request_id: request_id,
        }
    }

    /// What the reply's end closes before it is recorded.
    pub trait Closing<Op: Operation> {
        fn close(shared: &mut Shared<Op>);
    }
}

/// What a runtime accounts for about a model. Each field matters to the
/// operations that name it and is left at its default by the others.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Capabilities {
    /// What a completion provider composes.
    pub completion: crate::completion::ProviderCapabilities,
    /// The most documents one embedding or reranking request takes.
    pub max_documents: usize,
    /// The dimensionality of the returned vectors. Zero is unknown.
    pub ndims: usize,
    /// The embedding width the caller asked for, when they named one. `None`
    /// and zero disable the check that replies honour it.
    pub declared: Option<usize>,
}

impl Capabilities {
    /// A completion provider's capabilities.
    pub const fn completion(completion: crate::completion::ProviderCapabilities) -> Self {
        Self {
            completion,
            ..Self::embedding(0, 0)
        }
    }

    /// An embedding model's batch limit and width, with no width declared.
    pub const fn embedding(max_documents: usize, ndims: usize) -> Self {
        Self {
            completion: crate::completion::ProviderCapabilities::new(),
            max_documents,
            ndims,
            declared: None,
        }
    }

    /// A reranking model's batch limit.
    pub const fn rerank(max_documents: usize) -> Self {
        Self::embedding(max_documents, 0)
    }

    /// Record the width the caller named, when they named one.
    pub const fn declaring(mut self, declared: Option<usize>) -> Self {
        self.declared = declared;
        self
    }

    /// Returns [`ProviderError::MismatchedDimensions`] for the first width
    /// differing from a positive caller declaration. Otherwise succeeds.
    pub(crate) fn honour_declaration(
        &self,
        provider: &str,
        widths: impl IntoIterator<Item = usize>,
    ) -> Result<(), ProviderError> {
        // Zero is rig's sentinel for an unknown width, never a claim about
        // one: a model absent from every table this build knows resolves to
        // it, and treating that as a declaration would fail every reply.
        let Some(requested) = self.declared.filter(|declared| *declared > 0) else {
            return Ok(());
        };
        let Some(returned) = widths.into_iter().find(|width| *width != requested) else {
            return Ok(());
        };
        Err(ProviderError::MismatchedDimensions {
            provider: provider.to_owned(),
            requested,
            returned,
        })
    }
}

/// What a wire says about itself: plain data, read before every call.
#[derive(Debug, Clone)]
pub struct Descriptor<'a> {
    /// The model replay shapes a completion history for. Every completion
    /// wire names one; other operations leave it `None`.
    pub replay: Option<&'a dyn crate::completion::ReplayTarget>,
    /// The provider descriptor name (`"anthropic"`), as records and
    /// telemetry name it.
    pub name: &'a str,
    /// The model id the wire addresses, for telemetry. `None` for
    /// operations that address no model.
    pub model: Option<&'a str>,
    /// What a runtime accounts for.
    pub capabilities: Capabilities,
    /// The telemetry operation for a call in each mode, when the endpoint
    /// has a canonical name of its own (Gemini `generate_content`). The mode
    /// itself is recorded separately, as `gen_ai.request.stream`.
    pub telemetry: Option<fn(Mode) -> crate::telemetry::GenAiOperation>,
    /// The model facts a completion wire encodes with, which its replies are
    /// priced by. `None` answers from the built-in catalog.
    pub facts: Option<&'a crate::catalog::ModelFacts>,
    /// What the model kept from preparing its last request, which the
    /// driver lends [`Operation::prepare`]; `None` from the wire itself.
    pub prepared: Option<&'a Prepared>,
}

impl<'a> Descriptor<'a> {
    /// A wire named `name`, addressing no model, with default capabilities.
    pub fn new(name: &'a str) -> Self {
        Self {
            replay: None,
            name,
            model: None,
            capabilities: Capabilities::default(),
            telemetry: None,
            facts: None,
            prepared: None,
        }
    }

    /// The spec of the model the wire addresses, from its facts under its
    /// own name: the spec it was connected to, else the catalog's entry.
    /// `None` for a wire that addresses no model, or a model the catalog
    /// does not list.
    pub fn spec(&self) -> Option<&'a crate::catalog::ModelSpec> {
        let model = self.model?;
        match self.facts {
            Some(facts) => facts.for_model(self.name, model),
            None => crate::catalog::Catalog::builtin()
                .get_vendor(self.name, model)
                .map(|resolved| resolved.spec),
        }
    }

    /// The model replay shapes a completion history for.
    pub fn replay(mut self, target: &'a dyn crate::completion::ReplayTarget) -> Self {
        self.replay = Some(target);
        self
    }

    /// The model id the wire addresses, when it addresses one.
    pub fn model(mut self, model: impl Into<Option<&'a str>>) -> Self {
        self.model = model.into();
        self
    }

    /// What a runtime accounts for.
    pub fn capabilities(mut self, capabilities: Capabilities) -> Self {
        self.capabilities = capabilities;
        self
    }

    /// The model facts the wire encodes with.
    pub fn facts(mut self, facts: &'a crate::catalog::ModelFacts) -> Self {
        self.facts = Some(facts);
        self
    }

    /// The endpoint's own telemetry operation for each mode.
    pub fn telemetry(mut self, telemetry: fn(Mode) -> crate::telemetry::GenAiOperation) -> Self {
        self.telemetry = Some(telemetry);
        self
    }
}

/// One classified wire frame.
#[derive(Debug)]
pub enum WireEvent<T> {
    /// The frame carries a discriminator this client models and its payload
    /// decoded fully.
    Known(T),
    /// Valid JSON not recognized by this classifier.
    /// Drivers log structural metadata only and skip interpretation.
    Unknown {
        /// The unmodeled discriminator value.
        event_type: String,
        /// Full payload for raw passthrough, never warning logs. Debug is redacted.
        value: crate::streaming::UnknownPayload,
    },
    /// Invalid JSON or a recognized frame that failed typed decoding.
    /// Must not be demoted to `Unknown`.
    Corrupt(serde_json::Error),
}

impl<T> WireEvent<T> {
    /// Map the `Known` payload, preserving the classification.
    ///
    /// This is how an adapter layers a pure event-shape mapping on top of a
    /// classifier without restating the triage: `Unknown` and `Corrupt` pass
    /// through untouched, so policy stays with the driver.
    pub fn map<U>(self, f: impl FnOnce(T) -> U) -> WireEvent<U> {
        match self {
            Self::Known(event) => WireEvent::Known(f(event)),
            Self::Unknown { event_type, value } => WireEvent::Unknown { event_type, value },
            Self::Corrupt(error) => WireEvent::Corrupt(error),
        }
    }
}

/// Where a decoder writes one reply. The `'id` brand ties it to that one
/// reply: a writer cannot be kept past the decode step that received it.
///
/// ```compile_fail
/// use rig_core::operation::Completion;
/// use rig_core::wire::Out;
///
/// // A writer cannot be stashed to write into the reply later.
/// struct Stash(Option<Out<'static, Completion>>);
///
/// fn keep<'id>(stash: &mut Stash, out: Out<'id, Completion>) {
///     stash.0 = Some(out);
/// }
/// ```
pub struct Out<'id, Op: Operation> {
    shared: &'id std::sync::Mutex<Shared<Op>>,
    brand: std::marker::PhantomData<fn(&'id ()) -> &'id ()>,
}

impl<'id, Op: Operation> Out<'id, Op> {
    pub(crate) fn new(shared: &'id std::sync::Mutex<Shared<Op>>) -> Self {
        Self {
            shared,
            brand: std::marker::PhantomData,
        }
    }

    pub(crate) fn lock(&self) -> std::sync::MutexGuard<'id, Shared<Op>> {
        self.shared
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// End the reply with what the provider sent at its end. It consumes the
    /// writer: nothing is written after the end.
    pub fn end(self, end: Op::End) -> Flow {
        let mut shared = self.lock();
        <Op::Emit as reply::Closing<Op>>::close(&mut shared);
        shared.end = Some(end);
        Flow::Ended(Ended(()))
    }

    /// A payload the provider sent that this decoder does not model. It
    /// reaches the consumer as [`Item::Unknown`](crate::streaming::Item).
    pub fn unknown(&mut self, payload: crate::streaming::UnknownPayload) {
        self.lock()
            .items
            .push_back(Ok(crate::streaming::Item::Unknown(payload)));
    }
}

impl<Op: Operation<Emit = Free>> Out<'_, Op> {
    /// Record the reply's provider document, the response's `raw`. A whole
    /// JSON body the transport reported outranks it: a decoder whose reply
    /// is not one JSON document records it here.
    ///
    /// Only an operation whose decoders build their own events has this. A
    /// completion's `raw` is what its wire's
    /// [`Reassembler`](Wire::Reassembler) rebuilds, never what a decoder
    /// writes:
    ///
    /// ```compile_fail,E0599
    /// use rig_core::operation::Completion;
    /// use rig_core::wire::Out;
    ///
    /// fn record(out: &mut Out<'_, Completion>) {
    ///     out.raw(serde_json::json!({ "response_id": "x" }));
    /// }
    /// ```
    pub fn raw(&mut self, raw: serde_json::Value) {
        self.lock().raw = Some(raw);
    }

    /// One event of the reply.
    ///
    /// Only an operation whose decoders build their own events has this; a
    /// completion's events come from its part handles:
    ///
    /// ```compile_fail,E0599
    /// use rig_core::operation::Completion;
    /// use rig_core::streaming::StreamEvent;
    /// use rig_core::wire::Out;
    ///
    /// fn reinject(out: &mut Out<'_, Completion>, seen: &StreamEvent) {
    ///     out.event(seen.clone());
    /// }
    /// ```
    ///
    /// An event rebuilt from its serialized form is no different:
    ///
    /// ```compile_fail,E0599
    /// use rig_core::operation::Completion;
    /// use rig_core::streaming::StreamRecord;
    /// use rig_core::wire::Out;
    ///
    /// fn rebuild(out: &mut Out<'_, Completion>, recorded: serde_json::Value) {
    ///     let Ok(transcript) = StreamRecord::parse(recorded) else { return };
    ///     for event in transcript.events() {
    ///         out.event(event.clone());
    ///     }
    /// }
    /// ```
    pub fn event(&mut self, event: Op::Event) {
        self.lock()
            .items
            .push_back(Ok(crate::streaming::Item::Event(event)));
    }
}

/// Synchronous state machine for one reply. It classifies frames and
/// decodes known events into the reply's writer, without transport access
/// and without being told how the reply arrives. HTTP wires read
/// [`WireFrame`]s; other transports name their own frame type.
pub trait Decoder<'id, Op: Operation, Frame = WireFrame> {
    /// The wire's typed event, produced by this decoder's classifier.
    type Event;

    /// Decode and classify one frame. A JSON wire MUST delegate to a
    /// classifier in [`crate::providers::internal::wire`], so the
    /// decode-then-validate policy is not re-derived per provider.
    fn classify(&self, frame: Frame) -> WireEvent<Self::Event>;

    /// Write one `Known` event into the reply. An error ends the reply with
    /// it.
    fn decode(&mut self, event: Self::Event, out: Out<'id, Op>) -> Result<Flow, ProviderError>;

    /// The frames ran out without an end. A decoder that already saw the
    /// provider's end, in a form the provider sends before its last frame,
    /// ends the reply here; otherwise the reply is truncated.
    fn eof(&mut self, out: Out<'id, Op>) -> Result<Flow, ProviderError> {
        let _ = out;
        Err(ProviderError::Truncated)
    }
}

/// A provider endpoint: plain data that encodes requests and names a decoder
/// for each reply.
///
/// No transport, no future, no type parameter. Implementations are plain
/// data (`Clone + PartialEq + Debug + Serialize + Deserialize`, with
/// credentials held in [`Secret`]), so a host can store one in a scene, a
/// component, or a config file. `Clone` is a supertrait: the driver clones
/// the wire into every call's `'static` stream.
pub trait Wire: Clone + WasmCompatSend + WasmCompatSync + 'static {
    /// The operation this wire performs.
    type Op: Operation;
    /// What [`Self::encode`] produces for the transport to send.
    type Payload: WasmCompatSend + 'static;
    /// One unit of a reply, as the transport delivers it.
    type Frame: WasmCompatSend + 'static;
    /// The decoder's classified event, which the reassembler reads too.
    type Event: WasmCompatSend + 'static;
    /// The decoder for one of its replies, branded with that reply.
    type Decoder<'id>: Decoder<'id, Self::Op, Self::Frame, Event = Self::Event> + WasmCompatSend;
    /// What rebuilds a reply's provider document from its events when the
    /// transport reports no whole document, as for a stream. The driver
    /// hands it every known event before the decoder decodes it, and
    /// records what it finishes with as `raw`. A wire whose operation's
    /// decoders record `raw` themselves names [`document::Unreassembled`];
    /// a completion wire cannot, because it does not
    /// [`Serve`](document::Serves) completions.
    type Reassembler: document::Reassemble<Self::Event> + document::Serves<Self::Op>;

    /// What the wire says about itself.
    fn describe(&self) -> Descriptor<'_>;

    /// The request to send. Pure: it may read `self`, `request` and `mode`,
    /// and nothing else. A request that cannot be built is an
    /// [`EncodeError`], which always reports as a request failure.
    fn encode(&self, request: Request<Self>, mode: Mode) -> Result<Self::Payload, EncodeError>;

    /// A fresh decoder for one reply.
    fn decoder<'id>(&self) -> Self::Decoder<'id>;

    /// A fresh reassembler for one reply. A wire that picks its API per
    /// reply builds the matching one.
    fn reassembler(&self) -> Self::Reassembler {
        Self::Reassembler::default()
    }
}

/// One reply of `wire` to `request`, decoded from frames already in hand as
/// a live call in `mode` would decode them: through the wire's decoder,
/// reassembler and fold, with no transport. A reply that arrives some other
/// way, such as a batch result's body ([`WireFrame::Bytes`], unary) or a
/// recorded stream, becomes an ordinary response. `raw` is what the wire's
/// reassembler rebuilds from the frames, and the request is the one sent,
/// as [`Wire::encode`] took it.
///
/// ```
/// use rig_core::completion::{CompletionRequest, CompletionResponse};
/// use rig_core::error::ProviderError;
/// use rig_core::providers::anthropic::wire::Messages;
/// use rig_core::wire::{Mode, WireFrame, decode_reply};
///
/// /// A batch result's body, as the reply to the `request` it answers.
/// fn batch_reply(
///     wire: &Messages,
///     request: &CompletionRequest,
///     body: Vec<u8>,
/// ) -> Result<CompletionResponse, ProviderError> {
///     decode_reply(wire, request, Mode::Unary, [WireFrame::Bytes(body)])
/// }
/// ```
///
/// # Errors
///
/// What the reply fails with when called live: a frame the decoder rejects,
/// a provider error in the reply, or [`ProviderError::Truncated`] when the
/// frames end before the provider ended the reply.
pub fn decode_reply<W: Wire>(
    wire: &W,
    request: &Request<W>,
    mode: Mode,
    frames: impl IntoIterator<Item = W::Frame>,
) -> Result<Response<W>, ProviderError> {
    crate::driver::decode_reply(wire, request, mode, frames, serde_json::Value::Null)
}

/// The decoder of a reply that is one JSON document: the document is the
/// operation's end, and is kept as the reply's `raw`.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Json;

impl<'id, Op> Decoder<'id, Op, WireFrame> for Json
where
    Op: Operation<Emit = Free>,
    Op::End: serde::de::DeserializeOwned,
{
    type Event = (Op::End, serde_json::Value);

    fn classify(&self, frame: WireFrame) -> WireEvent<Self::Event> {
        let document = match serde_json::from_str::<serde_json::Value>(&frame.as_str()) {
            Ok(document) => document,
            Err(error) => return WireEvent::Corrupt(error),
        };
        match <Op::End as serde::Deserialize>::deserialize(&document) {
            Ok(end) => WireEvent::Known((end, document)),
            Err(error) => WireEvent::Corrupt(error),
        }
    }

    fn decode(
        &mut self,
        (end, document): Self::Event,
        mut out: Out<'id, Op>,
    ) -> Result<Flow, ProviderError> {
        out.raw(document);
        Ok(out.end(end))
    }
}

/// The decoder of a reply that is one document: `classify` reads the
/// document, and `end` makes it the operation's end.
pub struct DocumentDecoder<Op: Operation, Event> {
    classify: fn(&str) -> WireEvent<Event>,
    end: fn(Event) -> Result<Op::End, ProviderError>,
}

impl<Op: Operation, Event> DocumentDecoder<Op, Event> {
    /// The decoder that reads a reply with `classify` and ends it with `end`.
    pub fn new(
        classify: fn(&str) -> WireEvent<Event>,
        end: fn(Event) -> Result<Op::End, ProviderError>,
    ) -> Self {
        Self { classify, end }
    }
}

impl<'id, Op: Operation, Event> Decoder<'id, Op> for DocumentDecoder<Op, Event> {
    type Event = Event;

    fn classify(&self, frame: WireFrame) -> WireEvent<Event> {
        (self.classify)(&frame.as_str())
    }

    fn decode(&mut self, event: Event, out: Out<'id, Op>) -> Result<Flow, ProviderError> {
        Ok(out.end((self.end)(event)?))
    }
}

/// A wire's request type.
pub type Request<W> = <<W as Wire>::Op as Operation>::Request;
/// A wire's response type.
pub type Response<W> = <<W as Wire>::Op as Operation>::Response;
/// A wire's event type.
pub type Event<W> = <<W as Wire>::Op as Operation>::Event;

#[cfg(test)]
mod tests;