wyrd-client 0.3.0

//! Terminal-safe Oracle query client and bounded result collection.

use std::collections::VecDeque;
use std::io::Cursor;
use std::pin::Pin;

use crate::WyrdClient;
use crate::error::WyrdClientError;
use arrow::datatypes::SchemaRef;
use arrow::ipc::reader::StreamReader;
use arrow::record_batch::RecordBatch;
use futures_util::{Stream, StreamExt};
use wyrd_queue::WyrdQueueError;
use wyrd_spec::error::WyrdError;
use wyrd_spec::request_id::RequestId;
use wyrd_spec::vala::api::{
    BifrostQueryRequest, BifrostTableDescription, BifrostTableEntry, CancelRunningQueryResponse,
    ListRunningQueriesResponse, QueryStreamFrame, QueryTerminalFrame, QueryTerminalOutcome,
    RunningQuerySummary,
};
use wyrd_spec::vala::error::BifrostError;
use wyrd_tonic::frame_codec::FrameDecoder;
use wyrd_tonic::query_conversion::QueryStreamConverter;
// Aliased: the wire message shares its name with the domain frame it converts to.
use wyrd_tonic::wyrd::v1::QueryStreamFrame as WireQueryStreamFrame;
use wyrd_tonic::wyrd::v1::query_stream_frame::Frame as WireFrame;

/// Maximum encoded protobuf frame accepted by the public query client.
const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;

/// Boxed response-body stream retained so dropping a query cancels body consumption.
type ResponseBytes = Pin<Box<dyn Stream<Item = Result<bytes::Bytes, reqwest::Error>> + Send>>;

/// Errors returned while querying Oracle through the public client contract.
#[derive(Debug, thiserror::Error)]
pub enum BifrostClientError {
    /// The authenticated HTTP transport rejected or could not send a request.
    #[error("query transport failed: {0}")]
    Transport(#[from] wyrd_spec::error::WyrdError),
    /// The length-delimited protobuf stream violated framing or conversion rules.
    #[error("query stream protocol failed: {0}")]
    Protocol(String),
    /// Arrow IPC schema or batch bytes were invalid or inconsistent.
    #[error("query Arrow IPC decode failed: {0}")]
    Arrow(String),
    /// EOF arrived before the required terminal frame.
    #[error("query stream ended before its required terminal frame")]
    IncompleteQueryStream,
    /// Oracle reported a validated late failure after zero or more batches.
    #[error("query terminal reported failure")]
    FailedTerminal {
        /// Validated terminal metadata retained for diagnostics.
        terminal: QueryTerminalFrame,
    },
    /// The caller's bounded collection limits were exceeded.
    #[error("query result exceeds configured bounds")]
    ResultTooLarge,
    /// A write was attempted with no table bound as the active target.
    #[error("no active Bifrost table is bound for writes")]
    NoActiveTable,
    /// The client-tier producer queue refused, timed out, or failed a write.
    ///
    /// Carried verbatim so the public projection can name the queue's own
    /// catalog variant — notably `WYRD_CLIENT_429_QUEUE_FULL` — instead of
    /// reporting saturation as an internal error to every language SDK.
    #[error("bifrost write failed: {0}")]
    Queue(#[from] WyrdQueueError),
    /// One completed result row did not deserialize into the caller's type.
    ///
    /// Client-tier `WYRD_CLIENT_422_ROW_DESERIALIZATION`: the query itself
    /// succeeded and its terminal validated, so this blames the local typed
    /// projection rather than the server, and never reports the rows that did
    /// convert as a partial success.
    #[error("query row deserialization failed: {0}")]
    RowDeserialization(String),
    /// A client-tier configuration, credential, or transport failure.
    ///
    /// Carried verbatim for the same reason [`BifrostClientError::Queue`] is: the
    /// public projection names the matching `WYRD_CLIENT_*` catalog variant so
    /// an unresolvable credential chain stays a 401 in every language SDK.
    #[error("{0}")]
    Client(#[from] WyrdClientError),
}
impl From<wyrd_spec::vala::BifrostError> for BifrostClientError {
    /// Carry a refusal from the shared column-type mappings as the queue
    /// error it has always surfaced as, so its stable code is unchanged.
    fn from(error: wyrd_spec::vala::BifrostError) -> Self {
        Self::Queue(error.into())
    }
}

impl BifrostClientError {
    /// Returns terminal metadata when a validated failed terminal caused this error.
    #[must_use]
    pub fn terminal(&self) -> Option<&QueryTerminalFrame> {
        match self {
            Self::FailedTerminal { terminal } => Some(terminal),
            Self::Transport(_)
            | Self::Protocol(_)
            | Self::Arrow(_)
            | Self::RowDeserialization(_)
            | Self::IncompleteQueryStream
            | Self::ResultTooLarge
            | Self::NoActiveTable
            | Self::Queue(_)
            | Self::Client(_) => None,
        }
    }
}

impl From<BifrostClientError> for WyrdError {
    /// Project one owned SDK error onto the shared derive-backed catalog.
    fn from(error: BifrostClientError) -> Self {
        Self::from(&error)
    }
}

impl From<&BifrostClientError> for WyrdError {
    /// Project one borrowed SDK error onto the shared derive-backed catalog.
    ///
    /// This is the crate's only public error projection: every stable code,
    /// status, title, and remediation the SDK reports comes from the catalog
    /// variant named here, so no language boundary keeps a second copy. A
    /// transport failure is already a catalog error and is returned untouched.
    ///
    /// Framing and Arrow decode strings stay on the Rust-side `Display` rather
    /// than the public projection: the catalog owns the scrubbed public text,
    /// and a raw decoder string is a local diagnostic, not a stable contract.
    fn from(error: &BifrostClientError) -> Self {
        match error {
            BifrostClientError::Transport(inner) => inner.clone(),
            BifrostClientError::Protocol(_) | BifrostClientError::Arrow(_) => Self::Vala {
                error: BifrostError::QueryStreamProtocol,
            },
            BifrostClientError::IncompleteQueryStream => Self::Vala {
                error: BifrostError::QueryStreamIncomplete,
            },
            BifrostClientError::FailedTerminal { terminal } => terminal.error.as_deref().map_or(
                Self::Vala {
                    error: BifrostError::QueryExecutionFailed,
                },
                crate::error::from_problem,
            ),
            BifrostClientError::ResultTooLarge => Self::Vala {
                error: BifrostError::QueryResultTooLarge,
            },
            BifrostClientError::NoActiveTable => Self::Vala {
                error: BifrostError::NoActiveTable,
            },
            BifrostClientError::RowDeserialization(detail) => Self::ClientRowDeserialization {
                message: detail.clone(),
                details: serde_json::json!({}),
            },
            BifrostClientError::Queue(inner) => inner.into(),
            BifrostClientError::Client(inner) => inner.into(),
        }
    }
}

/// Rust-owned query handle that reuses an assembled client's auth and HTTP pools.
#[derive(Debug, Clone)]
pub struct QueryClient {
    /// Cheap client clone sharing the original authentication state and connection pools.
    client: WyrdClient,
}

impl QueryClient {
    /// Creates a query handle sharing the supplied Wyrd client's transport state.
    #[must_use]
    pub(crate) fn new(client: &WyrdClient) -> Self {
        Self {
            client: client.clone(),
        }
    }

    /// The authenticated client this handle shares its transport with.
    ///
    /// Exposed so [`crate::bifrost::Bifrost`] issues its catalog calls — register and
    /// describe — over the same auth and connection pools its queries use,
    /// instead of assembling a second transport for the same credential.
    #[must_use]
    pub fn client(&self) -> &WyrdClient {
        &self.client
    }

    /// Starts one authenticated terminal-safe Oracle query.
    ///
    /// The returned stream owns the HTTP response body. Dropping it stops body
    /// consumption, which propagates cancellation through the transport.
    ///
    /// # Arguments
    /// * `request` - The complete query request: SQL, bind values, and deadline.
    ///
    /// # Errors
    ///
    /// Returns a contract error before IO when the request is invalid —
    /// `WYRD_VALA_400_QUERY_INVALID_SQL` for empty SQL, a non-finite bind
    /// value, or an out-of-range deadline — or a transport error when
    /// authentication or the HTTP request fails.
    ///
    /// # Cancellation
    ///
    /// Cancellation before the response arrives abandons the request future.
    /// After return, dropping the result stream cancels response-body consumption.
    pub async fn query(
        &self,
        request: &BifrostQueryRequest,
    ) -> Result<QueryResultStream, BifrostClientError> {
        request
            .validate()
            .map_err(|error| BifrostClientError::Transport(error.into()))?;
        let request_id = RequestId::now_v7();
        let response = self
            .client
            .http
            .request_json_stream_with_id(reqwest::Method::POST, "/v1/query", request, &request_id)
            .await?;
        let deadline_ms = response
            .headers()
            .get("x-wyrd-query-deadline-ms")
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.parse::<i64>().ok())
            .filter(|deadline| *deadline >= 0)
            .ok_or_else(|| {
                BifrostClientError::Protocol(
                    "query response omits a valid x-wyrd-query-deadline-ms".to_owned(),
                )
            })?;
        Ok(QueryResultStream::new(
            RawQueryStream::new(response.bytes_stream()),
            request_id,
            self.clone(),
            deadline_ms,
        ))
    }

    /// Lists active queries visible to the authenticated tenant.
    ///
    /// Cancelling this future abandons the pending HTTP request without
    /// creating client-owned lifecycle state.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, audit, availability, or protocol errors.
    pub async fn running(&self) -> Result<Vec<RunningQuerySummary>, BifrostClientError> {
        let response: ListRunningQueriesResponse = self
            .client
            .http
            .request_json::<(), _>(reqwest::Method::GET, "/v1/query/running", None)
            .await?;
        Ok(response.queries)
    }

    /// Gets one active query visible to the authenticated tenant.
    ///
    /// Cancelling this future abandons the pending HTTP request without
    /// changing the active query.
    ///
    /// # Arguments
    /// * `request_id` - The id of the active query to look up.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, not-found, availability, or protocol errors.
    pub async fn status(
        &self,
        request_id: &RequestId,
    ) -> Result<RunningQuerySummary, BifrostClientError> {
        self.client
            .http
            .request_json::<(), _>(
                reqwest::Method::GET,
                &format!("/v1/query/{request_id}"),
                None,
            )
            .await
            .map_err(Into::into)
    }

    /// Requests server-side cancellation without closing a local response stream.
    ///
    /// Once the server accepts cancellation, cancelling this future does not
    /// reverse the server-side lifecycle transition.
    ///
    /// # Arguments
    /// * `request_id` - The id of the active query to cancel.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, not-found, availability, or protocol errors.
    pub async fn cancel(
        &self,
        request_id: &RequestId,
    ) -> Result<CancelRunningQueryResponse, BifrostClientError> {
        self.client
            .http
            .request_json::<(), _>(
                reqwest::Method::DELETE,
                &format!("/v1/query/{request_id}"),
                None,
            )
            .await
            .map_err(Into::into)
    }

    /// Lists the compact entries of every table the caller may read.
    ///
    /// Reads `GET /v1/bifrost/tables`: the server returns the visible built-in
    /// and tenant tables without their schemas, so a caller describes one
    /// table with [`Self::describe_table`] only when it needs that schema.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, availability, or
    /// protocol errors.
    pub async fn list_tables(&self) -> Result<Vec<BifrostTableEntry>, BifrostClientError> {
        self.client
            .http
            .request_json::<(), _>(reqwest::Method::GET, "/v1/bifrost/tables", None)
            .await
            .map_err(Into::into)
    }

    /// Describes one registered table's stored physical schema.
    ///
    /// The description is the server's own projection of the stored schema: the
    /// user fields a caller writes, the correlation fields the write path
    /// resolves, the managed candidates it may supply, and — for a canonical
    /// signal table — the canonical physical fingerprint. A client builds its
    /// insertable Arrow schema from this rather than from a local table
    /// definition, so no client owns a second copy of the physical contract.
    ///
    /// # Arguments
    /// * `namespace` - The table's namespace.
    /// * `name` - The table's name within `namespace`.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, not-found, availability,
    /// or protocol errors.
    ///
    /// # Cancellation
    ///
    /// Cancelling the future abandons the pending request; describe reads
    /// nothing durable, so an abandoned call leaves no server state behind.
    pub async fn describe_table(
        &self,
        namespace: &str,
        name: &str,
    ) -> Result<BifrostTableDescription, BifrostClientError> {
        self.client
            .http
            .request_json::<(), _>(
                reqwest::Method::GET,
                &format!("/v1/bifrost/tables/{namespace}/{name}"),
                None,
            )
            .await
            .map_err(Into::into)
    }

    /// Collects a query while enforcing explicit row and encoded-byte limits.
    ///
    /// # Arguments
    /// * `request` - The complete query request: SQL, bind values, and deadline.
    /// * `limits` - The row and encoded-byte ceilings the collected result must fit.
    ///
    /// # Errors
    ///
    /// Returns a protocol, Arrow, transport, failed-terminal, incomplete-stream,
    /// or bounds error. Exceeding a bound drops the live stream and never returns
    /// truncated success.
    ///
    /// # Cancellation
    ///
    /// Cancelling the future drops the response stream and its HTTP body.
    #[cfg(feature = "internal")]
    pub async fn collect_bounded(
        &self,
        request: &BifrostQueryRequest,
        limits: CollectedQueryLimits,
    ) -> Result<CollectedQueryResult, BifrostClientError> {
        self.query(request).await?.collect_bounded(limits).await
    }
}

/// Raw decoded protocol stream before Arrow IPC projection.
pub struct RawQueryStream {
    /// Incremental authenticated HTTP response body.
    body: ResponseBytes,
    /// Bounded protobuf length-delimited decoder.
    decoder: FrameDecoder,
    /// Complete protobuf frames from the most recent body chunk, kept with
    /// their Arrow fragments still encoded.
    ///
    /// Conversion and Arrow decoding run only when a frame is popped for
    /// delivery, so however many frames one transport chunk coalesces, at most
    /// one decoded `RecordBatch` exists in this stream's pending path.
    pending: VecDeque<WireQueryStreamFrame>,
    /// Shared wire-to-domain ordering and terminal validator.
    converter: QueryStreamConverter,
    /// The one Arrow IPC decoder for this query's single split stream.
    ipc: QueryIpcDecoder,
    /// Validated terminal retained after it is yielded.
    terminal: Option<QueryTerminalFrame>,
    /// Exact length-delimited response bytes received from the HTTP body.
    received_bytes: usize,
}

impl RawQueryStream {
    /// Creates a raw stream over arbitrary response chunks.
    fn new<S>(body: S) -> Self
    where
        S: Stream<Item = Result<bytes::Bytes, reqwest::Error>> + Send + 'static,
    {
        Self {
            body: Box::pin(body),
            decoder: FrameDecoder::new(MAX_FRAME_BYTES),
            pending: VecDeque::new(),
            converter: QueryStreamConverter::new(),
            ipc: QueryIpcDecoder::new(),
            terminal: None,
            received_bytes: 0,
        }
    }

    /// Returns the next validated frame, or `None` only after a valid terminal.
    ///
    /// # Errors
    ///
    /// Returns a typed protocol or Arrow error for malformed, truncated,
    /// out-of-order, duplicate, post-terminal, or invalid-terminal streams.
    /// EOF before terminal returns [`BifrostClientError::IncompleteQueryStream`].
    ///
    /// # Cancellation
    ///
    /// Cancelling this operation preserves decoder state. Dropping the stream
    /// drops the HTTP response body and stops further reads.
    #[cfg(test)]
    pub async fn next_frame(&mut self) -> Result<Option<QueryStreamFrame>, BifrostClientError> {
        Ok(self
            .next_decoded_frame()
            .await?
            .map(|(frame, _batch)| frame))
    }

    /// Returns the next validated frame with the batch its fragment decoded to.
    ///
    /// Each Arrow fragment is decoded exactly once, here, because the stream is
    /// stateful: a batch consumed twice would advance the shared decoder twice
    /// and desynchronise every later fragment. The Arrow projection therefore
    /// takes the already-decoded batch rather than re-reading the bytes.
    ///
    /// # Errors
    ///
    /// Returns a typed protocol or Arrow error for malformed, truncated,
    /// out-of-order, duplicate, post-terminal, or invalid-terminal streams.
    /// EOF before terminal returns [`BifrostClientError::IncompleteQueryStream`].
    async fn next_decoded_frame(
        &mut self,
    ) -> Result<Option<(QueryStreamFrame, Option<RecordBatch>)>, BifrostClientError> {
        loop {
            if let Some(frame) = self.pending.pop_front() {
                return self.decode_frame(frame).map(Some);
            }
            match self.body.next().await {
                Some(Ok(bytes)) => {
                    self.received_bytes =
                        self.received_bytes
                            .checked_add(bytes.len())
                            .ok_or_else(|| {
                                BifrostClientError::Protocol(
                                    "encoded response byte count overflow".to_owned(),
                                )
                            })?;
                    let frames = self
                        .decoder
                        .push::<WireQueryStreamFrame>(&bytes)
                        .map_err(|error| BifrostClientError::Protocol(error.to_string()))?;
                    self.pending.extend(frames);
                }
                Some(Err(error)) => {
                    return Err(query_body_transport_error(error));
                }
                None => {
                    self.decoder
                        .finish()
                        .map_err(|error| BifrostClientError::Protocol(error.to_string()))?;
                    if self.terminal.is_none() {
                        return Err(BifrostClientError::IncompleteQueryStream);
                    }
                    return Ok(None);
                }
            }
        }
    }

    /// Converts one wire frame at delivery time, decoding its Arrow fragment.
    ///
    /// The Arrow fragment is consumed before conversion so the converter's row
    /// accounting and the stateful decoder's stream position advance from the
    /// same bytes exactly once. A non-failed terminal closes the Arrow stream
    /// with its end-of-stream bytes, and every terminal is retained as this
    /// raw stream's provisional terminal.
    ///
    /// # Errors
    ///
    /// Returns a typed Arrow error for an undecodable fragment and a protocol
    /// error for out-of-order, duplicate, post-terminal, or invalid frames.
    fn decode_frame(
        &mut self,
        frame: WireQueryStreamFrame,
    ) -> Result<(QueryStreamFrame, Option<RecordBatch>), BifrostClientError> {
        let decoded = match frame.frame.as_ref() {
            Some(WireFrame::Batch(batch)) => Some(self.ipc.accept_batch(&batch.arrow_ipc_batch)?),
            Some(WireFrame::Schema(schema)) => {
                self.ipc.accept_schema(&schema.arrow_ipc_schema)?;
                None
            }
            Some(WireFrame::Terminal(_)) | None => None,
        };
        let batch_rows = decoded
            .as_ref()
            .map(|batch| {
                u64::try_from(batch.num_rows()).map_err(|_| {
                    BifrostClientError::Protocol("row count does not fit u64".to_owned())
                })
            })
            .transpose()?;
        let frame = self
            .converter
            .convert(frame, batch_rows)
            .map_err(|error| BifrostClientError::Protocol(error.to_string()))?;
        if let QueryStreamFrame::Terminal(terminal) = &frame {
            // The terminal's own validation already refused a success that
            // omits its end-of-stream; closing the Arrow stream here proves the
            // bytes it carries are the real end of this decoder's stream.
            if terminal.outcome != QueryTerminalOutcome::Failed {
                self.ipc.accept_eos(&terminal.arrow_ipc_eos)?;
            }
            self.terminal = Some(terminal.clone());
        }
        Ok((frame, decoded))
    }

    /// Returns the validated terminal metadata, if it has been observed.
    #[must_use]
    pub fn terminal(&self) -> Option<&QueryTerminalFrame> {
        self.terminal.as_ref()
    }

    /// Returns exact length-delimited response bytes received so far.
    #[must_use]
    fn received_bytes(&self) -> usize {
        self.received_bytes
    }

    /// Returns the stream schema once its initial fragment was accepted.
    #[must_use]
    pub fn arrow_schema(&self) -> Option<&SchemaRef> {
        self.ipc.schema.as_ref()
    }

    /// Reports whether the stream's Arrow IPC end-of-stream was accepted.
    #[must_use]
    pub const fn arrow_ipc_closed(&self) -> bool {
        self.ipc.eos_accepted()
    }

    /// Returns the largest single Arrow fragment this stream held while decoding.
    ///
    /// This is the client half of the bounded-memory contract: it must never
    /// exceed the largest individual fragment, because fragments are decoded
    /// and released one at a time rather than accumulated.
    #[must_use]
    pub const fn peak_pending_frame_bytes(&self) -> usize {
        self.ipc.peak_pending_frame_bytes()
    }

    /// Returns total Arrow IPC bytes consumed across every fragment.
    #[must_use]
    pub const fn arrow_ipc_bytes(&self) -> usize {
        self.ipc.total_fragment_bytes()
    }
}

/// Maps a Reqwest body failure without confusing transport with framing.
///
/// The stable projection retains machine-readable phase and Reqwest
/// classifications. The structured diagnostic keeps the original error as the
/// tracing source for operators while the public detail remains scrubbed.
fn query_body_transport_error(error: reqwest::Error) -> BifrostClientError {
    let details = serde_json::json!({
        "transport": "http",
        "phase": "response_body",
        "timeout": error.is_timeout(),
        "connect": error.is_connect(),
        "body": error.is_body(),
        "decode": error.is_decode(),
    });
    tracing::error!(
        error = ?error,
        transport = "http",
        phase = "response_body",
        timeout = error.is_timeout(),
        connect = error.is_connect(),
        body = error.is_body(),
        decode = error.is_decode(),
        "Oracle query response body failed"
    );
    BifrostClientError::Transport(WyrdError::UpstreamFailure {
        message: "Oracle query response body failed".to_owned(),
        details,
    })
}

/// What one incomplete stream still owes the server before the caller walks away.
///
/// A stream that reached its terminal owes nothing. Anything else left a query
/// running on a server that has no other way to learn the caller is gone, so
/// exactly one settlement is owed — and which one depends entirely on whether
/// the body can still be trusted.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StreamSettlement {
    /// The body is intact and can be drained to its real terminal.
    Healthy,
    /// The body failed to decode or transport and must never be read again.
    Broken,
    /// A terminal was observed, or settlement already ran exactly once.
    Settled,
}

/// Stable code proving one query is no longer running and needs no settlement.
#[cfg(feature = "internal")]
const RUNNING_QUERY_RETIRED_CODE: &str = "WYRD_VALA_404_RUNNING_QUERY_NOT_FOUND";

/// Interval between status polls while proving a broken stream was cleaned up.
#[cfg(feature = "internal")]
const SETTLEMENT_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100);

