wyrd-client 0.3.0

//! The one Bifrost client: query any authorized table, write to the active one.
//!
//! Reads and writes already share one credential, one [`WyrdClient`], and one
//! connection pool, so they share one client type here too. What differs is
//! binding: `POST /v1/query` accepts SQL over any table the caller may see,
//! while an ingest batch carries exactly one table and one Arrow schema. That
//! asymmetry is the server's, and it is why [`Bifrost::sql`] takes a query and
//! [`Bifrost::insert`] takes only a row.

use std::collections::{HashMap, HashSet};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use crate::WyrdClient;
#[cfg(feature = "internal")]
use crate::config::ClientConfig;
#[cfg(feature = "internal")]
use crate::environment::Environment;
#[cfg(feature = "internal")]
use crate::error::WyrdClientError;
#[cfg(feature = "internal")]
use crate::global_config::GlobalConfig;
use arrow::datatypes::SchemaRef;
use arrow::json::WriterBuilder;
use arrow::json::writer::JsonArray;
use arrow::record_batch::RecordBatch;
use serde::de::DeserializeOwned;
use tokio::sync::Mutex as AsyncMutex;
use wyrd_queue::QueueConfig;
use wyrd_queue::variant::WyrdJsonEncoderFactory;
#[cfg(feature = "internal")]
use wyrd_queue::{BatchSink, ClientByteGuard, DurableBatchAck, SealedBatch, SinkError};
use wyrd_spec::error::WyrdError;
use wyrd_spec::request_id::RequestId;
use wyrd_spec::vala::api::{
    BifrostQueryRequest, BifrostTableDescription, BifrostTableEntry, CancelRunningQueryResponse,
    QueryParam, QueryTerminalFrame, RegisterOutcome, RegisterTableResponse, RunningQuerySummary,
};
use wyrd_spec::vala::ids::TraceId;
use wyrd_spec::vala::trace::SpanRecord;

#[cfg(feature = "internal")]
use crate::bifrost::BifrostMetrics;
use crate::bifrost::grpc::BifrostGrpcTransport;
use crate::bifrost::handle::WriterPool;
use crate::bifrost::query::{BifrostClientError, QueryClient, QueryResultStream};
#[cfg(feature = "internal")]
use crate::bifrost::query::{CollectedQueryLimits, CollectedQueryResult};
use crate::bifrost::rows::{SPAN_COLUMNS, span_record};
use crate::bifrost::scope::ClientScope;
use crate::bifrost::sink::BifrostIngestSink;
use crate::bifrost::table::{Correlation, TableConfig, WriterTable};

/// The catalog code a query of a tenant's span table reports before its
/// first span export creates the table.
const SPANS_NOT_CREATED: &str = "WYRD_VALA_404_BIFROST_TABLE_NOT_FOUND";

/// Construction options for [`Bifrost::connect`] and
/// [`Bifrost::connect_with`].
///
/// The default is a query-only client with the default producer queue; set
/// `table` to bind an initial write target, or `queue` to tune the producer
/// (for example its handle-wide byte budget).
#[derive(Debug, Default)]
pub struct BifrostOptions {
    /// The initial write target, or `None` to bind one later with
    /// [`Bifrost::use_table`].
    pub table: Option<TableConfig>,
    /// Producer tuning, validated before any IO.
    pub queue: QueueConfig,
}

/// The one Bifrost client: query any authorized table, write to the active one.
///
/// Writes are bound because an ingest batch carries exactly one table and one
/// Arrow schema; switching targets is [`Bifrost::use_table`], and a
/// swapped-away table's pooled producer keeps draining rather than being
/// discarded, so no buffered row is stranded by a swap.
pub struct Bifrost {
    /// The read plane, sharing the client's auth and HTTP connection pools.
    query: QueryClient,
    /// The write plane: one pooled producer per table this client has written.
    ///
    /// Shared because the blocking drains run on a worker thread, which needs
    /// an owned handle rather than a borrow of this client.
    writer: Arc<WriterPool>,
    /// The table [`Bifrost::insert`] enqueues into, if one is bound.
    active: Mutex<Option<TableConfig>>,
    /// Described user schemas, keyed by fully-qualified table name.
    ///
    /// Populated by [`Bifrost::writer_table`] and never evicted: one schema is
    /// authoritative per table name for this connected writer's lifetime, so a
    /// describe is paid once per table rather than once per observation.
    described: Mutex<HashMap<Arc<str>, WriterTable>>,
    /// Serializes cache-miss describes so racing first uses of one table pay
    /// one describe (and one authorization decision), not one each.
    ///
    /// ponytail: one gate for every table, so first misses of different
    /// tables also queue behind each other; per-table gating only if distinct
    /// first uses ever contend on a hot path.
    describe_gate: AsyncMutex<()>,
}

impl Bifrost {
    /// Connect a client through the ambient [`WyrdClient`].
    ///
    /// Endpoints come from the global config file, then `WYRD_GRPC_URL` /
    /// `WYRD_SERVER_URL`, then the compiled defaults, and the credential from
    /// the `ClientConfig::resolve_credential` chain, exactly as
    /// [`WyrdClient::from_global`] resolves them. Then this is
    /// [`Bifrost::connect_with`] over that client.
    ///
    /// # Arguments
    /// * `options` - The initial write target and producer tuning;
    ///   [`BifrostOptions::default`] is a query-only client with the default
    ///   queue.
    ///
    /// # Errors
    ///
    /// Returns a configuration error when the global config file cannot be
    /// read or parsed, the stable no-credentials error when the chain yields
    /// nothing, and otherwise as [`Bifrost::connect_with`].
    pub async fn connect(options: BifrostOptions) -> Result<Self, BifrostClientError> {
        let client = WyrdClient::from_global().map_err(BifrostClientError::from)?;
        Self::connect_with(&client, options).await
    }

    /// Connect a client as `client`'s principal, used as given.
    ///
    /// Completes any credential exchange `client` requires, then dials the
    /// ingest channel; both planes share `client`'s credential, connection
    /// pools, and token cache. No authentication probe is added, so a later
    /// operation may still be refused.
    ///
    /// # Arguments
    /// * `client` - The client both planes act as, such as a delegated client
    ///   from [`WyrdClient::on_behalf_of`].
    /// * `options` - The initial write target and producer tuning.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Queue`] carrying
    /// `WYRD_CLIENT_400_CONFIG_INVALID` before any IO when the queue tuning
    /// fails [`QueueConfig::validate`], the server's stable error when a
    /// required exchange is refused or unreachable, and a transport error when
    /// the ingest channel cannot be dialled.
    pub async fn connect_with(
        client: &WyrdClient,
        options: BifrostOptions,
    ) -> Result<Self, BifrostClientError> {
        let BifrostOptions { table, queue } = options;
        queue.validate().map_err(BifrostClientError::Queue)?;
        let client = client.connected().await?;
        Self::assemble(&client, table, queue).await
    }

    /// Build a client over a caller-supplied batch sink.
    ///
    /// The seam that was `WriterPool::new` before the facade existed: it
    /// performs no IO, so a `wyrd-queue` mock or stall sink stands in for the
    /// gRPC transport while the pooling, backpressure, and drop-counting
    /// behavior under test stays the production one.
    ///
    /// # Arguments
    /// * `client` - The authenticated client the query plane and producer scope use.
    /// * `table` - The initial write target, or `None` to bind one later.
    /// * `sink` - The batch sink every pooled producer seals batches into.
    /// * `config` - Producer tuning; not validated by this constructor.
    #[must_use]
    #[cfg(feature = "internal")]
    pub fn with_sink(
        client: &WyrdClient,
        table: Option<TableConfig>,
        sink: Arc<dyn BatchSink<ClientByteGuard>>,
        config: QueueConfig,
    ) -> Self {
        Self {
            query: QueryClient::new(client),
            writer: Arc::new(WriterPool::new(
                ClientScope::from_client(client),
                sink,
                config,
            )),
            active: Mutex::new(table),
            described: Mutex::new(HashMap::new()),
            describe_gate: AsyncMutex::new(()),
        }
    }

    /// Build a read-only client that never dials the ingest channel.
    ///
    /// For callers that only query, describe, or manage query lifecycle — the
    /// `wyrd query` command, for example — and must not require a reachable
    /// gRPC endpoint. It performs no IO. Any write reaches a sink that refuses
    /// it with a stable validation error instead of sending a batch.
    ///
    /// # Arguments
    /// * `client` - The authenticated client the query plane uses.
    #[must_use]
    #[cfg(feature = "internal")]
    pub fn query_only(client: &WyrdClient) -> Self {
        Self::with_sink(
            client,
            None,
            Arc::new(QueryOnlySink),
            QueueConfig::default(),
        )
    }

    /// Dial the ingest channel and assemble both planes over one client.
    ///
    /// The transport is the single owner of a batch's attempt budget, so the
    /// producer's send deadline is raised past that whole budget; a shorter
    /// deadline would cancel the transport and restart its attempts later.
    /// `config` is already validated by the caller. Crate-internal transport tests call this directly so a
    /// counted or blocked token mint stays on the send path under test.
    ///
    /// # Errors
    ///
    /// Returns a transport error when the ingest channel cannot be dialled.
    pub(crate) async fn assemble(
        client: &WyrdClient,
        table: Option<TableConfig>,
        mut config: QueueConfig,
    ) -> Result<Self, BifrostClientError> {
        let transport = BifrostGrpcTransport::connect(client)
            .await
            .map_err(BifrostClientError::from)?;
        let deadline_ms = u64::try_from(transport.send_deadline().as_millis()).unwrap_or(u64::MAX);
        config.flush_timeout_ms = config.flush_timeout_ms.max(deadline_ms);
        Ok(Self {
            query: QueryClient::new(client),
            writer: Arc::new(WriterPool::new(
                ClientScope::from_client(client),
                Arc::new(BifrostIngestSink {
                    transport: Arc::new(transport),
                }),
                config,
            )),
            active: Mutex::new(table),
            described: Mutex::new(HashMap::new()),
            describe_gate: AsyncMutex::new(()),
        })
    }

    // ── table lifecycle ────────────────────────────────────────────────────

    /// Create the active table, returning whether it was created or matched.
    ///
    /// Idempotent: `POST /v1/bifrost/tables` answers
    /// [`RegisterOutcome::AlreadyExists`] for a matching fingerprint. On either
    /// outcome the active config is resolved in place, so
    /// `table().resolved()` reports the server's uid and fingerprint
    /// afterwards.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::NoActiveTable`] when no table is bound, the
    /// stable fingerprint-mismatch error when a table of the same name exists
    /// with different columns, and the stable reserved-column or validation
    /// error the server reports for a rejected declaration.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future may leave the table created server-side with the
    /// local config still unresolved; re-registering is idempotent and
    /// resolves it.
    ///
    /// # Panics
    ///
    /// Panics if the active-table lock is poisoned.
    pub async fn register(&self) -> Result<RegisterOutcome, BifrostClientError> {
        let request = {
            let active = self.active.lock().expect("active table lock poisoned");
            active
                .as_ref()
                .ok_or(BifrostClientError::NoActiveTable)?
                .register_request()
        };
        let response: RegisterTableResponse = self
            .query
            .client()
            .http
            .request_json(reqwest::Method::POST, "/v1/bifrost/tables", Some(&request))
            .await?;
        // The lock is released for the network round trip, so `use_table` may
        // have replaced the binding meanwhile. An identity is only ever the
        // answer to the declaration that asked for it: apply it when the still-
        // active table would send this exact request, and otherwise leave the
        // new binding with its own resolution state rather than stamping one
        // table's uid and fingerprint onto another.
        let mut active = self.active.lock().expect("active table lock poisoned");
        if let Some(table) = active.as_mut()
            && table.register_request() == request
        {
            table.resolve(&response);
        }
        Ok(response.outcome)
    }

    /// Bind `table` as the write target, returning the previous binding.
    ///
    /// Cheap and synchronous: the producer for the new table is built lazily on
    /// its first row. The previous table's producer stays in the pool, so its
    /// buffered rows still flush.
    ///
    /// # Arguments
    /// * `table` - The table to bind as the new write target.
    ///
    /// # Panics
    ///
    /// Panics if the active-table lock is poisoned, which indicates an
    /// invariant-breaking panic in another client operation.
    pub fn use_table(&self, table: TableConfig) -> Option<TableConfig> {
        self.active
            .lock()
            .expect("active table lock poisoned")
            .replace(table)
    }