/// Arrow-projecting query stream that preserves terminal metadata.
pub struct QueryResultStream {
    /// Canonical server-visible request identity available before body polling.
    request_id: RequestId,
    /// Client reused for the one cancellation and any status proof this stream
    /// owes; only internal settlement reads it.
    #[cfg(feature = "internal")]
    client: QueryClient,
    /// Server-pinned absolute deadline bounding every settlement wait.
    deadline_ms: i64,
    /// Client transport timeout bounding the close after a received terminal.
    ///
    /// The query deadline bounds terminal production; once a terminal has
    /// arrived, the EOF behind it is the connection closing, which this
    /// configured transport bound governs instead.
    close_timeout: std::time::Duration,
    /// Settlement this stream still owes, or [`StreamSettlement::Settled`].
    settlement: StreamSettlement,
    /// Validated logical-frame stream.
    raw: RawQueryStream,
    /// Schema established by the unique initial schema frame.
    schema: Option<SchemaRef>,
    /// Terminal retained after successful, degraded, or failed completion.
    terminal: Option<QueryTerminalFrame>,
    /// Rows yielded to the caller before the current state.
    emitted_rows: u64,
    /// Length-delimited response bytes received so far.
    encoded_bytes: usize,
}

impl QueryResultStream {
    /// Constructs the Arrow projection over a raw protocol stream.
    ///
    /// The close bound after a terminal is read from `client`'s configured
    /// transport timeout here, so every stream carries the same value its
    /// transport applies to retried requests.
    fn new(
        raw: RawQueryStream,
        request_id: RequestId,
        client: QueryClient,
        deadline_ms: i64,
    ) -> Self {
        let close_timeout = client.client.http.request_timeout();
        // Without `internal` nothing settles a stream, so the client it would
        // settle with is not retained.
        #[cfg(not(feature = "internal"))]
        let _ = client;
        Self {
            request_id,
            #[cfg(feature = "internal")]
            client,
            deadline_ms,
            close_timeout,
            settlement: StreamSettlement::Healthy,
            raw,
            schema: None,
            terminal: None,
            emitted_rows: 0,
            encoded_bytes: 0,
        }
    }

    /// Returns the canonical request identity used by server lifecycle controls.
    #[must_use]
    pub const fn request_id(&self) -> &RequestId {
        &self.request_id
    }

    /// Decodes and returns the next Arrow batch, or `None` after valid completion.
    ///
    /// # Errors
    ///
    /// Returns protocol and Arrow errors, including a schema mismatch, multiple
    /// record batches inside one batch frame, missing terminal, failed terminal,
    /// or mismatched terminal row count. A failed terminal is retained, and its
    /// error returned, only after the body reaches clean EOF.
    ///
    /// # Cancellation
    ///
    /// Cancelling preserves the stream for a later call. Dropping the stream
    /// cancels response-body consumption.
    pub async fn next_batch(&mut self) -> Result<Option<RecordBatch>, BifrostClientError> {
        if self.terminal.is_none()
            && self.settlement != StreamSettlement::Broken
            && let Some(terminal) = self.raw.terminal().cloned()
        {
            if let Err(error) = terminal.validate_emitted_rows(self.emitted_rows) {
                return Err(self.mark_broken(BifrostClientError::Protocol(error.to_string())));
            }
            return self.finish_at_clean_eof(terminal).await;
        }
        loop {
            let frame = self.raw.next_decoded_frame().await;
            self.encoded_bytes = self.raw.received_bytes();
            let frame = match frame {
                Ok(frame) => frame,
                Err(error) => return Err(self.mark_broken(error)),
            };
            let Some((frame, decoded)) = frame else {
                return Ok(None);
            };
            match frame {
                QueryStreamFrame::Schema(_) => {
                    self.schema = self.raw.arrow_schema().cloned();
                }
                QueryStreamFrame::Batch(_) => {
                    let Some(decoded) = decoded else {
                        return Err(self.mark_broken(BifrostClientError::Arrow(
                            "batch frame contains no record batch".to_owned(),
                        )));
                    };
                    let Ok(rows) = u64::try_from(decoded.num_rows()) else {
                        return Err(self.mark_broken(BifrostClientError::Protocol(
                            "row count does not fit u64".to_owned(),
                        )));
                    };
                    let Some(emitted) = self.emitted_rows.checked_add(rows) else {
                        return Err(self.mark_broken(BifrostClientError::Protocol(
                            "row count overflow".to_owned(),
                        )));
                    };
                    self.emitted_rows = emitted;
                    return Ok(Some(decoded));
                }
                QueryStreamFrame::Terminal(terminal) => {
                    if let Err(error) = terminal.validate_emitted_rows(self.emitted_rows) {
                        return Err(
                            self.mark_broken(BifrostClientError::Protocol(error.to_string()))
                        );
                    }
                    return self.finish_at_clean_eof(terminal).await;
                }
            }
        }
    }

    /// Marks the body untrusted when `error` means it can no longer be read.
    ///
    /// A decode, protocol, transport, or incomplete-body failure leaves the
    /// response stream in an unknown position, so settlement must prove cleanup
    /// through the query's own status route instead of draining it. The error
    /// is returned unchanged: this marker never replaces what the caller sees.
    fn mark_broken(&mut self, error: BifrostClientError) -> BifrostClientError {
        if matches!(
            error,
            BifrostClientError::Protocol(_)
                | BifrostClientError::Arrow(_)
                | BifrostClientError::Transport(_)
                | BifrostClientError::IncompleteQueryStream
        ) {
            self.settlement = StreamSettlement::Broken;
        }
        error
    }

    /// Accepts a validated terminal only after the body reaches EOF.
    ///
    /// A terminal is a claim about a stream that has not ended yet. Retaining it
    /// before the body closes would let a duplicate terminal, a late schema, or
    /// a trailing batch arrive behind a result the caller already believes is
    /// complete. The raw stream retains the provisional terminal across cancelled
    /// reads. Only clean EOF promotes it to this stream's public result. A
    /// failed terminal follows the same proof, so a duplicate terminal or any
    /// other post-failure frame is a broken stream, not a validated failure.
    ///
    /// The two waits have different owners. The server-pinned query deadline
    /// bounds terminal production, and a timeout terminal is produced at or
    /// after that deadline by design, so the deadline has done its job once a
    /// terminal arrives. The read for the EOF behind it is the connection
    /// closing, which the client's configured transport timeout bounds; each
    /// call, including one resumed after cancellation, waits at most that long.
    ///
    /// # Errors
    /// Returns [`BifrostClientError::FailedTerminal`] when a failed terminal is
    /// followed by clean EOF, [`BifrostClientError::Protocol`] when any frame follows the terminal,
    /// the raw stream's own error unchanged when the body fails, and
    /// [`BifrostClientError::IncompleteQueryStream`] when the transport timeout
    /// passes before the body closes. Every one of those marks the body broken first.
    async fn finish_at_clean_eof(
        &mut self,
        terminal: QueryTerminalFrame,
    ) -> Result<Option<RecordBatch>, BifrostClientError> {
        let observed =
            tokio::time::timeout(self.close_timeout, self.raw.next_decoded_frame()).await;
        self.encoded_bytes = self.raw.received_bytes();
        match observed {
            Ok(Ok(None)) => {
                self.terminal = Some(terminal.clone());
                // A terminal followed by clean EOF is the server's own
                // settlement. Nothing is owed.
                self.settlement = StreamSettlement::Settled;
                if terminal.outcome == QueryTerminalOutcome::Failed {
                    return Err(BifrostClientError::FailedTerminal { terminal });
                }
                Ok(None)
            }
            Ok(Ok(Some(_))) => Err(self.mark_broken(BifrostClientError::Protocol(
                "a frame followed the query terminal".to_owned(),
            ))),
            Ok(Err(error)) => Err(self.mark_broken(error)),
            Err(_) => Err(self.mark_broken(BifrostClientError::IncompleteQueryStream)),
        }
    }

    /// Collects this stream within explicit result ceilings.
    ///
    /// The returned result retains the authoritative schema decoded from the
    /// initial schema frame, including when the query produces no batches. A
    /// successful terminal without that schema is rejected as incomplete.
    ///
    /// # Arguments
    /// * `limits` - The row and encoded-byte ceilings the collected result must fit.
    ///
    /// # Errors
    ///
    /// Returns a stream error or [`BifrostClientError::ResultTooLarge`] before
    /// retaining a batch that would exceed either ceiling.
    ///
    /// # Cancellation
    ///
    /// Cancelling drops this owned stream and stops response-body consumption.
    #[cfg(feature = "internal")]
    pub async fn collect_bounded(
        mut self,
        limits: CollectedQueryLimits,
    ) -> Result<CollectedQueryResult, BifrostClientError> {
        let mut batches = Vec::new();
        let mut rows = 0usize;
        loop {
            let batch = match self.next_batch().await {
                Ok(batch) => batch,
                Err(error) => return Err(self.settle_with(error).await),
            };
            if self.encoded_bytes > limits.max_encoded_bytes {
                return Err(self.settle_with(BifrostClientError::ResultTooLarge).await);
            }
            let Some(batch) = batch else {
                break;
            };
            let next_rows = match rows.checked_add(batch.num_rows()) {
                Some(next_rows) if next_rows <= limits.max_rows => next_rows,
                _ => return Err(self.settle_with(BifrostClientError::ResultTooLarge).await),
            };
            rows = next_rows;
            batches.push(batch);
        }
        let terminal = self
            .terminal
            .ok_or(BifrostClientError::IncompleteQueryStream)?;
        let schema = self
            .schema
            .ok_or(BifrostClientError::IncompleteQueryStream)?;
        Ok(CollectedQueryResult {
            schema,
            batches,
            terminal,
            rows,
            encoded_bytes: self.encoded_bytes,
        })
    }

    /// Settles this stream because `error` ended it, and returns that error.
    ///
    /// The originating error is always what the caller sees: settlement is the
    /// client's obligation to the server, not a second failure to report. A
    /// decode or transport error also means the body can no longer be trusted,
    /// so it downgrades settlement to the status-polling proof before running.
    #[cfg(feature = "internal")]
    async fn settle_with(&mut self, error: BifrostClientError) -> BifrostClientError {
        let error = self.mark_broken(error);
        self.settle().await;
        error
    }

    /// Settles one incomplete stream exactly once.
    ///
    /// A stream that already reached its terminal returns immediately, and a
    /// second call is a no-op, so cancellation is issued at most once per
    /// query. A healthy body is drained to its real terminal, which is both the
    /// cheapest proof of cleanup and the only one that keeps the connection
    /// reusable. A broken body is never read again; the query's own status
    /// route is polled instead until it reports the entry is gone.
    ///
    /// Every wait is bounded by the server-pinned deadline this stream was
    /// opened with, so settlement can never outlive the query it settles. If
    /// the deadline passes without proof, that is reported as scrubbed
    /// telemetry rather than raised: the caller's own error is the one that
    /// matters, and the server still owns its cleanup.
    #[cfg(feature = "internal")]
    pub async fn settle(&mut self) {
        let owed = std::mem::replace(&mut self.settlement, StreamSettlement::Settled);
        match owed {
            StreamSettlement::Settled => return,
            StreamSettlement::Healthy | StreamSettlement::Broken => {}
        }
        let remaining = self.remaining();
        if remaining.is_zero() {
            Self::warn_unconfirmed(&self.request_id, self.deadline_ms);
            return;
        }
        let client = self.client.clone();
        let request_id = self.request_id.clone();
        let _ = tokio::time::timeout(remaining, client.cancel(&request_id)).await;
        if owed == StreamSettlement::Healthy && self.drain_to_terminal().await {
            return;
        }
        Self::poll_until_retired(client, request_id, self.deadline_ms).await;
        // A drain failure marked the body broken; the completed status cycle
        // discharged that, so this stream owes nothing further.
        self.settlement = StreamSettlement::Settled;
    }

    /// Drains a healthy body to its terminal within the query deadline.
    ///
    /// The query deadline bounds only terminal production. When it expires
    /// after the terminal arrived but before its EOF, the cancelled read left
    /// that terminal provisional in the raw stream, so one more read proves the
    /// close under the transport timeout [`Self::finish_at_clean_eof`] applies.
    ///
    /// Returns whether the terminal was accepted. A drain whose terminal never
    /// arrived by the deadline, or a late decode failure, leaves the caller
    /// with no proof, so it reports `false` and the status poll takes over.
    #[cfg(feature = "internal")]
    async fn drain_to_terminal(&mut self) -> bool {
        let produced = tokio::time::timeout(self.remaining(), async {
            while let Ok(Some(_)) = self.next_batch().await {}
        })
        .await;
        if produced.is_err() && self.terminal.is_none() && self.raw.terminal().is_some() {
            let _ = self.next_batch().await;
        }
        self.terminal.is_some()
    }

    /// Polls one query's status until the server proves it is no longer running.
    ///
    /// It takes its owners by value rather than borrowing the stream, so the
    /// returned future stays `Send` without requiring the stream itself to be
    /// `Sync` — callers spawn settlement onto a multi-threaded runtime.
    ///
    /// Only the not-found projection is proof; every other outcome — still
    /// running, unavailable, a transport failure — means the answer is not in
    /// yet, so the poll simply waits out its fixed interval and asks again
    /// until the deadline retires it.
    #[cfg(feature = "internal")]
    async fn poll_until_retired(client: QueryClient, request_id: RequestId, deadline_ms: i64) {
        loop {
            let remaining = Self::remaining_until(deadline_ms);
            if remaining.is_zero() {
                Self::warn_unconfirmed(&request_id, deadline_ms);
                return;
            }
            if let Ok(outcome) = tokio::time::timeout(remaining, client.status(&request_id)).await
                && outcome.is_err_and(|error| {
                    WyrdError::from(&error).code() == RUNNING_QUERY_RETIRED_CODE
                })
            {
                return;
            }
            let remaining = Self::remaining_until(deadline_ms);
            if remaining.is_zero() {
                Self::warn_unconfirmed(&request_id, deadline_ms);
                return;
            }
            tokio::time::sleep(SETTLEMENT_POLL_INTERVAL.min(remaining)).await;
        }
    }

    /// Reports unproven settlement as scrubbed telemetry rather than an error.
    ///
    /// The caller's own error is the one that matters and the server still owns
    /// its cleanup, so an unconfirmed settlement is only ever observability.
    #[cfg(feature = "internal")]
    fn warn_unconfirmed(request_id: &RequestId, deadline_ms: i64) {
        tracing::warn!(
            request_id = %request_id,
            deadline_ms,
            "Oracle query settlement remains unconfirmed at its deadline"
        );
    }

    /// Returns the time left before this query's server-pinned deadline.
    ///
    /// A deadline that has already passed yields zero rather than a negative
    /// duration, which makes every bounded settlement wait resolve immediately
    /// instead of panicking on the conversion.
    #[cfg(feature = "internal")]
    fn remaining(&self) -> std::time::Duration {
        Self::remaining_until(self.deadline_ms)
    }

    /// Returns the time left before `deadline_ms`, without borrowing a stream.
    ///
    /// Settlement's status poll owns its collaborators by value so its future
    /// stays `Send`, and it must recompute the remainder before every wait, so
    /// the deadline arithmetic lives here rather than on `&self`.
    #[cfg(feature = "internal")]
    fn remaining_until(deadline_ms: i64) -> std::time::Duration {
        let now = std::time::SystemTime::UNIX_EPOCH
            .elapsed()
            .map_or(i64::MAX, |since| {
                i64::try_from(since.as_millis()).unwrap_or(i64::MAX)
            });
        u64::try_from(deadline_ms - now).map_or(std::time::Duration::ZERO, |remaining| {
            std::time::Duration::from_millis(remaining)
        })
    }

    /// Returns this stream's server-pinned absolute deadline in epoch milliseconds.
    #[must_use]
    pub const fn deadline_ms(&self) -> i64 {
        self.deadline_ms
    }

    /// Returns terminal metadata only after validated completion or failure.
    #[must_use]
    pub fn terminal(&self) -> Option<&QueryTerminalFrame> {
        self.terminal.as_ref()
    }

    /// Returns the decoded query schema after the schema frame is consumed.
    #[must_use]
    pub fn schema(&self) -> Option<&SchemaRef> {
        self.schema.as_ref()
    }

    /// Returns exact length-delimited response bytes received so far.
    #[must_use]
    #[cfg(feature = "internal")]
    pub fn encoded_bytes(&self) -> usize {
        self.encoded_bytes
    }

    /// Reports whether the query's Arrow IPC stream was explicitly closed.
    ///
    /// A successful or degraded query closes its single IPC stream with the
    /// end-of-stream delta carried by the terminal frame. A caller that must
    /// distinguish a complete result from a truncated one reads this rather
    /// than inferring completeness from an absent batch.
    #[must_use]
    pub const fn arrow_ipc_closed(&self) -> bool {
        self.raw.arrow_ipc_closed()
    }

    /// Returns the largest single Arrow fragment this stream held while decoding.
    #[must_use]
    pub const fn peak_pending_frame_bytes(&self) -> usize {
        self.raw.peak_pending_frame_bytes()
    }

    /// Returns total Arrow IPC bytes this query carried across every fragment.
    ///
    /// The query is one split IPC stream, so this counts one schema prefix,
    /// each batch's bare delta, and the single end-of-stream delta — strictly
    /// less than re-encoding each batch as its own standalone stream once the
    /// query returns more than one batch.
    #[must_use]
    pub const fn arrow_ipc_bytes(&self) -> usize {
        self.raw.arrow_ipc_bytes()
    }
}

/// Explicit collection ceilings used by CLI, MCP, and tests.
#[derive(Debug, Clone, Copy)]
pub struct CollectedQueryLimits {
    /// Maximum decoded rows.
    pub max_rows: usize,
    /// Maximum length-delimited response bytes accounted for by the collector.
    pub max_encoded_bytes: usize,
}

/// Bounded Arrow query result with its authoritative schema and validated terminal metadata.
#[derive(Debug)]
pub struct CollectedQueryResult {
    /// Schema decoded from the stream's required initial schema frame.
    pub schema: SchemaRef,
    /// Collected record batches.
    pub batches: Vec<RecordBatch>,
    /// Terminal metadata.
    pub terminal: QueryTerminalFrame,
    /// Total decoded rows.
    pub rows: usize,
    /// Total length-delimited response bytes accounted for by the collector.
    pub encoded_bytes: usize,
}

/// Stateful Arrow IPC decoder for one public query stream.
///
/// The public stream is a single Arrow IPC stream split across Wyrd frames, so
/// the schema and every dictionary arrive once, in earlier fragments. A decoder
/// constructed per frame cannot read that stream at all; this owner is
/// constructed once per query and consumes fragments in order.
///
/// The server tier has its own copy of this state machine in
/// `vala-bifrost-redux`. Client-tier crates may not depend on the Bifrost
/// server engine, so the wire contract — not shared code — is what keeps the
/// two honest, and the journeys in `tests/integration/pg_bifrost_e2e.rs` are what prove it.
///
/// End-of-stream receipt is tracked here rather than delegated to
/// [`arrow::ipc::reader::StreamDecoder::finish`], which reports success both
/// for a stream that closed cleanly and for one that never started.
struct QueryIpcDecoder {
    /// Arrow's push decoder, retaining schema and dictionary state.
    decoder: arrow::ipc::reader::StreamDecoder,
    /// Schema recovered from the required initial schema fragment.
    schema: Option<SchemaRef>,
    /// Whether the terminal's explicit end-of-stream delta was accepted.
    eos_accepted: bool,
    /// Largest single fragment this decoder has held while decoding.
    peak_pending_frame_bytes: usize,
    /// Total Arrow IPC bytes this decoder consumed across every fragment.
    total_fragment_bytes: usize,
}

impl QueryIpcDecoder {
    /// Creates a decoder positioned before the required schema fragment.
    fn new() -> Self {
        Self {
            decoder: arrow::ipc::reader::StreamDecoder::new(),
            schema: None,
            eos_accepted: false,
            peak_pending_frame_bytes: 0,
            total_fragment_bytes: 0,
        }
    }