    /// Bind an already-registered table by name, describing it first.
    ///
    /// # Arguments
    /// * `fqn` - The registered table's `<namespace>.<name>`.
    ///
    /// # Errors
    ///
    /// As [`TableConfig::describe_with`].
    pub async fn use_table_by_name(&self, fqn: &str) -> Result<(), BifrostClientError> {
        let table = TableConfig::describe_with(self.query.client(), fqn).await?;
        self.use_table(table);
        Ok(())
    }

    /// Describe `fqn` once and return its cached destination for writes.
    ///
    /// This is the routing door for a caller that writes more than one table
    /// from one client — an instrumented application emitting observations
    /// alongside its own records. It deliberately does **not** touch the active
    /// binding: [`Self::use_table_by_name`] performs a remote describe *and*
    /// replaces the handle's shared active table, so two concurrent scoped
    /// callers would race each other's destination. Returning a
    /// [`WriterTable`] instead gives each caller its own immutable route.
    ///
    /// The first call for a table describes it; later calls return the cached
    /// schema without network IO. A miss takes this writer's describe gate and
    /// rechecks the cache under it, so concurrent first calls for one table
    /// perform one describe and every caller receives that one schema and its
    /// one pooled producer for this writer's lifetime.
    ///
    /// Describing proves only that the table exists and is accessible. Row
    /// values are checked against the schema when the queue seals a batch, and
    /// the server checks the registered fingerprint, so this returning is not a
    /// durability or whole-row-validation acknowledgement.
    ///
    /// # Arguments
    /// * `fqn` - The registered table's `<namespace>.<name>`; also the cache key.
    ///
    /// # Errors
    ///
    /// Returns the stable not-found, authentication, authorization,
    /// availability, or protocol error the server reported for an unknown,
    /// unauthorized, or unavailable table, or a schema-parse error when the
    /// description cannot be mapped back to Arrow.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future leaves no server state behind and caches nothing;
    /// describe is a read, and dropping the future releases the gate to the
    /// next waiter, which describes in its place.
    ///
    /// # Panics
    ///
    /// Panics if the described-table lock is poisoned.
    #[cfg(feature = "internal")]
    pub async fn writer_table(&self, fqn: &str) -> Result<WriterTable, BifrostClientError> {
        self.resolve_writer_table(fqn).await
    }

    /// Crate-internal body of [`Self::writer_table`], which the Run observe
    /// path calls whether or not the `internal` feature exposes the public
    /// method.
    ///
    /// Serves a cached destination without network IO; a miss takes the
    /// describe gate, rechecks the cache, then describes and caches once.
    ///
    /// # Arguments
    /// * `fqn` - The registered table's `<namespace>.<name>`; also the cache key.
    ///
    /// # Errors
    ///
    /// Returns the errors of [`TableConfig::describe_with`].
    ///
    /// # Cancellation
    ///
    /// Abandoning the future caches nothing and releases the gate.
    ///
    /// # Panics
    ///
    /// Panics if the described-table lock is poisoned.
    pub(crate) async fn resolve_writer_table(
        &self,
        fqn: &str,
    ) -> Result<WriterTable, BifrostClientError> {
        if let Some(table) = self.described_table(fqn) {
            return Ok(table);
        }
        let _gate = self.describe_gate.lock().await;
        if let Some(table) = self.described_table(fqn) {
            return Ok(table);
        }
        let config = TableConfig::describe_with(self.query.client(), fqn).await?;
        let described = WriterTable::new(fqn, config.user_schema().clone());
        self.described
            .lock()
            .expect("described table lock poisoned")
            .insert(Arc::from(fqn), described.clone());
        Ok(described)
    }

    /// The cached destination for `fqn`, if this writer has described it.
    ///
    /// # Arguments
    /// * `fqn` - The table's `<namespace>.<name>`, as passed to `writer_table`.
    ///
    /// # Panics
    ///
    /// Panics if the described-table lock is poisoned.
    #[must_use]
    #[cfg(feature = "internal")]
    pub fn cached_writer_table(&self, fqn: &str) -> Option<WriterTable> {
        self.described_table(fqn)
    }

    /// Look up `fqn` in this writer's described-table cache.
    ///
    /// # Arguments
    /// * `fqn` - The table's `<namespace>.<name>`.
    ///
    /// # Panics
    ///
    /// Panics if the described-table lock is poisoned.
    fn described_table(&self, fqn: &str) -> Option<WriterTable> {
        self.described
            .lock()
            .expect("described table lock poisoned")
            .get(fqn)
            .cloned()
    }

    /// Enqueue one JSON row into an explicitly named described table.
    ///
    /// The routing counterpart to [`Self::insert`]: the destination travels
    /// with the row rather than coming from the handle's shared active binding,
    /// so concurrent callers writing different tables never disturb each other.
    /// Correlation columns are added by the queue; they are not fields in
    /// `row`.
    ///
    /// Synchronous for the same reason [`Self::insert`] is — enqueueing is a
    /// bounded, non-blocking channel send — and durable only after
    /// [`Self::flush`] or [`Self::shutdown`] resolves.
    ///
    /// # Arguments
    /// * `table` - The described destination returned by `writer_table`.
    /// * `row` - One JSON-encoded row whose fields match `table`'s schema.
    /// * `correlation` - The card and run identities stamped onto the row.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Queue`] with `WYRD_CLIENT_429_QUEUE_FULL`
    /// when the producer channel, its staging ring, and the byte budget are all
    /// occupied.
    #[cfg(feature = "internal")]
    pub fn insert_into(
        &self,
        table: &WriterTable,
        row: Vec<u8>,
        correlation: Correlation,
    ) -> Result<(), BifrostClientError> {
        self.writer
            .insert(
                table.fqn(),
                table.user_schema(),
                row,
                correlation.card_uid,
                correlation.run_id,
            )
            .map_err(Into::into)
    }

    /// Enqueue every JSON row of one logical record into a described table, or
    /// none of them.
    ///
    /// The multi-row counterpart to [`Self::insert_into`]: every row carries
    /// the same correlation, and admission is all-or-none, so after a
    /// `WYRD_CLIENT_429_QUEUE_FULL` refusal the caller may drain with
    /// [`Self::flush`] or back off and resubmit the whole record without
    /// duplicating any row. Durable only after [`Self::flush`] or
    /// [`Self::shutdown`] resolves.
    ///
    /// # Arguments
    /// * `table` - The described destination returned by `writer_table`.
    /// * `rows` - Every JSON-encoded row of the record; admitted all-or-none.
    /// * `correlation` - The card and run identities stamped onto every row.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Queue`] with `WYRD_CLIENT_429_QUEUE_FULL`
    /// when the producer cannot admit every row now, and with
    /// `WYRD_CLIENT_413_PAYLOAD_TOO_LARGE` when the record has more rows than
    /// one producer can ever admit at once.
    #[cfg(feature = "internal")]
    pub fn insert_rows_into(
        &self,
        table: &WriterTable,
        rows: Vec<Vec<u8>>,
        correlation: Correlation,
    ) -> Result<(), BifrostClientError> {
        self.insert_observation(table, rows, correlation)
    }

    /// Enqueue every JSON row of one observation into a described table, or
    /// none of them.
    ///
    /// The crate-owned path behind [`Self::insert_rows_into`] that `observe`
    /// emits through without the `internal` feature. Each row carries its own
    /// time column, such as `created_at`, which the observation projection
    /// mints at emit, so batching, linger, retry, and flush never move it.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Queue`] with `WYRD_CLIENT_429_QUEUE_FULL`
    /// when the producer cannot admit every row now, and with
    /// `WYRD_CLIENT_413_PAYLOAD_TOO_LARGE` when the record has more rows than
    /// one producer can ever admit at once.
    pub(crate) fn insert_observation(
        &self,
        table: &WriterTable,
        rows: Vec<Vec<u8>>,
        correlation: Correlation,
    ) -> Result<(), BifrostClientError> {
        self.writer
            .insert_rows(
                table.fqn(),
                table.user_schema(),
                rows,
                correlation.card_uid,
                correlation.run_id,
            )
            .map_err(Into::into)
    }

    /// The active write binding, if any.
    ///
    /// # Panics
    ///
    /// Panics if the active-table lock is poisoned.
    #[must_use]
    pub fn table(&self) -> Option<TableConfig> {
        self.active
            .lock()
            .expect("active table lock poisoned")
            .clone()
    }

    // ── write ──────────────────────────────────────────────────────────────

    /// Enqueue one JSON row into the active table, propagating queue-full.
    ///
    /// Synchronous in every language surface because enqueueing is a bounded,
    /// non-blocking channel send. The row is durable only after
    /// [`Self::flush`] or [`Self::shutdown`] resolves.
    ///
    /// # Arguments
    /// * `row` - One JSON-encoded row whose fields match the active table's schema.
    /// * `correlation` - The card and run identities stamped onto the row.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::NoActiveTable`] when no table is bound, or
    /// [`BifrostClientError::Queue`] with `WYRD_CLIENT_429_QUEUE_FULL` when the
    /// producer channel, its staging ring, and the byte budget are all occupied.
    ///
    /// # Panics
    ///
    /// Panics if the active-table lock is poisoned.
    pub fn insert(&self, row: Vec<u8>, correlation: Correlation) -> Result<(), BifrostClientError> {
        let (table, schema) = {
            let active = self.active.lock().expect("active table lock poisoned");
            let table = active.as_ref().ok_or(BifrostClientError::NoActiveTable)?;
            (table.fqn(), table.user_schema().clone())
        };
        self.writer
            .insert(
                &table,
                &schema,
                row,
                correlation.card_uid,
                correlation.run_id,
            )
            .map_err(Into::into)
    }

    /// Write one already-built Arrow batch to `table` and await its durability.
    ///
    /// This is the write door for data whose columns the JSON row path cannot
    /// express — binary payloads, fixed-size identities, and nested list or
    /// struct columns — which is what the canonical signal tables are made of.
    /// Build the batch from the table's own published contract
    /// ([`Self::describe_table`] plus `wyrd_types::schema::writable_schema`) rather
    /// than from a restated schema.
    ///
    /// The table is named explicitly instead of taken from the bound active
    /// table: a batch of this kind targets one specific existing table, and
    /// binding one would invite a registration this caller does not want.
    ///
    /// The batch is sent verbatim. Whether its columns satisfy the destination
    /// table's contract is the server's judgement: it matches columns to the
    /// table by name, in any order, and answers any mismatch with its own
    /// stable whole-batch refusal. This client does not pre-check it, because
    /// a second implementation of that contract is exactly what would drift.
    ///
    /// Unlike [`Self::insert`], durability is complete when this resolves — the
    /// batch is not buffered and needs no [`Self::flush`].
    ///
    /// # Arguments
    /// * `table` - The destination table's `<namespace>.<name>`.
    /// * `batch` - The Arrow batch, sent verbatim in the table's writable schema.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Queue`] when the batch cannot be encoded,
    /// exceeds the accepted frame ceiling, or cannot fit this client's byte
    /// envelope, and the server's stable refusal when the batch is rejected.
    pub async fn write_batch(
        &self,
        table: &str,
        batch: &RecordBatch,
    ) -> Result<(), BifrostClientError> {
        self.writer
            .write_batch(table, batch)
            .await
            .map_err(Into::into)
    }

    /// Enqueue one owned Arrow batch for `table` without waiting for publication.
    ///
    /// This is the bounded, fire-and-return counterpart of
    /// [`Self::write_batch`] for Rust-native callers that must never delay
    /// their own work on Bifrost, such as the embedded server capture path.
    /// Admission charges this client's byte budget and one bounded producer
    /// slot, then returns. The background producer encodes the batch, seals it
    /// alone under a stable batch identity, and publishes it under `request_id`
    /// when supplied (so a server call's observations share its request), and retries, flushes,
    /// and drains it like buffered rows; it is durable only after that owner,
    /// [`Self::flush`], or [`Self::shutdown`] settles it.
    ///
    /// # Arguments
    /// * `table` - The destination table's `<namespace>.<name>`.
    /// * `batch` - The owned Arrow batch, sealed alone in the table's writable schema.
    /// * `request_id` - The request the batch publishes under, or `None` for a
    ///   fresh one.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Queue`] with backpressure or
    /// `WYRD_CLIENT_429_QUEUE_FULL` when the producer, channel, or byte
    /// envelope cannot admit the batch, or the client is shutting down.
    #[cfg(feature = "internal")]
    pub fn enqueue_batch(
        &self,
        table: &str,
        batch: RecordBatch,
        request_id: Option<RequestId>,
    ) -> Result<(), BifrostClientError> {
        self.writer
            .enqueue_batch(table, batch, request_id)
            .map_err(Into::into)
    }