    /// Consumes the initial schema fragment and returns the stream schema.
    ///
    /// The schema is read with a throwaway [`StreamReader`] because Arrow's
    /// push decoder only completes a zero-body message once the next fragment
    /// arrives; the same bytes are still pushed through the push decoder, which
    /// is what carries schema and dictionary state into later fragments.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Protocol`] for a duplicate or post-terminal
    /// schema, and [`BifrostClientError::Arrow`] when the fragment is not exactly one
    /// schema message.
    fn accept_schema(&mut self, bytes: &[u8]) -> Result<SchemaRef, BifrostClientError> {
        if self.schema.is_some() || self.eos_accepted {
            return Err(BifrostClientError::Protocol(
                "query stream carries more than one schema frame".to_owned(),
            ));
        }
        let mut prefix = StreamReader::try_new(Cursor::new(bytes), None)
            .map_err(|error| BifrostClientError::Arrow(error.to_string()))?;
        let schema = prefix.schema();
        if prefix
            .next()
            .transpose()
            .map_err(|error| BifrostClientError::Arrow(error.to_string()))?
            .is_some()
        {
            return Err(BifrostClientError::Arrow(
                "schema frame unexpectedly contains a record batch".to_owned(),
            ));
        }
        if self.feed(bytes)?.is_some() {
            return Err(BifrostClientError::Arrow(
                "schema frame unexpectedly contains a record batch".to_owned(),
            ));
        }
        self.schema = Some(SchemaRef::clone(&schema));
        Ok(schema)
    }

    /// Consumes one batch fragment and returns its decoded record batch.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Protocol`] before the schema or after
    /// end-of-stream, and [`BifrostClientError::Arrow`] when the fragment does not
    /// decode to exactly one record batch matching the stream schema.
    fn accept_batch(&mut self, bytes: &[u8]) -> Result<RecordBatch, BifrostClientError> {
        let expected = self.schema.as_ref().ok_or_else(|| {
            BifrostClientError::Protocol("batch arrived before schema".to_owned())
        })?;
        if self.eos_accepted {
            return Err(BifrostClientError::Protocol(
                "query stream frame arrived after its end-of-stream".to_owned(),
            ));
        }
        let expected = SchemaRef::clone(expected);
        let batch = self.feed(bytes)?.ok_or_else(|| {
            BifrostClientError::Arrow("batch frame contains no record batch".to_owned())
        })?;
        if batch.schema().as_ref() != expected.as_ref() {
            return Err(BifrostClientError::Arrow(
                "batch schema does not match the initial schema".to_owned(),
            ));
        }
        Ok(batch)
    }

    /// Consumes the terminal's end-of-stream delta and closes the stream.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Protocol`] before the schema or on a second
    /// end-of-stream, and [`BifrostClientError::Arrow`] when the delta is absent,
    /// carries a record batch, or leaves a partial message behind.
    fn accept_eos(&mut self, bytes: &[u8]) -> Result<(), BifrostClientError> {
        if self.schema.is_none() || self.eos_accepted {
            return Err(BifrostClientError::Protocol(
                "query stream end-of-stream is out of order".to_owned(),
            ));
        }
        if bytes.is_empty() {
            return Err(BifrostClientError::Arrow(
                "successful query terminal carries no Arrow IPC end-of-stream".to_owned(),
            ));
        }
        if self.feed(bytes)?.is_some() {
            return Err(BifrostClientError::Arrow(
                "query end-of-stream unexpectedly contains a record batch".to_owned(),
            ));
        }
        self.decoder
            .finish()
            .map_err(|error| BifrostClientError::Arrow(error.to_string()))?;
        self.eos_accepted = true;
        Ok(())
    }

    /// Reports whether the explicit end-of-stream delta was accepted.
    const fn eos_accepted(&self) -> bool {
        self.eos_accepted
    }

    /// Returns the largest single fragment this decoder held while decoding.
    const fn peak_pending_frame_bytes(&self) -> usize {
        self.peak_pending_frame_bytes
    }

    /// Returns the total Arrow IPC bytes consumed across every fragment.
    const fn total_fragment_bytes(&self) -> usize {
        self.total_fragment_bytes
    }