    /// Flush every pooled producer and await each durable acknowledgement.
    ///
    /// Every table this client has written is drained, not only the active one,
    /// so a swap followed by a flush loses nothing.
    ///
    /// # Errors
    ///
    /// Returns the first producer or sink failure after every producer has been
    /// attempted, including the server's own stable refusal of a sealed batch.
    ///
    /// # Panics
    ///
    /// Panics when the blocking drain task cannot be joined.
    pub async fn flush(&self) -> Result<(), BifrostClientError> {
        self.drain(Drain::Flush).await
    }

    /// Drain every producer and stop its background task.
    ///
    /// # Errors
    ///
    /// As [`Self::flush`].
    ///
    /// # Panics
    ///
    /// Panics when the blocking drain task cannot be joined.
    pub async fn shutdown(&self) -> Result<(), BifrostClientError> {
        self.drain(Drain::Shutdown).await
    }

    /// Run one blocking producer drain off the async runtime's worker threads.
    ///
    /// The pool's drains block on channel acknowledgement, so they must not run
    /// on a runtime thread that the sink's own IO needs to make progress.
    ///
    /// # Errors
    ///
    /// Returns the first producer or sink failure.
    ///
    /// # Panics
    ///
    /// Panics when the blocking drain task cannot be joined.
    async fn drain(&self, kind: Drain) -> Result<(), BifrostClientError> {
        let writer = Arc::clone(&self.writer);
        tokio::task::spawn_blocking(move || match kind {
            Drain::Flush => writer.flush(),
            Drain::Shutdown => writer.shutdown(),
        })
        .await
        .expect("bifrost drain task joins")
        .map_err(Into::into)
    }

    // ── read ───────────────────────────────────────────────────────────────

    /// Run one SQL SELECT and collect every batch.
    ///
    /// Any authorized table, not just the active one. Bounded by the server's
    /// query floor; use [`Self::stream`] for a result set larger than memory.
    /// `params[i]` binds placeholder `$(i + 1)` in `query`; values travel as
    /// typed data beside the SQL text and are never interpolated into it.
    ///
    /// # Arguments
    /// * `query` - The SQL SELECT text.
    /// * `params` - Positional bind values; `params[i]` binds `$(i + 1)`.
    ///
    /// # Errors
    ///
    /// Returns the stable invalid-SQL error, the floor's non-SELECT or
    /// oversized refusal, an incomplete-stream failure when the response ends
    /// without its required terminal, and stable authentication, authorization,
    /// availability, or protocol errors.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future abandons the request; the server cancels the query
    /// when the response body is dropped.
    pub async fn sql(
        &self,
        query: &str,
        params: &[QueryParam],
    ) -> Result<QueryResult, BifrostClientError> {
        let mut stream = self.stream(query, params, None).await?;
        let mut batches = Vec::new();
        while let Some(batch) = stream.next_batch().await? {
            batches.push(batch);
        }
        let schema = stream.schema().cloned().ok_or_else(|| {
            BifrostClientError::Protocol("query stream omitted its schema".to_owned())
        })?;
        let terminal = stream
            .terminal()
            .cloned()
            .ok_or(BifrostClientError::IncompleteQueryStream)?;
        Ok(QueryResult {
            batches,
            schema,
            terminal,
        })
    }

    /// Run one SQL SELECT and deserialize every row into `T`.
    ///
    /// The typed counterpart of [`Self::sql`]: the same query, authorization,
    /// limits, and validated terminal, projected onto the caller's own type
    /// after the result is complete. `T` describes the columns the query
    /// selects; it is never sent to the server and says nothing about how a
    /// table is stored.
    ///
    /// # Arguments
    /// * `query` - The SQL SELECT text.
    /// * `params` - Positional bind values; `params[i]` binds `$(i + 1)`.
    ///
    /// # Errors
    ///
    /// As [`Self::sql`], plus [`BifrostClientError::RowDeserialization`] when any row
    /// does not fit `T`. That failure is total: no partially converted result
    /// is returned.
    ///
    /// # Cancellation
    ///
    /// As [`Self::sql`]; conversion happens only after the query completes.
    pub async fn sql_as<T: DeserializeOwned>(
        &self,
        query: &str,
        params: &[QueryParam],
    ) -> Result<Vec<T>, BifrostClientError> {
        self.sql(query, params).await?.deserialize()
    }

    /// Read the span tree of `trace_id` as canonical span records: every span
    /// of that trace plus every span of each trace holding a span that links
    /// directly to it.
    ///
    /// The closure is one inbound-link hop: a trace linking only to a linked
    /// trace is not read. Spans are ordered by start time, then trace id, then
    /// span id, and a span id stored more than once keeps only its first copy
    /// in that order. The read is one
    /// query over the caller's authorized `vala.traces.spans` with the trace
    /// id bound as a parameter, so it inherits the query deadline and result
    /// limits, and another tenant's trace or one never seen reads empty.
    ///
    /// # Errors
    ///
    /// As [`Self::sql`], with the span table not yet created reading empty,
    /// plus [`BifrostClientError::RowDeserialization`] when a span row does
    /// not decode. Either failure is total: no partial timeline is returned.
    pub async fn trace_timeline(
        &self,
        trace_id: TraceId,
    ) -> Result<Vec<SpanRecord>, BifrostClientError> {
        let query = format!(
            "SELECT {SPAN_COLUMNS} FROM vala.traces.spans \
             WHERE trace_id = decode($1, 'hex') OR trace_id IN ( \
               SELECT trace_id FROM ( \
                 SELECT trace_id, unnest(links) AS link FROM vala.traces.spans) \
               WHERE link['trace_id'] = decode($1, 'hex'))"
        );
        let result = match self
            .sql(&query, &[QueryParam::String(trace_id.to_hex())])
            .await
        {
            Ok(result) => result,
            Err(error) if WyrdError::from(&error).code() == SPANS_NOT_CREATED => {
                return Ok(Vec::new());
            }
            Err(error) => return Err(error),
        };
        let mut spans = result
            .batches()
            .iter()
            .flat_map(|batch| (0..batch.num_rows()).map(move |row| span_record(batch, row)))
            .collect::<Result<Vec<_>, _>>()
            .map_err(BifrostClientError::RowDeserialization)?;
        spans.sort_by(|left, right| {
            (
                left.start_time,
                left.trace_id.as_bytes(),
                left.span_id.as_bytes(),
            )
                .cmp(&(
                    right.start_time,
                    right.trace_id.as_bytes(),
                    right.span_id.as_bytes(),
                ))
        });
        let mut seen = HashSet::new();
        spans.retain(|span| seen.insert((span.trace_id, span.span_id)));
        Ok(spans)
    }

    /// Run one SQL SELECT and return its batches as they arrive.
    ///
    /// The stream owns the HTTP response body: dropping it propagates
    /// cancellation. The terminal frame is required, so a stream that ends
    /// without one fails rather than presenting partial rows as success.
    ///
    /// # Arguments
    /// * `query` - The SQL SELECT text.
    /// * `params` - Positional bind values, exactly as in [`Self::sql`].
    /// * `deadline` - The server-side query deadline, or `None` for the
    ///   server default. Whole milliseconds are sent; the server accepts 1 ms
    ///   to `u32::MAX` ms.
    ///
    /// # Errors
    ///
    /// As [`Self::sql`], plus `WYRD_VALA_400_QUERY_INVALID_SQL` for a
    /// deadline outside the accepted range and a decode error on a malformed
    /// Arrow IPC frame.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future abandons the request before any row is read.
    pub async fn stream(
        &self,
        query: &str,
        params: &[QueryParam],
        deadline: Option<Duration>,
    ) -> Result<QueryResultStream, BifrostClientError> {
        self.query
            .query(&BifrostQueryRequest {
                params: params.to_vec(),
                sql: query.to_owned(),
                deadline_ms: deadline
                    .map(|deadline| i64::try_from(deadline.as_millis()).unwrap_or(i64::MAX)),
            })
            .await
    }

    /// Start one query from a complete request and return its batches as they arrive.
    ///
    /// The raw form [`Self::stream`] and [`Self::sql`] wrap: the caller chooses
    /// the deadline. The request is validated before any
    /// IO, and the returned stream owns the HTTP response body.
    ///
    /// # Arguments
    /// * `request` - The complete query request: SQL, bind values, and deadline.
    ///
    /// # Errors
    ///
    /// Returns the stable invalid-SQL error for an invalid request, and the
    /// authentication, authorization, availability, or protocol error the
    /// transport reported.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future abandons the request; dropping the returned stream
    /// cancels response-body consumption.
    #[cfg(feature = "internal")]
    pub async fn query(
        &self,
        request: &BifrostQueryRequest,
    ) -> Result<QueryResultStream, BifrostClientError> {
        self.query.query(request).await
    }

    /// Run one query and collect it within 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
    ///
    /// As [`Self::query`], plus failed-terminal, incomplete-stream, Arrow, and
    /// bounds errors. Exceeding a bound drops the live stream and never returns
    /// truncated success.
    ///
    /// # Cancellation
    ///
    /// Abandoning 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.collect_bounded(request, limits).await
    }

    /// List the active queries visible to the authenticated tenant.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, audit, availability, or
    /// protocol errors.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future abandons the request without client-owned state.
    pub async fn running(&self) -> Result<Vec<RunningQuerySummary>, BifrostClientError> {
        self.query.running().await
    }

    /// Get one active query visible to the authenticated tenant.
    ///
    /// # Arguments
    /// * `request_id` - The id of the active query to look up.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, not-found, availability,
    /// or protocol errors.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future leaves the active query unchanged.
    pub async fn status(
        &self,
        request_id: &RequestId,
    ) -> Result<RunningQuerySummary, BifrostClientError> {
        self.query.status(request_id).await
    }

    /// Request server-side cancellation of one active query.
    ///
    /// Does not close any local response stream.
    ///
    /// # Arguments
    /// * `request_id` - The id of the active query to cancel.
    ///
    /// # Errors
    ///
    /// Returns stable authentication, authorization, not-found, availability,
    /// or protocol errors.
    ///
    /// # Cancellation
    ///
    /// Once the server accepts cancellation, abandoning the future does not
    /// reverse the server-side transition.
    pub async fn cancel(
        &self,
        request_id: &RequestId,
    ) -> Result<CancelRunningQueryResponse, BifrostClientError> {
        self.query.cancel(request_id).await
    }

    /// List the compact entries of every table the caller may read.
    ///
    /// Returns the server's authorized [`BifrostTableEntry`] values unchanged:
    /// visible built-in tables and the tenant's own, without schemas. Call
    /// [`Bifrost::describe_table`] for one table's schema when it is needed.
    ///
    /// # Errors
    ///
    /// Returns the stable authentication, authorization (a principal without
    /// `bifrost_table:read`), availability, or protocol error the server
    /// reported.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future leaves no server state behind; listing is a read.
    pub async fn list_tables(&self) -> Result<Vec<BifrostTableEntry>, BifrostClientError> {
        self.query.list_tables().await
    }

    /// Read one registered table's server-owned description.
    ///
    /// The canonical introspection door on this client: schema, identity, and
    /// physical layout exactly as the server holds them.
    ///
    /// # Arguments
    /// * `namespace` - The table's namespace, such as `vala.traces`.
    /// * `name` - The table's name within `namespace`, such as `spans`.
    ///
    /// # Errors
    ///
    /// Returns the stable not-found, authentication, authorization,
    /// availability, or protocol error the server reported.
    ///
    /// # Cancellation
    ///
    /// Abandoning the future leaves no server state behind; describe is a read.
    pub async fn describe_table(
        &self,
        namespace: &str,
        name: &str,
    ) -> Result<BifrostTableDescription, BifrostClientError> {
        self.query.describe_table(namespace, name).await
    }

    /// The client scope every pooled producer is keyed under.
    #[must_use]
    #[cfg(feature = "internal")]
    pub fn scope(&self) -> &ClientScope {
        self.writer.scope()
    }

    /// Number of distinct table producers currently pooled.
    ///
    /// One per table this client has written since construction, including
    /// tables it has since swapped away from — which is what makes a swap
    /// lossless.
    #[must_use]
    pub fn producer_count(&self) -> usize {
        self.writer.producer_count()
    }

    /// Rows dropped by the fire-and-forget [`crate::bifrost::observe::record`] path.
    ///
    /// Always zero for [`Self::insert`], which refuses rather than drops.
    #[must_use]
    pub fn dropped(&self) -> u64 {
        self.writer.dropped()
    }

    /// Point-in-time bounded-ownership accounting for this client's producers.
    #[must_use]
    #[cfg(feature = "internal")]
    pub fn metrics(&self) -> BifrostMetrics {
        self.writer.metrics()
    }

    /// Registers `observer` to receive the row count of every accepted row
    /// this client settles as lost after admission, such as a terminal
    /// publication refusal or a retry slot it could not retain, as the loss
    /// settles. The loss's bytes and retry slot are already released, so
    /// [`Self::metrics`] read from the observer reports settled ownership.
    ///
    /// The observer runs on the producer task and must not block. Only the
    /// first registration takes effect.
    ///
    /// # Arguments
    /// * `observer` - Callback receiving the row count of each settled loss; it
    ///   runs on the producer task and must not block.
    #[cfg(feature = "internal")]
    pub fn observe_losses(&self, observer: impl Fn(u64) + Send + Sync + 'static) {
        self.writer.observe_losses(observer);
    }

    /// The producer pool this client writes through.
    ///
    /// Crate-private so [`crate::bifrost::observe::record`] can reach the drop-counting
    /// path without the pool becoming a second public write door.
    pub(crate) fn writer(&self) -> &WriterPool {
        &self.writer
    }
}