    /// Pushes one fragment through Arrow's decoder, allowing at most one batch.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Arrow`] when Arrow rejects the bytes or the
    /// fragment yields more than one record batch.
    fn feed(&mut self, bytes: &[u8]) -> Result<Option<RecordBatch>, BifrostClientError> {
        self.peak_pending_frame_bytes = self.peak_pending_frame_bytes.max(bytes.len());
        self.total_fragment_bytes = self.total_fragment_bytes.saturating_add(bytes.len());
        let mut buffer = arrow::buffer::Buffer::from_vec(bytes.to_vec());
        let mut decoded = None;
        while !buffer.is_empty() {
            match self
                .decoder
                .decode(&mut buffer)
                .map_err(|error| BifrostClientError::Arrow(error.to_string()))?
            {
                Some(batch) if decoded.is_none() => decoded = Some(batch),
                Some(_) => {
                    return Err(BifrostClientError::Arrow(
                        "batch frame contains more than one record batch".to_owned(),
                    ));
                }
                None => {}
            }
        }
        Ok(decoded)
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
    use std::task::{Context, Poll};

    use arrow::array::Int64Array;
    use arrow::datatypes::{DataType, Field, Schema};
    use arrow::ipc::writer::StreamWriter;
    use bytes::Bytes;
    use futures_util::stream;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;
    use wyrd_spec::error::WyrdError;
    use wyrd_spec::vala::api::{
        QueryBatchFrame, QuerySchemaFrame, QuerySource, QueryWarning, SourceCompletion,
        SourceCompletionOutcome,
    };
    use wyrd_tonic::frame_codec::FrameEncoder;
    use wyrd_tonic::wyrd::v1 as proto;

    use super::*;

    /// Counts the lifecycle calls one settling stream makes against a server.
    #[derive(Debug, Default)]
    struct LifecycleCounts {
        /// Cancellation requests received on the query's own route.
        cancels: AtomicUsize,
        /// Status polls received on the query's own route.
        statuses: AtomicUsize,
    }

    /// Serves the two lifecycle routes settlement uses, counting every call.
    ///
    /// The status route always reports the query is gone, which is the proof a
    /// broken stream polls for. Each connection is answered once and closed, so
    /// the counters record exactly how many requests the client actually made.
    fn lifecycle_server(counts: Arc<LifecycleCounts>) -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("test listener binds");
        let address = listener.local_addr().expect("listener has an address");
        listener
            .set_nonblocking(true)
            .expect("listener converts to tokio");
        let listener = TcpListener::from_std(listener).expect("listener adopts the runtime");
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let counts = Arc::clone(&counts);
                tokio::spawn(async move {
                    let mut request = [0_u8; 2048];
                    let Ok(read) = socket.read(&mut request).await else {
                        return;
                    };
                    let request = String::from_utf8_lossy(&request[..read]).to_string();
                    let response = if request.starts_with("POST /auth/token") {
                        // The client exchanges its API key before any lifecycle
                        // route; a far-future expiry keeps one exchange enough.
                        let body = format!(
                            "{{\"access_token\":\"test-token\",\"token_type\":\"Bearer\",\"expires_at\":\"{}\"}}",
                            "2099-01-01T00:00:00Z"
                        );
                        format!(
                            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                            body.len()
                        )
                    } else if request.starts_with("DELETE ") {
                        counts.cancels.fetch_add(1, Ordering::AcqRel);
                        let body = "{\"cancellation_requested\":true}";
                        format!(
                            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                            body.len()
                        )
                    } else {
                        counts.statuses.fetch_add(1, Ordering::AcqRel);
                        let body = WyrdError::from(BifrostError::RunningQueryNotFound)
                            .as_problem_json()
                            .to_string();
                        format!(
                            "HTTP/1.1 404 Not Found\r\ncontent-type: application/problem+json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                            body.len()
                        )
                    };
                    let _ = socket.write_all(response.as_bytes()).await;
                });
            }
        });
        format!("http://{address}")
    }

    /// Builds a stream bound to a live lifecycle server with a live deadline.
    fn settling_stream(
        chunks: Vec<Vec<u8>>,
        base_url: &str,
        deadline_ms: i64,
    ) -> QueryResultStream {
        let body = stream::iter(chunks.into_iter().map(|chunk| Ok(Bytes::from(chunk))));
        QueryResultStream::new(
            RawQueryStream::new(body),
            RequestId::now_v7(),
            client_for(base_url),
            deadline_ms,
        )
    }

    /// Returns an absolute epoch-millisecond deadline `millis` from now.
    fn deadline_in(millis: i64) -> i64 {
        let now = i64::try_from(
            std::time::SystemTime::UNIX_EPOCH
                .elapsed()
                .expect("the test clock is after the epoch")
                .as_millis(),
        )
        .expect("the epoch millisecond fits an i64");
        now + millis
    }

    /// A healthy stream whose settlement drain fails settles exactly once.
    ///
    /// The drain meets an undecodable frame, which marks the body broken and
    /// falls back to the status proof. A completed cycle must leave the stream
    /// terminally settled, so a second call neither cancels nor polls again.
    ///
    /// # Panics
    ///
    /// Panics if settlement cancels or polls status more than once.
    #[tokio::test]
    async fn failed_healthy_drain_settles_once() {
        let schema = test_schema();
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let batch = ipc.batch(&schema, &[1, 2, 3]);
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let mut stream = settling_stream(
            vec![
                encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                    schema_fingerprint: String::new(),
                    arrow_ipc_schema: prefix,
                })),
                encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                    arrow_ipc_batch: batch,
                })),
                encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                    arrow_ipc_batch: vec![0xff, 0xff, 0xff, 0xff],
                })),
            ],
            &base_url,
            deadline_in(30_000),
        );
        stream.next_batch().await.expect("first batch decodes");

        stream.settle().await;
        stream.settle().await;

        assert_eq!(counts.cancels.load(Ordering::Acquire), 1);
        assert_eq!(counts.statuses.load(Ordering::Acquire), 1);
        assert_eq!(stream.settlement, StreamSettlement::Settled);
    }

    /// Every incomplete exit settles the server exactly once, under the deadline.
    ///
    /// The five exits a caller can actually take are covered together because
    /// the contract they share is a single one: whatever ends the stream, the
    /// server learns about it once, the caller keeps its own error, and nothing
    /// waits past the deadline the server pinned.
    #[tokio::test]
    async fn query_result_stream_settles_every_incomplete_exit_once() {
        let schema = test_schema();
        let (_, prefix_only) = TestQueryIpc::open(&schema);
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let batch = ipc.batch(&schema, &[1, 2, 3]);
        let eos = ipc.close();
        let complete = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: String::new(),
                arrow_ipc_schema: prefix.clone(),
            })),
            encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: batch.clone(),
            })),
            encoded(QueryStreamFrame::Terminal(success_terminal(3, eos))),
        ];

        // A stream that reached its terminal owes the server nothing.
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let mut settled = settling_stream(complete.clone(), &base_url, deadline_in(30_000));
        while settled
            .next_batch()
            .await
            .expect("complete stream decodes")
            .is_some()
        {}
        settled.settle().await;
        assert_eq!(
            counts.cancels.load(Ordering::Acquire),
            0,
            "a terminal is the server's own settlement"
        );

        // A healthy stream abandoned early cancels once and drains its body.
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let mut healthy = settling_stream(complete, &base_url, deadline_in(30_000));
        healthy.next_batch().await.expect("first batch decodes");
        healthy.settle().await;
        healthy.settle().await;
        assert_eq!(
            counts.cancels.load(Ordering::Acquire),
            1,
            "settlement cancels at most once across repeated calls"
        );
        assert!(
            healthy.terminal().is_some(),
            "a healthy body is drained to its real terminal"
        );
        assert_eq!(
            counts.statuses.load(Ordering::Acquire),
            0,
            "a drained terminal needs no status proof"
        );

        // A bounded-result overflow settles before the caller sees the error.
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let wide = ipc.batch(&schema, &[1, 2, 3, 4, 5]);
        let eos = ipc.close();
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let overflow = settling_stream(
            vec![
                encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                    schema_fingerprint: String::new(),
                    arrow_ipc_schema: prefix,
                })),
                encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                    arrow_ipc_batch: wide,
                })),
                encoded(QueryStreamFrame::Terminal(success_terminal(5, eos))),
            ],
            &base_url,
            deadline_in(30_000),
        );
        let error = overflow
            .collect_bounded(CollectedQueryLimits {
                max_rows: 1,
                max_encoded_bytes: usize::MAX,
            })
            .await
            .expect_err("the row ceiling refuses this result");
        assert!(matches!(error, BifrostClientError::ResultTooLarge));
        assert_eq!(
            counts.cancels.load(Ordering::Acquire),
            1,
            "an overflow exit settles the server once"
        );

        // A protocol failure keeps its own error and proves cleanup by status.
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let broken = settling_stream(
            vec![encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: vec![0xff, 0xff, 0xff, 0xff],
            }))],
            &base_url,
            deadline_in(30_000),
        );
        let error = broken
            .collect_bounded(CollectedQueryLimits {
                max_rows: usize::MAX,
                max_encoded_bytes: usize::MAX,
            })
            .await
            .expect_err("an undecodable batch fails");
        assert!(
            matches!(
                error,
                BifrostClientError::Protocol(_) | BifrostClientError::Arrow(_)
            ),
            "settlement never replaces the originating error: {error:?}"
        );
        assert_eq!(counts.cancels.load(Ordering::Acquire), 1);
        assert!(
            counts.statuses.load(Ordering::Acquire) >= 1,
            "a broken body is proven retired by status, never by reading it again"
        );

        // A deadline already in the past bounds settlement to no waiting at all,
        // and no leg of it receives a fresh budget: an expired stream neither
        // cancels, drains, nor polls status.
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let mut expired = settling_stream(Vec::new(), &base_url, deadline_in(-1));
        expired.settlement = StreamSettlement::Broken;
        tokio::time::timeout(std::time::Duration::from_secs(5), expired.settle())
            .await
            .expect("an expired deadline settles without waiting");
        assert_eq!(
            counts.cancels.load(Ordering::Acquire),
            0,
            "an elapsed deadline gives cancellation no fresh budget"
        );
        assert_eq!(
            counts.statuses.load(Ordering::Acquire),
            0,
            "an elapsed deadline gives status polling no fresh budget"
        );

        // A cancellation endpoint that never answers cannot outlive the
        // stream's own deadline.
        let base_url = unanswering_server();
        let mut hung = settling_stream(Vec::new(), &base_url, deadline_in(400));
        hung.settlement = StreamSettlement::Broken;
        tokio::time::timeout(std::time::Duration::from_secs(10), hung.settle())
            .await
            .expect("settlement returns by the stream deadline, not the server's");

        // Every direct body failure marks the stream broken before it returns,
        // so settlement never reads that body again.
        for (label, chunks) in [
            (
                "an undecodable frame",
                vec![vec![0xff_u8, 0xff, 0xff, 0xff, 0xff, 0xff]],
            ),
            (
                "a body that ends before its terminal",
                vec![encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                    schema_fingerprint: String::new(),
                    arrow_ipc_schema: prefix_only.clone(),
                }))],
            ),
        ] {
            let counts = Arc::new(LifecycleCounts::default());
            let base_url = lifecycle_server(Arc::clone(&counts));
            let polls = Arc::new(AtomicUsize::new(0));
            let mut stream =
                counted_stream(chunks, Arc::clone(&polls), &base_url, deadline_in(30_000));
            let error = stream.next_batch().await.expect_err(label).to_string();
            let polled = polls.load(Ordering::Acquire);
            assert_eq!(
                stream.settlement,
                StreamSettlement::Broken,
                "{label} marks the body broken before returning: {error}"
            );
            stream.settle().await;
            assert_eq!(
                polls.load(Ordering::Acquire),
                polled,
                "{label} leaves a body settlement must never repoll"
            );
            assert_eq!(counts.cancels.load(Ordering::Acquire), 1);
            assert!(counts.statuses.load(Ordering::Acquire) >= 1);
        }

        // A real transport failure takes the same path: the truncated body is
        // marked broken and never read again.
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let mut truncated = QueryResultStream::new(
            RawQueryStream::new(truncated_body().await),
            RequestId::now_v7(),
            client_for(&base_url),
            deadline_in(30_000),
        );
        let error = truncated
            .next_batch()
            .await
            .expect_err("a truncated body is a transport failure");
        assert!(
            matches!(error, BifrostClientError::Transport(_)),
            "the caller keeps its transport error: {error:?}"
        );
        assert_eq!(truncated.settlement, StreamSettlement::Broken);

        // A successful terminal is provisional until the body closes: anything
        // that follows it is a protocol failure and can never become a result.
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let batch = ipc.batch(&schema, &[1, 2, 3]);
        let eos = ipc.close();
        let head = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: String::new(),
                arrow_ipc_schema: prefix.clone(),
            })),
            encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: batch.clone(),
            })),
            encoded(QueryStreamFrame::Terminal(success_terminal(3, eos.clone()))),
        ];
        for (label, trailing) in [
            (
                "a duplicate terminal",
                encoded(QueryStreamFrame::Terminal(success_terminal(3, eos.clone()))),
            ),
            (
                "a late schema",
                encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                    schema_fingerprint: String::new(),
                    arrow_ipc_schema: prefix.clone(),
                })),
            ),
            (
                "a trailing batch",
                encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                    arrow_ipc_batch: batch.clone(),
                })),
            ),
        ] {
            let counts = Arc::new(LifecycleCounts::default());
            let base_url = lifecycle_server(Arc::clone(&counts));
            let mut chunks = head.clone();
            chunks.push(trailing);
            let mut stream = settling_stream(chunks, &base_url, deadline_in(30_000));
            stream.next_batch().await.expect("the batch decodes");
            let error = stream
                .next_batch()
                .await
                .expect_err("{label} after a terminal is refused");
            assert!(
                matches!(
                    error,
                    BifrostClientError::Protocol(_) | BifrostClientError::Arrow(_)
                ),
                "{label} after a terminal is a protocol failure: {error:?}"
            );
            assert_eq!(
                stream.settlement,
                StreamSettlement::Broken,
                "{label} after a terminal marks the body broken"
            );
            assert!(
                stream.terminal().is_none(),
                "{label} after a terminal never becomes a successful result"
            );
        }

        // The same terminal followed by clean EOF alone is what settles.
        let counts = Arc::new(LifecycleCounts::default());
        let base_url = lifecycle_server(Arc::clone(&counts));
        let mut clean = settling_stream(head, &base_url, deadline_in(30_000));
        clean.next_batch().await.expect("the batch decodes");
        assert!(
            clean
                .next_batch()
                .await
                .expect("clean EOF completes the stream")
                .is_none()
        );
        assert!(
            clean.terminal().is_some(),
            "clean EOF alone promotes the terminal to this stream's result"
        );
        clean.settle().await;
        assert_eq!(
            counts.cancels.load(Ordering::Acquire),
            0,
            "a terminal proven by clean EOF owes the server nothing"
        );
    }

    /// Serves one canned JSON body and records every request line it answers.
    ///
    /// The recorded lines are the proof of a typed method's HTTP contract: the
    /// exact verb, path, and query string a caller's request produced. Each
    /// connection is answered once and closed so the recording stays ordered.
    fn recording_server(body: &'static str) -> (String, Arc<std::sync::Mutex<Vec<String>>>) {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("test listener binds");
        let address = listener.local_addr().expect("listener has an address");
        listener
            .set_nonblocking(true)
            .expect("listener converts to tokio");
        let listener = TcpListener::from_std(listener).expect("listener adopts the runtime");
        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
        let recorded = Arc::clone(&seen);
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let recorded = Arc::clone(&recorded);
                tokio::spawn(async move {
                    let mut request = [0_u8; 4096];
                    let Ok(read) = socket.read(&mut request).await else {
                        return;
                    };
                    let request = String::from_utf8_lossy(&request[..read]).to_string();
                    let response = if request.starts_with("POST /auth/token") {
                        let token = "{\"access_token\":\"test-token\",\"token_type\":\"Bearer\",\"expires_at\":\"2099-01-01T00:00:00Z\"}";
                        format!(
                            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{token}",
                            token.len()
                        )
                    } else {
                        let line = request.lines().next().unwrap_or_default().to_owned();
                        if let Ok(mut recorded) = recorded.lock() {
                            recorded.push(line);
                        }
                        format!(
                            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                            body.len()
                        )
                    };
                    let _ = socket.write_all(response.as_bytes()).await;
                });
            }
        });
        (format!("http://{address}"), seen)
    }

    /// Serves the token exchange and captures the first `POST /v1/query` body.
    ///
    /// The query is answered with an empty `503`, so the caller fails fast
    /// after its request has been fully written; the captured JSON body is the
    /// proof of exactly what the client put on the wire.
    fn query_body_server() -> (String, tokio::sync::oneshot::Receiver<serde_json::Value>) {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("test listener binds");
        let address = listener.local_addr().expect("listener has an address");
        listener
            .set_nonblocking(true)
            .expect("listener converts to tokio");
        let listener = TcpListener::from_std(listener).expect("listener adopts the runtime");
        let (sender, receiver) = tokio::sync::oneshot::channel();
        let sender = Arc::new(std::sync::Mutex::new(Some(sender)));
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let sender = Arc::clone(&sender);
                tokio::spawn(async move {
                    let mut request = Vec::new();
                    let mut chunk = [0_u8; 4096];
                    let (head, body) = loop {
                        let Ok(read) = socket.read(&mut chunk).await else {
                            return;
                        };
                        if read == 0 {
                            return;
                        }
                        request.extend_from_slice(&chunk[..read]);
                        let text = String::from_utf8_lossy(&request).to_string();
                        let Some((head, body)) = text.split_once("\r\n\r\n") else {
                            continue;
                        };
                        let length = head
                            .lines()
                            .find_map(|line| {
                                line.to_ascii_lowercase()
                                    .strip_prefix("content-length:")
                                    .map(|value| value.trim().parse::<usize>().unwrap_or(0))
                            })
                            .unwrap_or(0);
                        if body.len() >= length {
                            break (head.to_owned(), body.to_owned());
                        }
                    };
                    let response = if head.starts_with("POST /auth/token") {
                        let token = "{\"access_token\":\"test-token\",\"token_type\":\"Bearer\",\"expires_at\":\"2099-01-01T00:00:00Z\"}";
                        format!(
                            "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{token}",
                            token.len()
                        )
                    } else {
                        if head.starts_with("POST /v1/query ")
                            && let Some(sender) =
                                sender.lock().ok().and_then(|mut slot| slot.take())
                        {
                            let _ = sender.send(
                                serde_json::from_str(&body).unwrap_or(serde_json::Value::Null),
                            );
                        }
                        "HTTP/1.1 503 Service Unavailable\r\ncontent-length: 0\r\nconnection: close\r\n\r\n".to_owned()
                    };
                    let _ = socket.write_all(response.as_bytes()).await;
                });
            }
        });
        (format!("http://{address}"), receiver)
    }

    /// `sql(query, params)` sends bind values as ordered typed data.
    ///
    /// A parameter holding SQL text must reach the server as one JSON string
    /// in `params` while `sql` stays byte-for-byte the caller's placeholder
    /// template, so no client path can splice a value into the statement.
    ///
    /// # Panics
    ///
    /// Panics when the request never reaches the query route, or when the SQL
    /// text or the ordered params differ from what the caller supplied.
    #[tokio::test]
    async fn sql_forwards_bind_values_without_interpolation() {
        let (base_url, body) = query_body_server();
        let config = crate::config::ClientConfig {
            credential: Some(secrecy::SecretString::from("test-key")),
            http: crate::transport::config::HttpConfig {
                base_url,
                timeout_ms: 2_000,
                ..crate::transport::config::HttpConfig::default()
            },
            ..crate::config::ClientConfig::default()
        };
        let client = WyrdClient::with_config(config).expect("static config builds a client");
        let template = "SELECT v FROM t WHERE name = $1 AND n = $2";
        let injection = "x'); DROP TABLE t; --";
        let _ = crate::bifrost::Bifrost::query_only(&client)
            .sql(
                template,
                &[
                    wyrd_spec::vala::api::QueryParam::String(injection.to_owned()),
                    wyrd_spec::vala::api::QueryParam::Int(7),
                ],
            )
            .await;
        let body = tokio::time::timeout(std::time::Duration::from_secs(5), body)
            .await
            .expect("the query request arrives")
            .expect("the query body is captured");
        assert_eq!(
            body,
            serde_json::json!({ "sql": template, "params": [injection, 7], "deadline_ms": null })
        );
    }

    /// A described canonical table builds its insertable Arrow schema directly.
    ///
    /// The three describe classes are exactly what a writer needs: it declares
    /// the user fields, supplies the correlation inputs, and may supply the
    /// managed candidate. Building a schema from that description must append
    /// each correlation column exactly once — a writer that also re-declared
    /// `card_uid` would collide with the one the write path resolves.
    ///
    /// # Panics
    ///
    /// Panics when describe does not reach its route, when the projected
    /// schema loses a declared column or its stable field id, or when a
    /// correlation column appears twice.
    #[tokio::test]
    async fn describe_builds_writable_schema_without_duplicate_correlation() {
        let body = r#"{
            "entry": {
                "namespace": "vala.traces",
                "name": "spans",
                "table_uid": "0102030405060708090a0b0c0d0e0f10",
                "status": "Active",
                "fingerprint": "aa",
                "registered_at": "2026-07-01T00:00:00Z",
                "updated_at": "2026-07-01T00:00:00Z"
            },
            "user_fields": [
                { "name": "trace_id", "data_type": "Binary", "nullable": false,
                  "metadata": { "PARQUET:field_id": "1" } }
            ],
            "correlation_fields": [
                { "name": "card_uid", "data_type": "Utf8", "nullable": true,
                  "metadata": { "PARQUET:field_id": "1001" } },
                { "name": "run_id", "data_type": "Utf8", "nullable": true,
                  "metadata": { "PARQUET:field_id": "1000" } }
            ],
            "canonical_physical_fingerprint": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
            "physical_layout": { "partition_granularity": "hour" }
        }"#;
        let (base_url, seen) = recording_server(body);
        let described = client_for(&base_url)
            .describe_table("vala.traces", "spans")
            .await
            .expect("describe returns the stored schema projection");

        assert_eq!(
            seen.lock().expect("recording is readable").as_slice(),
            ["GET /v1/bifrost/tables/vala.traces/spans HTTP/1.1"],
            "describe reads the one table route and sends no body"
        );
        assert_eq!(
            described.canonical_physical_fingerprint.as_deref(),
            Some("0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"),
            "a canonical table publishes its exact physical fingerprint"
        );

        let writable = wyrd_types::schema::writable_schema(&described)
            .expect("the description projects an Arrow schema");
        assert_eq!(
            writable
                .fields()
                .iter()
                .map(|field| field.name().as_str())
                .collect::<Vec<_>>(),
            ["trace_id", "card_uid", "run_id"],
            "user fields precede correlation inputs and no managed column is writable"
        );
        assert!(
            writable
                .fields()
                .iter()
                .all(|field| field.metadata().is_empty()),
            "no writable column repeats the server-owned field id back on the wire"
        );

        let builder = wyrd_queue::RowPreflight::from_description(&described)
            .expect("the description builds a JSON row preflight");
        assert_eq!(
            builder
                .output_schema()
                .fields()
                .iter()
                .filter(|field| field.name() == "card_uid")
                .count(),
            1,
            "the write path appends each correlation column exactly once"
        );
        assert!(
            writable.field(1).is_nullable(),
            "card_uid is nullable: a writer may submit a row with no Card correlation"
        );
        assert!(
            builder.output_schema().field(1).is_nullable(),
            "the JSON row path carries the same optional correlation contract"
        );
    }

    /// observable against a peer that never does.
    fn unanswering_server() -> String {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("test listener binds");
        let address = listener.local_addr().expect("listener has an address");
        listener
            .set_nonblocking(true)
            .expect("listener converts to tokio");
        let listener = TcpListener::from_std(listener).expect("listener adopts the runtime");
        tokio::spawn(async move {
            let mut held = Vec::new();
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                let mut request = [0_u8; 2048];
                let Ok(read) = socket.read(&mut request).await else {
                    continue;
                };
                if String::from_utf8_lossy(&request[..read]).starts_with("POST /auth/token") {
                    let body = "{\"access_token\":\"test-token\",\"token_type\":\"Bearer\",\"expires_at\":\"2099-01-01T00:00:00Z\"}";
                    let response = format!(
                        "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
                        body.len()
                    );
                    let _ = socket.write_all(response.as_bytes()).await;
                } else {
                    // Held open and unanswered for the rest of the test.
                    held.push(socket);
                }
            }
        });
        format!("http://{address}")
    }

    /// Builds a settling stream whose body counts every poll it receives.
    ///
    /// The proof that a broken body is never read again is exactly that this
    /// counter stops moving once the failure is returned.
    fn counted_stream(
        chunks: Vec<Vec<u8>>,
        polls: Arc<AtomicUsize>,
        base_url: &str,
        deadline_ms: i64,
    ) -> QueryResultStream {
        let body = stream::unfold(
            (chunks.into_iter(), polls),
            |(mut chunks, polls)| async move {
                polls.fetch_add(1, Ordering::AcqRel);
                chunks
                    .next()
                    .map(|chunk| (Ok(Bytes::from(chunk)), (chunks, polls)))
            },
        );
        QueryResultStream::new(
            RawQueryStream::new(body),
            RequestId::now_v7(),
            client_for(base_url),
            deadline_ms,
        )
    }

    /// Returns a real HTTP response body that ends before its content length.
    ///
    /// A `reqwest::Error` has no constructor, so the only way to exercise the
    /// transport branch is to make one.
    async fn truncated_body() -> impl Stream<Item = Result<Bytes, reqwest::Error>> + Send + 'static
    {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("test listener binds");
        let address = listener.local_addr().expect("listener has an address");
        tokio::spawn(async move {
            let Ok((mut socket, _)) = listener.accept().await else {
                return;
            };
            let mut request = [0_u8; 1024];
            let _ = socket.read(&mut request).await;
            let _ = socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\ncontent-type: application/vnd.wyrd.bifrost-query-stream\r\ncontent-length: 100\r\nconnection: close\r\n\r\nabc",
                )
                .await;
        });
        reqwest::Client::new()
            .get(format!("http://{address}/v1/query"))
            .send()
            .await
            .expect("response headers arrive")
            .bytes_stream()
    }

    /// Proves an HTTP body failure remains a structured transport error.
    #[tokio::test]
    async fn query_body_failure_is_structured_transport_error() {
        let listener = TcpListener::bind("127.0.0.1:0")
            .await
            .expect("test listener binds");
        let address = listener.local_addr().expect("listener has an address");
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.expect("client connects");
            let mut request = [0_u8; 1024];
            let _ = socket.read(&mut request).await.expect("request reads");
            socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\ncontent-type: application/vnd.wyrd.bifrost-query-stream\r\ncontent-length: 100\r\nconnection: close\r\n\r\nabc",
                )
                .await
                .expect("truncated response writes");
        });
        let response = reqwest::Client::new()
            .get(format!("http://{address}/v1/query"))
            .send()
            .await
            .expect("response headers arrive");
        let mut stream = RawQueryStream::new(response.bytes_stream());

        let error = stream
            .next_frame()
            .await
            .expect_err("truncated body is a transport failure");
        let BifrostClientError::Transport(WyrdError::UpstreamFailure { details, .. }) = error
        else {
            panic!("body failure must retain the transport projection");
        };
        assert_eq!(details["transport"], "http");
        assert_eq!(details["phase"], "response_body");
        assert_eq!(details["timeout"], false);
        assert_eq!(details["connect"], false);
        assert_eq!(details["body"], false);
        assert_eq!(details["decode"], true);
        server.await.expect("test server exits");
    }

    /// The catalog projection preserves typed transport and terminal diagnostics.
    #[test]
    fn bifrost_client_error_projects_onto_the_catalog() {
        let transport = BifrostClientError::Transport(WyrdError::PermissionDeniedRbac {
            message: "query denied".to_owned(),
            details: serde_json::json!({"required_scope": "bifrost_query:read"}),
        });
        let projected = WyrdError::from(&transport);
        assert_eq!(projected.code(), "WYRD_PERMISSION_403_DENIED_RBAC");
        assert_eq!(projected.as_problem_json()["detail"], "query denied");
        assert_eq!(
            projected.as_problem_json()["details"],
            serde_json::json!({"required_scope": "bifrost_query:read"})
        );

        let mut terminal = failed_terminal(0);
        terminal.error = None;
        let failed = BifrostClientError::FailedTerminal { terminal };
        assert_eq!(
            WyrdError::from(&failed).code(),
            BifrostError::QueryExecutionFailed.code()
        );

        for (error, code) in [
            (
                BifrostClientError::Protocol("safe protocol detail".to_owned()),
                "WYRD_VALA_502_QUERY_STREAM_PROTOCOL",
            ),
            (
                BifrostClientError::Arrow("safe Arrow detail".to_owned()),
                "WYRD_VALA_502_QUERY_STREAM_PROTOCOL",
            ),
            (
                BifrostClientError::IncompleteQueryStream,
                "WYRD_VALA_502_QUERY_STREAM_INCOMPLETE",
            ),
            (
                BifrostClientError::ResultTooLarge,
                "WYRD_VALA_413_QUERY_RESULT_TOO_LARGE",
            ),
            (
                BifrostClientError::NoActiveTable,
                "WYRD_VALA_412_NO_ACTIVE_TABLE",
            ),
            (
                BifrostClientError::RowDeserialization("row does not fit".to_owned()),
                "WYRD_CLIENT_422_ROW_DESERIALIZATION",
            ),
            (
                BifrostClientError::Queue(WyrdQueueError::QueueFull),
                "WYRD_CLIENT_429_QUEUE_FULL",
            ),
            (
                BifrostClientError::Queue(WyrdQueueError::FlushTimeout),
                "WYRD_CLIENT_504_FLUSH_TIMEOUT",
            ),
            (
                BifrostClientError::Queue(WyrdQueueError::PayloadTooLarge),
                "WYRD_CLIENT_413_PAYLOAD_TOO_LARGE",
            ),
            (
                BifrostClientError::Client(WyrdClientError::NoCredentials),
                "WYRD_CLIENT_401_NO_CREDENTIALS",
            ),
        ] {
            let projected = WyrdError::from(&error);
            assert_eq!(projected.code(), code, "code for {error}");
            assert_ne!(
                projected.code(),
                "WYRD_SPEC_500_INTERNAL",
                "{error} degraded to the internal code"
            );
        }
    }

    /// A failed terminal's catalog problem rebuilds the exact typed error.
    ///
    /// # Panics
    ///
    /// Panics when any error loses its code, details, status, title, or
    /// remediation on the way back from the terminal.
    #[test]
    fn failed_terminal_problem_rebuilds_its_catalog_error() {
        let cases = [
            BifrostError::QueryTimeout,
            BifrostError::QueryVisibilityUnavailable,
            BifrostError::QueryTenantInvariant,
            BifrostError::CatalogUnreachable {
                detail: "source failed".to_owned(),
            },
            BifrostError::QueryResourcesExhausted,
            BifrostError::QueryExecutionFailed,
            BifrostError::VariantInvalidJson {
                field: "parse_json".to_owned(),
                row: 1,
                path: String::new(),
            },
            BifrostError::VariantNumericOutOfRange {
                field: "parse_json".to_owned(),
                row: 0,
                path: "/n".to_owned(),
                numeric_kind: "integer".to_owned(),
            },
        ];
        for expected in cases {
            let terminal = failed_terminal_with(expected.clone(), 0);
            let projected = WyrdError::from(&BifrostClientError::FailedTerminal { terminal });
            assert_eq!(
                projected.problem(),
                WyrdError::from(expected).problem(),
                "terminal problem rebuilds its catalog error"
            );
        }
    }

    /// Builds the schema used by client state-machine tests.
    fn test_schema() -> SchemaRef {
        Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]))
    }

    /// Test-owned encoder that emits the same split IPC stream the server emits.
    ///
    /// One writer opens the stream, each batch contributes only its own write
    /// delta, and [`TestQueryIpc::close`] yields the end-of-stream delta the
    /// terminal frame carries. Fixtures built this way exercise the stateful
    /// decoder exactly as a real query stream does.
    struct TestQueryIpc {
        /// Writer whose sink is drained once per emitted fragment.
        writer: StreamWriter<Vec<u8>>,
    }

    impl TestQueryIpc {
        /// Opens a stream over `schema` and returns its schema fragment.
        ///
        /// # Panics
        ///
        /// Panics when the test schema cannot start an Arrow IPC stream.
        fn open(schema: &SchemaRef) -> (Self, Vec<u8>) {
            let mut writer =
                StreamWriter::try_new(Vec::new(), schema.as_ref()).expect("schema writer starts");
            let prefix = std::mem::take(writer.get_mut());
            (Self { writer }, prefix)
        }

        /// Writes one `Int64` batch and returns only that write's fragment.
        ///
        /// # Panics
        ///
        /// Panics when the batch is invalid for `schema` or cannot be written.
        fn batch(&mut self, schema: &SchemaRef, values: &[i64]) -> Vec<u8> {
            let batch = RecordBatch::try_new(
                Arc::clone(schema),
                vec![Arc::new(Int64Array::from(values.to_vec()))],
            )
            .expect("test batch is valid");
            self.writer.write(&batch).expect("batch writes");
            std::mem::take(self.writer.get_mut())
        }

        /// Finishes the stream and returns its end-of-stream fragment.
        ///
        /// # Panics
        ///
        /// Panics when the writer cannot finish the stream.
        fn close(&mut self) -> Vec<u8> {
            self.writer.finish().expect("writer finishes");
            std::mem::take(self.writer.get_mut())
        }
    }

    /// Opens and immediately closes a stream, yielding its schema and EOS fragments.
    fn empty_ipc(schema: &SchemaRef) -> (Vec<u8>, Vec<u8>) {
        let (mut ipc, prefix) = TestQueryIpc::open(schema);
        let eos = ipc.close();
        (prefix, eos)
    }

    /// Builds the complete three-tier source set every terminal carries.
    fn sources() -> Vec<SourceCompletion> {
        vec![
            SourceCompletion {
                source: QuerySource::Iceberg,
                outcome: SourceCompletionOutcome::Complete,
            },
            SourceCompletion {
                source: QuerySource::HotSealed,
                outcome: SourceCompletionOutcome::Complete,
            },
            SourceCompletion {
                source: QuerySource::LiveTail,
                outcome: SourceCompletionOutcome::Complete,
            },
        ]
    }

    /// Builds a successful terminal for the supplied rows and EOS.
    fn success_terminal(rows: u64, arrow_ipc_eos: Vec<u8>) -> QueryTerminalFrame {
        QueryTerminalFrame {
            outcome: QueryTerminalOutcome::Success,
            query_class: wyrd_spec::vala::api::QueryClass::Interactive,
            row_count: rows,
            warnings: Vec::new(),
            source_completion: sources(),
            error: None,
            arrow_ipc_eos,
        }
    }

    /// Builds a degraded Fused terminal with the required warning and source state.
    fn degraded_terminal(rows: u64, arrow_ipc_eos: Vec<u8>) -> QueryTerminalFrame {
        QueryTerminalFrame {
            outcome: QueryTerminalOutcome::Degraded,
            query_class: wyrd_spec::vala::api::QueryClass::Interactive,
            row_count: rows,
            warnings: vec![QueryWarning::LiveTailUnavailable],
            source_completion: vec![
                SourceCompletion {
                    source: QuerySource::Iceberg,
                    outcome: SourceCompletionOutcome::Complete,
                },
                SourceCompletion {
                    source: QuerySource::HotSealed,
                    outcome: SourceCompletionOutcome::Complete,
                },
                SourceCompletion {
                    source: QuerySource::LiveTail,
                    outcome: SourceCompletionOutcome::Unavailable,
                },
            ],
            error: None,
            arrow_ipc_eos,
        }
    }

    /// Builds a generic failed terminal that retains the immutable cut metadata.
    fn failed_terminal(rows: u64) -> QueryTerminalFrame {
        failed_terminal_with(BifrostError::QueryExecutionFailed, rows)
    }

    /// Builds a failed terminal carrying `error`'s catalog problem.
    fn failed_terminal_with(error: BifrostError, rows: u64) -> QueryTerminalFrame {
        QueryTerminalFrame {
            outcome: QueryTerminalOutcome::Failed,
            query_class: wyrd_spec::vala::api::QueryClass::Interactive,
            row_count: rows,
            warnings: Vec::new(),
            source_completion: sources(),
            error: Some(Box::new(WyrdError::from(error).problem())),
            // A failed stream never calls `finish`, so it has no end-of-stream.
            arrow_ipc_eos: Vec::new(),
        }
    }

    /// Converts one domain frame into its length-delimited public protobuf bytes.
    fn encoded(frame: QueryStreamFrame) -> Vec<u8> {
        FrameEncoder::encode(&proto::QueryStreamFrame::from(frame)).expect("frame encodes")
    }

    /// Builds a result stream over arbitrary already-encoded response chunks.
    fn result_stream(chunks: Vec<Vec<u8>>) -> QueryResultStream {
        let body = stream::iter(chunks.into_iter().map(|chunk| Ok(Bytes::from(chunk))));
        QueryResultStream::new(
            RawQueryStream::new(body),
            RequestId::now_v7(),
            offline_client(),
            0,
        )
    }

    /// Builds a query client whose lifecycle routes are guaranteed unreachable.
    ///
    /// Tests that only exercise decoding never settle against a server, so the
    /// client they carry must be inert rather than absent: the stream's own
    /// contract is that it always holds one.
    fn offline_client() -> QueryClient {
        client_for("http://127.0.0.1:1")
    }

    /// Builds a query client bound to one base URL with a static credential.
    fn client_for(base_url: &str) -> QueryClient {
        client_with_timeout(base_url, 2_000)
    }

    /// Builds a query client for `base_url` with a configured transport
    /// timeout of `timeout_ms`, which also bounds each stream's close after
    /// its terminal.
    fn client_with_timeout(base_url: &str, timeout_ms: u64) -> QueryClient {
        let config = crate::config::ClientConfig {
            credential: Some(secrecy::SecretString::from("test-key")),
            http: crate::transport::config::HttpConfig {
                base_url: base_url.to_owned(),
                timeout_ms,
                ..crate::transport::config::HttpConfig::default()
            },
            ..crate::config::ClientConfig::default()
        };
        QueryClient::new(&WyrdClient::with_config(config).expect("static config builds a client"))
    }

    /// The converter rejects a second schema before any terminal can be accepted.
    #[test]
    fn bifrost_query_rejects_duplicate_schema() {
        let (prefix, _eos) = empty_ipc(&test_schema());
        let schema = QueryStreamFrame::Schema(QuerySchemaFrame {
            schema_fingerprint: "test".to_owned(),
            arrow_ipc_schema: prefix,
        });
        let proto = proto::QueryStreamFrame::from(schema);
        let mut converter = QueryStreamConverter::new();
        converter
            .convert(proto.clone(), None)
            .expect("schema accepted");
        assert!(converter.convert(proto, None).is_err());
    }

    /// Every byte boundary is accepted by the incremental protobuf decoder.
    #[test]
    fn bifrost_query_frame_decoder_accepts_arbitrary_chunk_splits() {
        let (prefix, _eos) = empty_ipc(&test_schema());
        let frame = QueryStreamFrame::Schema(QuerySchemaFrame {
            schema_fingerprint: "test".to_owned(),
            arrow_ipc_schema: prefix,
        });
        let encoded = encoded(frame);
        let mut decoder = FrameDecoder::new(MAX_FRAME_BYTES);
        let mut decoded = Vec::new();
        for byte in encoded {
            decoded.extend(
                decoder
                    .push::<proto::QueryStreamFrame>(&[byte])
                    .expect("split frame decodes"),
            );
        }
        decoder.finish().expect("complete frame");
        assert_eq!(decoded.len(), 1);
    }

    /// EOF without a terminal is always incomplete, never successful empty output.
    #[tokio::test]
    async fn bifrost_query_eof_before_terminal_is_rejected() {
        let body = stream::empty::<Result<Bytes, reqwest::Error>>();
        let mut stream = RawQueryStream::new(body);
        assert!(matches!(
            stream.next_frame().await,
            Err(BifrostClientError::IncompleteQueryStream)
        ));
    }

    /// Cancelling provisional-success EOF validation preserves stream-owned
    /// metadata for both incremental reads and subsequent bounded collection.
    ///
    /// # Panics
    /// Panics if cancellation loses terminal metadata or exposes success before EOF.
    #[tokio::test]
    async fn cancelled_terminal_eof_wait_resumes_safely() {
        for case in ["incremental", "collect", "close timeout", "late frame"] {
            let schema = test_schema();
            let (mut ipc, prefix) = TestQueryIpc::open(&schema);
            let batch = ipc.batch(&schema, &[1, 2]);
            let terminal = success_terminal(2, ipc.close());
            let (sender, mut receiver) = tokio::sync::mpsc::channel(3);
            for frame in [
                QueryStreamFrame::Schema(QuerySchemaFrame {
                    schema_fingerprint: String::new(),
                    arrow_ipc_schema: prefix,
                }),
                QueryStreamFrame::Batch(QueryBatchFrame {
                    arrow_ipc_batch: batch,
                }),
                QueryStreamFrame::Terminal(terminal.clone()),
            ] {
                sender
                    .try_send(Ok(Bytes::from(encoded(frame))))
                    .expect("frame fits");
            }
            let eof_polled = Arc::new(tokio::sync::Notify::new());
            let notify = Arc::clone(&eof_polled);
            let body = stream::poll_fn(move |cx| {
                let polled = receiver.poll_recv(cx);
                if polled.is_pending() {
                    notify.notify_one();
                }
                polled
            });
            let mut result = QueryResultStream::new(
                RawQueryStream::new(body),
                RequestId::now_v7(),
                if case == "close timeout" {
                    client_with_timeout("http://127.0.0.1:1", 200)
                } else {
                    offline_client()
                },
                deadline_in(30_000),
            );
            assert_eq!(
                result
                    .next_batch()
                    .await
                    .expect("batch decodes")
                    .expect("batch exists")
                    .num_rows(),
                2
            );
            {
                let read = result.next_batch();
                tokio::pin!(read);
                tokio::select! {
                    biased;
                    value = &mut read => panic!("EOF must remain gated: {value:?}"),
                    () = eof_polled.notified() => {},
                }
            }
            assert!(result.terminal().is_none());
            assert_eq!(result.settlement, StreamSettlement::Healthy);
            if case == "close timeout" {
                let resumed =
                    tokio::time::timeout(std::time::Duration::from_secs(1), result.next_batch())
                        .await
                        .expect("resumption remains bounded by the transport timeout");
                assert!(matches!(
                    resumed,
                    Err(BifrostClientError::IncompleteQueryStream)
                ));
                assert_eq!(result.settlement, StreamSettlement::Broken);
                assert!(result.terminal().is_none());
                continue;
            }
            if case == "late frame" {
                sender
                    .try_send(Ok(Bytes::from(encoded(QueryStreamFrame::Terminal(
                        terminal.clone(),
                    )))))
                    .expect("late frame fits");
                assert!(matches!(
                    result.next_batch().await,
                    Err(BifrostClientError::Protocol(_))
                ));
                assert_eq!(result.settlement, StreamSettlement::Broken);
                assert!(result.terminal().is_none());
                continue;
            }
            drop(sender);
            if case == "collect" {
                let collected = result
                    .collect_bounded(CollectedQueryLimits {
                        max_rows: 10,
                        max_encoded_bytes: 1024 * 1024,
                    })
                    .await
                    .expect("resumed collection retains completion");
                assert_eq!(collected.terminal, terminal);
            } else {
                assert!(result.next_batch().await.expect("EOF validates").is_none());
                assert_eq!(result.terminal(), Some(&terminal));
                assert_eq!(result.settlement, StreamSettlement::Settled);
            }
        }
    }

    /// Opens a stream whose deadline has passed and whose body has already
    /// delivered a schema and a typed timeout terminal, as a server that ended
    /// the query at its deadline sends them.
    ///
    /// The returned sender holds the body open, so the caller decides whether
    /// EOF, a trailing frame, or nothing at all follows the terminal. The
    /// client's configured transport timeout is `timeout_ms`.
    ///
    /// # Panics
    /// Panics if the two frames do not fit the body channel.
    fn deadline_terminal_stream(
        timeout_ms: u64,
    ) -> (
        QueryResultStream,
        QueryTerminalFrame,
        tokio::sync::mpsc::Sender<Result<Bytes, reqwest::Error>>,
    ) {
        let (prefix, _eos) = empty_ipc(&test_schema());
        let terminal = failed_terminal_with(BifrostError::QueryTimeout, 0);
        let (sender, receiver) = tokio::sync::mpsc::channel(4);
        for frame in [
            QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: String::new(),
                arrow_ipc_schema: prefix,
            }),
            QueryStreamFrame::Terminal(terminal.clone()),
        ] {
            sender
                .try_send(Ok(Bytes::from(encoded(frame))))
                .expect("frame fits");
        }
        let result = QueryResultStream::new(
            RawQueryStream::new(tokio_stream::wrappers::ReceiverStream::new(receiver)),
            RequestId::now_v7(),
            client_with_timeout("http://127.0.0.1:1", timeout_ms),
            deadline_in(-1),
        );
        (result, terminal, sender)
    }

    /// A timeout terminal the server sent at the deadline survives a late EOF.
    ///
    /// The server can only emit its typed timeout once the deadline has
    /// passed, so the terminal always arrives with no deadline left. Its EOF
    /// lands in a later body read here, which is what a loaded host produces;
    /// the caller must still see the typed 504 timeout, never an incomplete
    /// stream, because the deadline bounds terminal production, not the close.
    ///
    /// # Panics
    /// Panics if the late EOF turns the timeout terminal into any other error.
    #[tokio::test]
    async fn deadline_terminal_with_late_eof_keeps_the_typed_timeout() {
        let (mut result, terminal, sender) = deadline_terminal_stream(2_000);
        let eof = tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            drop(sender);
        });
        let error = result
            .next_batch()
            .await
            .expect_err("a timeout terminal is a failure");
        eof.await.expect("EOF task completes");
        let BifrostClientError::FailedTerminal { terminal: retained } = &error else {
            panic!("the late EOF must keep the typed terminal: {error:?}");
        };
        assert_eq!(retained, &terminal);
        assert_eq!(
            WyrdError::from(&error).code(),
            WyrdError::from(BifrostError::QueryTimeout).code()
        );
        assert_eq!(result.terminal(), Some(&terminal));
        assert_eq!(result.settlement, StreamSettlement::Settled);
    }

    /// A frame after a deadline terminal is still a protocol violation.
    ///
    /// Moving the close off the query deadline must not loosen the
    /// exactly-one-terminal rule: anything but EOF behind the terminal breaks
    /// the stream and is never accepted as the server's failure.
    ///
    /// # Panics
    /// Panics if a trailing frame yields anything but a protocol error.
    #[tokio::test]
    async fn deadline_terminal_followed_by_a_frame_is_a_protocol_error() {
        let (mut result, terminal, sender) = deadline_terminal_stream(2_000);
        sender
            .try_send(Ok(Bytes::from(encoded(QueryStreamFrame::Terminal(
                terminal,
            )))))
            .expect("trailing frame fits");
        let error = result
            .next_batch()
            .await
            .expect_err("a trailing frame never validates");
        assert!(
            matches!(error, BifrostClientError::Protocol(_)),
            "{error:?}"
        );
        assert!(result.terminal().is_none());
        assert_eq!(result.settlement, StreamSettlement::Broken);
    }

    /// A connection that never closes after its terminal fails at the
    /// configured transport timeout as an incomplete stream.
    ///
    /// # Panics
    /// Panics if the hang is not bounded by the transport timeout or is
    /// accepted as the terminal.
    #[tokio::test]
    async fn terminal_without_eof_fails_at_the_transport_timeout() {
        let (mut result, _terminal, sender) = deadline_terminal_stream(100);
        let error = tokio::time::timeout(std::time::Duration::from_secs(5), result.next_batch())
            .await
            .expect("the close is bounded by the transport timeout")
            .expect_err("an unclosed body never validates its terminal");
        drop(sender);
        assert!(
            matches!(error, BifrostClientError::IncompleteQueryStream),
            "{error:?}"
        );
        assert!(result.terminal().is_none());
        assert_eq!(result.settlement, StreamSettlement::Broken);
    }

    /// Schema, one batch, and terminal decode incrementally and retain metadata.
    #[tokio::test]
    async fn bifrost_query_decodes_arrow_and_terminal() {
        let schema = test_schema();
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let batch = ipc.batch(&schema, &[1, 2]);
        let eos = ipc.close();
        let chunks = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: batch,
            })),
            encoded(QueryStreamFrame::Terminal(success_terminal(2, eos))),
        ];
        let mut result = result_stream(chunks);
        let batch = result
            .next_batch()
            .await
            .expect("batch decodes")
            .expect("batch present");
        assert_eq!(batch.num_rows(), 2);
        assert!(
            result
                .next_batch()
                .await
                .expect("terminal validates")
                .is_none()
        );
        assert_eq!(result.terminal().expect("terminal retained").row_count, 2);
    }

    /// Degraded is a successful completion with explicit retained terminal state.
    #[tokio::test]
    async fn bifrost_query_degraded_terminal_completes() {
        let (prefix, eos) = empty_ipc(&test_schema());
        let chunks = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Terminal(degraded_terminal(0, eos))),
        ];
        let mut result = result_stream(chunks);
        assert!(
            result
                .next_batch()
                .await
                .expect("degraded is success")
                .is_none()
        );
        assert_eq!(
            result.terminal().expect("terminal retained").outcome,
            QueryTerminalOutcome::Degraded
        );
    }

    /// Bounded zero-row collection retains the authoritative schema without a batch.
    #[tokio::test]
    async fn bifrost_query_zero_rows_retains_authoritative_schema() {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Int64, false),
            Field::new("optional_value", DataType::Utf8, true),
        ]));
        let (prefix, eos) = empty_ipc(&schema);
        let chunks = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "zero-row-fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Terminal(success_terminal(0, eos))),
        ];

        let result = result_stream(chunks)
            .collect_bounded(CollectedQueryLimits {
                max_rows: 1,
                max_encoded_bytes: usize::MAX,
            })
            .await
            .expect("schema and zero-row terminal collect");

        assert!(result.batches.is_empty());
        assert_eq!(result.rows, 0);
        assert_eq!(result.terminal.row_count, 0);
        assert_eq!(result.schema.fields(), schema.fields());
        assert_eq!(result.schema.field(0).name(), "id");
        assert_eq!(result.schema.field(0).data_type(), &DataType::Int64);
        assert!(!result.schema.field(0).is_nullable());
        assert_eq!(result.schema.field(1).name(), "optional_value");
        assert_eq!(result.schema.field(1).data_type(), &DataType::Utf8);
        assert!(result.schema.field(1).is_nullable());
    }

    /// A successful terminal without the required schema never becomes a result.
    #[tokio::test]
    async fn bifrost_query_collection_rejects_missing_schema() {
        let (_prefix, eos) = empty_ipc(&test_schema());
        let chunks = vec![encoded(QueryStreamFrame::Terminal(success_terminal(
            0, eos,
        )))];

        assert!(
            result_stream(chunks)
                .collect_bounded(CollectedQueryLimits {
                    max_rows: 1,
                    max_encoded_bytes: usize::MAX,
                })
                .await
                .is_err()
        );
    }

    /// A failed terminal errors after retaining exact diagnostic metadata.
    #[tokio::test]
    async fn bifrost_query_failed_terminal_is_retained_and_rejected() {
        let (prefix, _eos) = empty_ipc(&test_schema());
        let chunks = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Terminal(failed_terminal(0))),
        ];
        let mut result = result_stream(chunks);
        let error = result
            .next_batch()
            .await
            .expect_err("failed terminal rejects");
        assert!(matches!(error, BifrostClientError::FailedTerminal { .. }));
        assert_eq!(
            result.terminal().expect("failed terminal retained").outcome,
            QueryTerminalOutcome::Failed
        );
        assert_eq!(result.settlement, StreamSettlement::Settled);
    }

    /// A resource-exhausted terminal after delivered rows rejects the whole
    /// result as the stable typed error; no partial rows are returned.
    ///
    /// # Panics
    ///
    /// Panics when collection returns rows or the error loses its stable code.
    #[tokio::test]
    async fn resource_terminal_rejects_partial_rows() {
        let schema = test_schema();
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let terminal = failed_terminal_with(BifrostError::QueryResourcesExhausted, 1);
        let chunks = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: ipc.batch(&schema, &[1]),
            })),
            encoded(QueryStreamFrame::Terminal(terminal)),
        ];
        let error = result_stream(chunks)
            .collect_bounded(CollectedQueryLimits {
                max_rows: usize::MAX,
                max_encoded_bytes: usize::MAX,
            })
            .await
            .expect_err("a resource terminal rejects rows already received");
        assert_eq!(
            WyrdError::from(&error).code(),
            "WYRD_VALA_503_QUERY_RESOURCES_EXHAUSTED"
        );
    }

    /// A failed terminal is a validated failure only when clean EOF follows it.
    ///
    /// A duplicate terminal, a trailing batch, or a late schema after a failed
    /// terminal breaks the exactly-one-terminal contract: the stream must not
    /// retain the terminal, must report a protocol or Arrow failure instead of
    /// the server's failure, and must owe exactly one settlement.
    ///
    /// # Panics
    ///
    /// Panics if any trailing frame yields a retained failed terminal or
    /// settlement is not owed exactly once.
    #[tokio::test]
    async fn failed_terminal_requires_clean_eof() {
        for trailing in ["duplicate terminal", "batch", "schema"] {
            let schema = test_schema();
            let (mut ipc, prefix) = TestQueryIpc::open(&schema);
            let batch = ipc.batch(&schema, &[1]);
            let schema_frame = || {
                QueryStreamFrame::Schema(QuerySchemaFrame {
                    schema_fingerprint: "fingerprint".to_owned(),
                    arrow_ipc_schema: prefix.clone(),
                })
            };
            let follower = match trailing {
                "duplicate terminal" => QueryStreamFrame::Terminal(failed_terminal(0)),
                "batch" => QueryStreamFrame::Batch(QueryBatchFrame {
                    arrow_ipc_batch: batch,
                }),
                _ => schema_frame(),
            };
            let chunks = vec![
                encoded(schema_frame()),
                encoded(QueryStreamFrame::Terminal(failed_terminal(0))),
                encoded(follower),
            ];
            let mut result = result_stream(chunks);
            let error = result
                .next_batch()
                .await
                .expect_err("a trailing frame never yields success");
            assert!(
                matches!(
                    error,
                    BifrostClientError::Protocol(_) | BifrostClientError::Arrow(_)
                ),
                "{trailing}: {error:?}"
            );
            assert!(result.terminal().is_none(), "{trailing}");
            assert_eq!(result.settlement, StreamSettlement::Broken, "{trailing}");
            result.settle().await;
            assert_eq!(result.settlement, StreamSettlement::Settled, "{trailing}");
        }
    }

    /// One transport chunk carrying schema, many batches, and terminal is
    /// decoded one frame per delivery.
    ///
    /// The pending queue holds only encoded wire frames, and the Arrow decoder
    /// has consumed exactly the fragments already yielded, so no more than one
    /// decoded batch is ever owned ahead of the caller. Order, row accounting,
    /// and terminal validation are unchanged.
    ///
    /// # Panics
    ///
    /// Panics if any batch is decoded before its delivery, rows reorder, or the
    /// terminal is not retained.
    #[tokio::test]
    async fn coalesced_chunk_decodes_one_batch_per_delivery() {
        let schema = test_schema();
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let values: Vec<i64> = (1..=16).collect();
        let fragments: Vec<Vec<u8>> = values
            .iter()
            .map(|value| ipc.batch(&schema, &[*value]))
            .collect();
        let eos = ipc.close();
        let mut body = encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
            schema_fingerprint: "fingerprint".to_owned(),
            arrow_ipc_schema: prefix.clone(),
        }));
        for fragment in &fragments {
            body.extend(encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: fragment.clone(),
            })));
        }
        body.extend(encoded(QueryStreamFrame::Terminal(success_terminal(
            16, eos,
        ))));

        let mut result = result_stream(vec![body]);
        let mut rows = Vec::new();
        let mut consumed = prefix.len();
        for fragment in &fragments {
            let batch = result
                .next_batch()
                .await
                .expect("coalesced frame decodes")
                .expect("batch present");
            consumed += fragment.len();
            let pending: &VecDeque<WireQueryStreamFrame> = &result.raw.pending;
            assert!(
                pending
                    .iter()
                    .all(|frame| !matches!(frame.frame, Some(WireFrame::Schema(_)))),
                "only undelivered frames remain pending"
            );
            assert_eq!(
                result.raw.arrow_ipc_bytes(),
                consumed,
                "only delivered fragments were decoded"
            );
            let column = batch
                .column(0)
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("int64 column");
            rows.extend(column.values().iter().copied());
        }
        assert!(
            result
                .next_batch()
                .await
                .expect("terminal validates")
                .is_none()
        );
        assert_eq!(rows, values);
        assert_eq!(result.terminal().expect("terminal retained").row_count, 16);
    }

    /// A standalone per-batch stream is refused; there is no compatibility mode.
    ///
    /// The public protocol is one split IPC stream, so a batch frame carries a
    /// bare write delta. A legacy frame that re-encodes its own schema message
    /// is a second schema on the same stream and must be rejected rather than
    /// silently reinterpreted under the initial schema.
    #[tokio::test]
    async fn bifrost_query_rejects_standalone_batch_stream() {
        let initial = test_schema();
        let other = Arc::new(Schema::new(vec![Field::new(
            "other",
            DataType::Int64,
            false,
        )]));
        let (_initial_ipc, prefix) = TestQueryIpc::open(&initial);
        let mut standalone = Vec::new();
        let mut writer = StreamWriter::try_new(&mut standalone, other.as_ref())
            .expect("standalone writer starts");
        writer
            .write(
                &RecordBatch::try_new(
                    Arc::clone(&other),
                    vec![Arc::new(Int64Array::from(vec![1]))],
                )
                .expect("standalone batch is valid"),
            )
            .expect("standalone batch writes");
        writer.finish().expect("standalone writer finishes");
        drop(writer);
        let chunks = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: standalone,
            })),
        ];
        let mut result = result_stream(chunks);
        assert!(matches!(
            result.next_batch().await,
            Err(BifrostClientError::Arrow(_))
        ));
    }

    /// The client decodes one split IPC stream per query with an explicit close.
    ///
    /// This is the client half of the stateful protocol: one schema fragment,
    /// one bare delta per batch, and one end-of-stream delta carried by the
    /// terminal. It proves ordering, closure, per-fragment bounded retention,
    /// and that the fragments are strictly smaller than standalone re-encodes.
    #[tokio::test]
    async fn stateful_ipc_decoder_contract() {
        let schema = test_schema();
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let batches = [
            ipc.batch(&schema, &[1, 2]),
            ipc.batch(&schema, &[3]),
            ipc.batch(&schema, &[4, 5, 6]),
        ];
        let eos = ipc.close();

        let mut standalone = Vec::new();
        let mut writer =
            StreamWriter::try_new(&mut standalone, schema.as_ref()).expect("standalone starts");
        writer
            .write(
                &RecordBatch::try_new(
                    Arc::clone(&schema),
                    vec![Arc::new(Int64Array::from(vec![1, 2]))],
                )
                .expect("standalone batch is valid"),
            )
            .expect("standalone batch writes");
        writer.finish().expect("standalone finishes");
        drop(writer);
        assert!(
            batches[0].len() < standalone.len(),
            "a continuation fragment must be smaller than a standalone stream"
        );

        let mut chunks = vec![encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
            schema_fingerprint: "fingerprint".to_owned(),
            arrow_ipc_schema: prefix,
        }))];
        chunks.extend(batches.iter().map(|fragment| {
            encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: fragment.clone(),
            }))
        }));
        chunks.push(encoded(QueryStreamFrame::Terminal(success_terminal(
            6,
            eos.clone(),
        ))));

        let mut result = result_stream(chunks);
        let mut rows = Vec::new();
        while let Some(batch) = result.next_batch().await.expect("split stream decodes") {
            assert_eq!(batch.schema().as_ref(), schema.as_ref());
            let column = batch
                .column(0)
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("int64 column");
            rows.extend(column.values().iter().copied());
        }
        assert_eq!(rows, vec![1, 2, 3, 4, 5, 6]);
        assert_eq!(result.terminal().expect("terminal retained").row_count, 6);
        assert!(
            result.raw.arrow_ipc_closed(),
            "terminal EOS closes the stream"
        );
        let largest = batches
            .iter()
            .map(Vec::len)
            .max()
            .expect("fixture has batches");
        assert!(
            result.raw.peak_pending_frame_bytes() <= largest,
            "the decoder retains at most one fragment at a time"
        );

        let (prefix, _unused) = empty_ipc(&schema);
        let missing_eos = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Terminal(QueryTerminalFrame {
                arrow_ipc_eos: Vec::new(),
                ..success_terminal(0, eos.clone())
            })),
        ];
        assert!(matches!(
            result_stream(missing_eos).next_batch().await,
            Err(BifrostClientError::Protocol(_)),
        ));

        let orphan_batch = vec![encoded(QueryStreamFrame::Batch(QueryBatchFrame {
            arrow_ipc_batch: batches[0].clone(),
        }))];
        assert!(matches!(
            result_stream(orphan_batch).next_batch().await,
            Err(BifrostClientError::Protocol(_))
        ));

        let (prefix, closing) = empty_ipc(&schema);
        let mut decoder = QueryIpcDecoder::new();
        decoder.accept_schema(&prefix).expect("schema accepted");
        assert!(
            decoder.accept_schema(&prefix).is_err(),
            "a stream carries exactly one schema"
        );
        decoder
            .accept_eos(&closing)
            .expect("end-of-stream accepted");
        assert!(decoder.eos_accepted());
        assert!(
            decoder.accept_batch(&batches[0]).is_err(),
            "no fragment may follow the end-of-stream"
        );
        assert!(
            decoder.accept_eos(&closing).is_err(),
            "a stream closes exactly once"
        );
    }

    /// Row and encoded-byte bounds reject the whole collection without truncation.
    #[tokio::test]
    async fn bifrost_query_bounded_collection_rejects_rows_and_bytes() {
        let schema = test_schema();
        let (mut ipc, prefix) = TestQueryIpc::open(&schema);
        let batch = ipc.batch(&schema, &[1, 2]);
        let eos = ipc.close();
        let chunks = vec![
            encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
                schema_fingerprint: "fingerprint".to_owned(),
                arrow_ipc_schema: prefix,
            })),
            encoded(QueryStreamFrame::Batch(QueryBatchFrame {
                arrow_ipc_batch: batch.clone(),
            })),
            encoded(QueryStreamFrame::Terminal(success_terminal(2, eos))),
        ];
        let rows = result_stream(chunks.clone())
            .collect_bounded(CollectedQueryLimits {
                max_rows: 1,
                max_encoded_bytes: usize::MAX,
            })
            .await;
        assert!(matches!(rows, Err(BifrostClientError::ResultTooLarge)));

        let bytes = result_stream(chunks)
            .collect_bounded(CollectedQueryLimits {
                max_rows: usize::MAX,
                max_encoded_bytes: batch.len().saturating_sub(1),
            })
            .await;
        assert!(matches!(bytes, Err(BifrostClientError::ResultTooLarge)));
    }

    /// Schema and framing bytes count toward the ceiling even with no result batches.
    #[tokio::test]
    async fn bifrost_query_bounded_collection_counts_schema_and_framing_bytes() {
        let fields = (0..128)
            .map(|index| {
                Field::new(
                    format!("wide_schema_field_{index:03}_with_descriptive_name"),
                    DataType::Int64,
                    true,
                )
            })
            .collect::<Vec<_>>();
        let schema = Arc::new(Schema::new(fields));
        let (prefix, eos) = empty_ipc(&schema);
        let schema_frame = encoded(QueryStreamFrame::Schema(QuerySchemaFrame {
            schema_fingerprint: "wide-schema".to_owned(),
            arrow_ipc_schema: prefix,
        }));
        let terminal_frame = encoded(QueryStreamFrame::Terminal(success_terminal(0, eos)));
        let encoded_bytes = schema_frame
            .len()
            .checked_add(terminal_frame.len())
            .expect("test response byte count fits usize");
        let chunks = vec![schema_frame, terminal_frame];

        let rejected = result_stream(chunks.clone())
            .collect_bounded(CollectedQueryLimits {
                max_rows: 0,
                max_encoded_bytes: encoded_bytes.saturating_sub(1),
            })
            .await;
        assert!(matches!(rejected, Err(BifrostClientError::ResultTooLarge)));

        let accepted = result_stream(chunks)
            .collect_bounded(CollectedQueryLimits {
                max_rows: 0,
                max_encoded_bytes: encoded_bytes,
            })
            .await
            .expect("exact response-byte ceiling is accepted");
        assert_eq!(accepted.encoded_bytes, encoded_bytes);
        assert!(accepted.batches.is_empty());
    }

    /// Pending response-body stream whose drop flag proves cancellation ownership.
    struct DropTrackedStream {
        /// Flag set when the response stream is dropped.
        dropped: Arc<AtomicBool>,
    }

    impl Stream for DropTrackedStream {
        type Item = Result<Bytes, reqwest::Error>;

        /// Remains pending so cancellation is driven only by dropping the owner.
        fn poll_next(self: Pin<&mut Self>, _context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
            Poll::Pending
        }
    }

    impl Drop for DropTrackedStream {
        /// Records that the raw query released its response body.
        fn drop(&mut self) {
            self.dropped.store(true, Ordering::SeqCst);
        }
    }

    /// Dropping a raw stream releases the underlying body and propagates cancellation.
    #[test]
    fn bifrost_query_drop_releases_response_stream() {
        let dropped = Arc::new(AtomicBool::new(false));
        let raw = RawQueryStream::new(DropTrackedStream {
            dropped: Arc::clone(&dropped),
        });
        drop(raw);
        assert!(dropped.load(Ordering::SeqCst));
    }
}