/// Which terminal drain [`Bifrost::drain`] should run.
///
/// The two differ only in whether the producer is stopped afterwards, so they
/// share one blocking-offload path rather than duplicating it.
#[derive(Debug, Clone, Copy)]
enum Drain {
    /// Seal and acknowledge, leaving every producer running.
    Flush,
    /// Seal, acknowledge, and stop every producer's background task.
    Shutdown,
}

/// A collected query result: the Arrow batches plus the validated terminal.
///
/// Held as decoded [`RecordBatch`]es rather than raw IPC so Rust callers do not
/// re-decode; the Python and TypeScript surfaces re-encode once at their
/// boundary, which is the only place a language-native table can be built.
#[derive(Debug)]
pub struct QueryResult {
    /// Every batch the query produced, in arrival order.
    batches: Vec<RecordBatch>,
    /// The schema decoded from the stream's required initial schema frame.
    schema: SchemaRef,
    /// The validated success terminal; a failed terminal never reaches here.
    terminal: QueryTerminalFrame,
}

impl QueryResult {
    /// Every batch the query produced, in arrival order.
    #[must_use]
    pub fn batches(&self) -> &[RecordBatch] {
        &self.batches
    }

    /// The authoritative result schema.
    #[must_use]
    pub fn schema(&self) -> &SchemaRef {
        &self.schema
    }

    /// The validated terminal frame the server closed the stream with.
    #[must_use]
    pub fn terminal(&self) -> &QueryTerminalFrame {
        &self.terminal
    }

    /// Total decoded rows across every batch.
    #[must_use]
    pub fn num_rows(&self) -> usize {
        self.batches.iter().map(RecordBatch::num_rows).sum()
    }

    /// Deserialize every row into `T` through the Arrow JSON projection.
    ///
    /// Arrow's own writer produces one JSON array over the retained batches, so
    /// the column names and types `T` sees are exactly the schema the server
    /// sent — there is no second, hand-written type mapping to disagree with
    /// it. A Variant column renders as its JSON value through the shared
    /// [`WyrdJsonEncoderFactory`],
    /// so `T` reads it as a `serde_json::Value` or any type that value fits.
    /// An empty result writes no array at all, which is the zero-row case.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Arrow`] when the JSON projection fails and
    /// [`BifrostClientError::RowDeserialization`] when any row does not fit `T`.
    fn deserialize<T: DeserializeOwned>(&self) -> Result<Vec<T>, BifrostClientError> {
        let mut bytes = Vec::new();
        {
            let mut writer = WriterBuilder::new()
                .with_encoder_factory(Arc::new(WyrdJsonEncoderFactory))
                .build::<_, JsonArray>(&mut bytes);
            for batch in &self.batches {
                writer
                    .write(batch)
                    .map_err(|error| BifrostClientError::Arrow(error.to_string()))?;
            }
            writer
                .finish()
                .map_err(|error| BifrostClientError::Arrow(error.to_string()))?;
        }
        if bytes.is_empty() {
            return Ok(Vec::new());
        }
        serde_json::from_slice(&bytes)
            .map_err(|error| BifrostClientError::RowDeserialization(error.to_string()))
    }

    /// Encode the whole result as one Arrow IPC stream.
    ///
    /// This is how the Python and TypeScript boundaries receive it: one stream
    /// their own Arrow implementation reads, rather than a per-language rebuild
    /// of the schema and arrays.
    ///
    /// # Errors
    ///
    /// Returns [`BifrostClientError::Arrow`] when IPC encoding fails.
    pub fn to_bytes(&self) -> Result<Vec<u8>, BifrostClientError> {
        let mut buffer = Vec::new();
        {
            let mut writer = arrow::ipc::writer::StreamWriter::try_new(&mut buffer, &self.schema)
                .map_err(|error| BifrostClientError::Arrow(error.to_string()))?;
            for batch in &self.batches {
                writer
                    .write(batch)
                    .map_err(|error| BifrostClientError::Arrow(error.to_string()))?;
            }
            writer
                .finish()
                .map_err(|error| BifrostClientError::Arrow(error.to_string()))?;
        }
        Ok(buffer)
    }
}

/// The SDK-facing spelling of one register outcome.
///
/// The wire enum serializes as `Created`/`AlreadyExists`; every language
/// client presents the snake_case names its users read. Naming them once here
/// is what keeps the Python and TypeScript surfaces from disagreeing about
/// what a successful registration answered.
///
/// # Arguments
/// * `outcome` - The wire outcome of one registration.
#[must_use]
#[cfg(feature = "internal")]
pub fn register_outcome_name(outcome: RegisterOutcome) -> &'static str {
    match outcome {
        RegisterOutcome::Created => "created",
        RegisterOutcome::AlreadyExists => "already_exists",
    }
}

/// Assemble a [`WyrdClient`] from optionally-overridden transport values.
///
/// Every argument is optional and every omitted one falls through to the
/// existing chain exactly once, the same resolution as
/// [`WyrdClient::from_global`]: the global config file, then
/// [`ClientConfig::from_global_with_overrides`] for the two endpoints (an
/// omitted gRPC endpoint derives from the effective `server_url`), then
/// `ClientConfig::resolve_credential` for the credential, whose floor is
/// `~/.config/wyrd/credentials.toml` `[default].api_key`. An explicit
/// value is written into the config's tier-0 slot, so it wins over the
/// environment rather than racing it. The tenant selector stays
/// `WYRD_TENANT`/configuration: a credential already names its tenant.
///
/// This is the one door the Python and TypeScript constructors use, so their
/// "omitted resolves, explicit overrides" behavior is the Rust one and cannot
/// drift per language.
///
/// # Arguments
/// * `server_url` - The HTTP endpoint, or `None` to resolve it from the
///   environment and configuration.
/// * `credential` - The API key or token, or `None` to resolve it from the
///   credential chain.
/// * `grpc_url` - The gRPC endpoint, or `None` to derive it from the
///   effective `server_url`.
///
/// # Errors
///
/// Returns `WYRD_CLIENT_400_CONFIG_INVALID` naming `server_url`, `grpc_url`,
/// or `credential` when that argument is given but empty or whitespace, a
/// configuration error when the global config file exists but cannot
/// be read or parsed, the stable no-credentials error when nothing in the chain
/// yields a credential, or a transport error when the HTTP client cannot be
/// built.
#[cfg(feature = "internal")]
pub fn client_from_options(
    server_url: Option<&str>,
    credential: Option<&str>,
    grpc_url: Option<&str>,
) -> Result<WyrdClient, BifrostClientError> {
    client_from_options_in(Environment::Process, server_url, credential, grpc_url)
}

/// Assemble a [`WyrdClient`] as `client_from_options` does, reading every
/// ambient value, including the config directory, from `environment`.
///
/// # Arguments
/// * `environment` - Source of every ambient variable: the config directory,
///   endpoint fallbacks, and credential chain.
/// * `server_url` - The HTTP endpoint, or `None` to resolve it.
/// * `credential` - The API key or token, or `None` to resolve it.
/// * `grpc_url` - The gRPC endpoint, or `None` to derive it.
///
/// # Errors
///
/// Returns `WYRD_CLIENT_400_CONFIG_INVALID` naming `server_url`, `grpc_url`,
/// or `credential` when that argument is given but blank, a configuration
/// error when the global config file cannot be read or parsed, the stable
/// no-credentials error when nothing yields a credential, or a transport
/// error when the HTTP client cannot be built.
#[cfg(feature = "internal")]
fn client_from_options_in(
    environment: Environment,
    server_url: Option<&str>,
    credential: Option<&str>,
    grpc_url: Option<&str>,
) -> Result<WyrdClient, BifrostClientError> {
    for (field, value) in [
        ("server_url", server_url),
        ("grpc_url", grpc_url),
        ("credential", credential),
    ] {
        if value.is_some_and(|value| value.trim().is_empty()) {
            return Err(BifrostClientError::Client(WyrdClientError::Config {
                field: field.to_owned(),
                reason: "must not be empty".to_owned(),
            }));
        }
    }
    let global = GlobalConfig::load_in(&environment).map_err(BifrostClientError::from)?;
    let mut config = ClientConfig::from_environment(environment, &global, server_url, grpc_url);
    if let Some(credential) = credential {
        config.credential = Some(secrecy::SecretString::from(credential.to_owned()));
    }
    WyrdClient::with_config(config).map_err(BifrostClientError::from)
}

/// Ingest sink behind [`Bifrost::query_only`]: refuses every batch.
///
/// A read-only client has no ingest channel, so a write that reaches the sink
/// settles terminally with a stable validation error rather than dialling a
/// transport or reporting success.
#[cfg(feature = "internal")]
struct QueryOnlySink;

#[cfg(feature = "internal")]
#[async_trait::async_trait]
impl BatchSink<ClientByteGuard> for QueryOnlySink {
    /// Refuse `batch` without sending it.
    ///
    /// # Errors
    ///
    /// Always returns a terminal `WYRD_SPEC_400_VALIDATION` naming the table.
    async fn send(
        &self,
        batch: &SealedBatch<ClientByteGuard>,
    ) -> Result<DurableBatchAck, SinkError> {
        Err(wyrd_queue::SinkError::Terminal(
            wyrd_spec::error::WyrdError::Validation {
                message: "this Bifrost client was built with Bifrost::query_only and cannot write"
                    .to_owned(),
                details: serde_json::json!({ "table": batch.table }),
            },
        ))
    }
}

#[cfg(all(test, feature = "internal"))]
mod tests {
    use super::client_from_options_in;
    use crate::environment::Environment;

    /// An omitted `server_url` resolves from the global config file in the
    /// Wyrd config directory, the same file [`crate::WyrdClient::from_global`]
    /// reads, so a language `WyrdClient()` and Rust `Cards::connect` reach the
    /// same server.
    #[test]
    fn omitted_server_url_comes_from_the_global_config_file() {
        let config_dir = tempfile::tempdir().expect("temporary config directory");
        std::fs::write(
            config_dir.path().join("config.toml"),
            "[client]\nhttp_url = \"http://127.0.0.1:9711\"\n",
        )
        .expect("write config.toml");
        let config_home = config_dir.path().to_str().expect("utf-8 temp path");
        let environment =
            Environment::from([("WYRD_CONFIG_HOME", config_home), ("HOME", config_home)]);

        let client = client_from_options_in(environment, None, Some("wyrd_sk_test"), None)
            .expect("client from the configured server");

        assert_eq!(client.server_url(), "http://127.0.0.1:9711");
    }

    /// An explicitly empty or blank `server_url`, `grpc_url`, or `credential`
    /// is refused with `WYRD_CLIENT_400_CONFIG_INVALID` naming the public
    /// parameter, never an internal config path.
    ///
    /// # Panics
    /// Panics when a blank value is accepted or the refusal names another field.
    #[test]
    fn blank_explicit_options_are_refused_by_their_public_name() {
        let cases = [
            ("server_url", Some(""), Some("wyrd_sk_test"), None),
            ("server_url", Some("  "), Some("wyrd_sk_test"), None),
            ("grpc_url", None, Some("wyrd_sk_test"), Some(" ")),
            ("credential", None, Some(""), None),
        ];
        for (field, server_url, credential, grpc_url) in cases {
            let error = client_from_options_in(
                Environment::from([("HOME", "/nonexistent")]),
                server_url,
                credential,
                grpc_url,
            )
            .expect_err("a blank explicit option is refused");
            let error = wyrd_spec::error::WyrdError::from(error);

            assert_eq!(error.code(), "WYRD_CLIENT_400_CONFIG_INVALID", "{field}");
            assert_eq!(
                error.problem().details,
                serde_json::json!({ "field": field, "reason": "must not be empty" }),
            );
        }
    }
}