laser-wire 0.10.2

use crate::destination::BackendResourceId;
use crate::error::InvalidError;
use crate::filter::{FILTER_EVALUATOR_VERSION, FilterCodec};
use crate::query::{AggFunc, Consistency, SqlDialect};
use crate::schema::LogicalTypeKind;
use crate::topology::WireTopology;
use crate::validate::Validate;
use serde::{Deserialize, Serialize};
use std::collections::BTreeSet;

/// Capability feature bits advertised in [`OpVersions::features`]. Each constant
/// names one managed sub-feature a server serves beyond the base surface, so a
/// binary client feature-detects it (before attempting the op) the way the HTTP
/// surface reads the boolean flags on `Capabilities`. Additive and pinned
/// cross-repo: a new bit is set by a newer server and ignored by an older
/// client (which simply does not light up that capability).
pub mod feature {
    /// The key-value store serves compare-and-swap (`AGDX_KV_CAS`).
    pub const KV_CAS: u64 = 1 << 0;
    /// The query surface honors `Consistency::ReadYourWrites`.
    pub const READ_YOUR_WRITES: u64 = 1 << 1;
    /// Retired with linearizable query reads. No server sets it and the bit is
    /// never reused.
    pub const STRONG_CONSISTENCY: u64 = 1 << 2;
    /// The key-value store serves fenced compare-and-swap (`AGDX_KV_CAS_FENCED`).
    pub const KV_CAS_FENCED: u64 = 1 << 3;
    /// Retired with the run registry. No server sets it and the bit is never
    /// reused.
    pub const AGENT_WORKFLOW: u64 = 1 << 4;
    /// The query surface serves lexical relevance search (`Query.text`).
    pub const KEYWORD_SEARCH: u64 = 1 << 5;
    /// The deployment publishes the change feed (`ChangeRecord`s on the
    /// changes topic) for bindings that opt into `notify`.
    pub const WATCH: u64 = 1 << 6;
    /// The streaming server serves the authorization control band (`AGDX_AUTHZ_*`).
    pub const AUTHZ: u64 = 1 << 7;
    /// Retired with lakehouse destinations. No server sets it and the bit is
    /// never reused.
    pub const DESTINATIONS: u64 = 1 << 8;
    /// The key-value store serves the revocable fenced-lease contract at
    /// `KV_LEASE_OP_VERSION`: holder-scoped acquire, renewal
    /// (`AGDX_KV_LEASE_RENEW`), holder-and-fence-validated release, fenced
    /// compare-and-swap requiring a live lease, and the barriered read
    /// (`KvGet::min_position`). Subsumes [`KV_CAS_FENCED`]: a client must not
    /// send the reshaped lease ops (or a barriered read) to a server without
    /// this bit, which would silently decode them under the old contract.
    pub const KV_FENCED_LEASES: u64 = 1 << 9;
    /// The streaming server serves consumer filters: filtered reads, fenced
    /// acknowledgments, previews, sample tests, and validation
    /// (`AGDX_FILTERED_POLL` and the rest of the delivery band's server-native
    /// decade). Server-native, so it holds with the managed backend disabled.
    /// The saved-filter catalog additionally needs the backend's `filter` op
    /// version.
    pub const CONSUMER_FILTERS: u64 = 1 << 10;
    /// The streaming server serves group-aware reads: `FilterRef::Group`
    /// resolves a consumer group's own policy, an unbound group is delivered
    /// unfiltered without payload decoding, pages and acknowledgments carry the
    /// execution mode and policy generation, and the request may bound the
    /// examined records and wait for a catalog position. A client must not send
    /// the automatic selector to a server without this bit.
    pub const GROUP_POLICY_READS: u64 = 1 << 11;
    /// The managed backend serves session reads for one named stream.
    pub const SESSIONS: u64 = 1 << 12;
    /// The managed backend runs in stream tenancy: every managed name is
    /// scoped to one stream, and the change feed and dead letters of each
    /// stream ride their own ops topics.
    pub const STREAM_TENANCY: u64 = 1 << 13;
    /// The managed backend serves whole-record projections, control replies
    /// that wait for the applied outcome, and binding status on the projection
    /// browse.
    pub const MATERIALIZATIONS: u64 = 1 << 14;
}

/// The wire op versions a server accepts, one per surface, plus the capability
/// feature bits it advertises. A pinned wire shape, mirrored by the HTTP
/// capabilities `versions` block.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[non_exhaustive]
pub struct OpVersions {
    pub query: u32,
    pub control: u32,
    pub kv: u32,
    pub fork: u32,
    /// The agent envelope (AGDX) version LaserData Cloud consumes for its
    /// conversation projections. `0` means "not advertised" and is skipped on
    /// encode, so pre-AGDX hello frames stay byte-identical and decode unchanged.
    #[serde(default, skip_serializing_if = "is_zero")]
    pub agent: u32,
    /// The knowledge-graph op version served. `0` means not served, skipped on
    /// encode so a pre-graph hello frame stays byte-identical. Mirrors the
    /// `managed_graph` HTTP capability flag. (Agentic memory rides this plus the
    /// query surface, so it has no op version of its own.)
    #[serde(default, skip_serializing_if = "is_zero")]
    pub graph: u32,
    /// The consumer-filter catalog op version served by the managed backend. Zero
    /// means the catalog (saved filters and group bindings) is not served.
    #[serde(default, skip_serializing_if = "is_zero")]
    pub filter: u32,
    /// Capability feature bits (see [`feature`]): managed sub-features served
    /// beyond the base surface (compare-and-swap, read-your-writes, and the
    /// rest). `0` (the default) is skipped on encode, so a pre-feature
    /// hello reply stays byte-identical and an old client just sees no extra
    /// capabilities.
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub features: u64,
}

fn is_zero(value: &u32) -> bool {
    *value == 0
}

fn is_zero_u64(value: &u64) -> bool {
    *value == 0
}

impl OpVersions {
    /// Versions per surface. The struct is `#[non_exhaustive]` (new surfaces
    /// land without a breaking change), so this is the constructor.
    pub fn new(query: u32, control: u32, kv: u32, fork: u32) -> Self {
        Self {
            query,
            control,
            kv,
            fork,
            agent: 0,
            graph: 0,
            filter: 0,
            features: 0,
        }
    }

    /// Returns a copy advertising this agent-envelope (AGDX) version.
    #[must_use]
    pub fn with_agent(mut self, agent: u32) -> Self {
        self.agent = agent;
        self
    }

    /// Returns a copy advertising the knowledge-graph op version served.
    #[must_use]
    pub fn with_graph(mut self, graph: u32) -> Self {
        self.graph = graph;
        self
    }

    /// Returns a copy advertising the consumer-filter catalog version served.
    #[must_use]
    pub fn with_filter(mut self, filter: u32) -> Self {
        self.filter = filter;
        self
    }

    /// Returns a copy advertising the capability feature bits in `features`
    /// (an OR of [`feature`] constants).
    #[must_use]
    pub fn with_features(mut self, features: u64) -> Self {
        self.features = features;
        self
    }

    /// Whether a [`feature`] bit (or set of bits) is advertised.
    pub const fn has_feature(&self, bit: u64) -> bool {
        self.features & bit == bit
    }
}

/// Body of the `AGDX_HELLO` probe reply: the wire op versions the server (and
/// its managed backend) accepts, mirroring the HTTP capabilities `versions`
/// block. A pinned wire shape. Pre-versioned
/// servers answer the probe with an empty body, which a client treats as "no
/// versions advertised", never an error.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[non_exhaustive]
pub struct HelloReply {
    pub versions: OpVersions,
}

impl HelloReply {
    /// Constructor for the non-exhaustive wire struct.
    pub fn new(versions: OpVersions) -> Self {
        Self { versions }
    }
}

pub const BACKEND_DESCRIPTOR_VERSION: u32 = 1;

/// One observed backend resource and its structured, secret-free capabilities.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[non_exhaustive]
pub struct BackendDescriptor {
    pub descriptor_version: u32,
    pub resource_id: BackendResourceId,
    pub mode: BackendMode,
    pub label: String,
    pub implementation: BackendImplementation,
    pub observed_backend_generation: u64,
    pub runtime_configuration_revision: u64,
    pub desired_state: BackendDesiredState,
    pub observed_state: BackendObservedState,
    pub readiness: BackendReadiness,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub query: Option<QueryCapabilities>,
    pub schema: SchemaCapabilities,
    pub limits: BackendLimits,
}

impl BackendDescriptor {
    pub fn new(
        resource_id: BackendResourceId,
        mode: BackendMode,
        label: impl Into<String>,
        implementation: BackendImplementation,
        observed_backend_generation: u64,
        runtime_configuration_revision: u64,
    ) -> Self {
        Self {
            descriptor_version: BACKEND_DESCRIPTOR_VERSION,
            resource_id,
            mode,
            label: label.into(),
            implementation,
            observed_backend_generation,
            runtime_configuration_revision,
            desired_state: BackendDesiredState::Disabled,
            observed_state: BackendObservedState::Disabled,
            readiness: BackendReadiness::not_ready(BackendReadinessCode::Disabled),
            query: None,
            schema: SchemaCapabilities::default(),
            limits: BackendLimits::default(),
        }
    }

    #[must_use]
    pub fn with_state(
        mut self,
        desired_state: BackendDesiredState,
        observed_state: BackendObservedState,
        readiness: BackendReadiness,
    ) -> Self {
        self.desired_state = desired_state;
        self.observed_state = observed_state;
        self.readiness = readiness;
        self
    }

    #[must_use]
    pub fn with_query(mut self, capabilities: QueryCapabilities) -> Self {
        self.query = Some(capabilities);
        self
    }

    #[must_use]
    pub fn with_schema(mut self, capabilities: SchemaCapabilities) -> Self {
        self.schema = capabilities;
        self
    }

    #[must_use]
    pub fn with_limits(mut self, limits: BackendLimits) -> Self {
        self.limits = limits;
        self
    }
}

impl Validate for BackendDescriptor {
    fn validate(&self) -> Result<(), InvalidError> {
        if self.descriptor_version != BACKEND_DESCRIPTOR_VERSION {
            return Err(InvalidError::new(format!(
                "backend descriptor version must be {BACKEND_DESCRIPTOR_VERSION}"
            )));
        }
        if self.resource_id.as_u128() == 0
            || self.observed_backend_generation == 0
            || self.runtime_configuration_revision == 0
        {
            return Err(InvalidError::new(
                "backend identity and observed revisions must be nonzero",
            ));
        }
        validate_descriptor_text("backend label", &self.label)?;
        validate_descriptor_text("backend implementation kind", &self.implementation.kind)?;
        validate_descriptor_text(
            "backend implementation version",
            &self.implementation.version,
        )?;
        self.readiness.validate()?;
        if self.readiness.ready && self.observed_state != BackendObservedState::Ready {
            return Err(InvalidError::new(
                "ready backend must report the ready observed state",
            ));
        }
        if let Some(query) = &self.query {
            query.validate()?;
        }
        if self.readiness.ready {
            if self.schema.logical_schema && self.limits.max_schema_fields == 0 {
                return Err(InvalidError::new(
                    "ready logical-schema backend must advertise a schema-field limit",
                ));
            }
            if self.query.is_some()
                && (self.limits.max_query_rows == 0
                    || self.limits.max_query_bytes == 0
                    || self.limits.max_scan_bytes == 0
                    || self.limits.max_query_micros == 0
                    || self.limits.max_concurrent_queries == 0)
            {
                return Err(InvalidError::new(
                    "ready query backend must advertise nonzero query limits",
                ));
            }
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum BackendMode {
    Operational,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct BackendImplementation {
    pub kind: String,
    pub version: String,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum BackendDesiredState {
    Disabled,
    Enabled,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum BackendObservedState {
    Disabled,
    Starting,
    Ready,
    Degraded,
    Unavailable,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct BackendReadiness {
    pub ready: bool,
    pub reasons: Vec<BackendReadinessReason>,
    pub observed_at_micros: u64,
}

impl BackendReadiness {
    pub fn not_ready(code: BackendReadinessCode) -> Self {
        Self {
            ready: false,
            reasons: vec![BackendReadinessReason { code, detail: None }],
            observed_at_micros: 0,
        }
    }

    pub fn ready(observed_at_micros: u64) -> Self {
        Self {
            ready: true,
            reasons: Vec::new(),
            observed_at_micros,
        }
    }
}

impl Validate for BackendReadiness {
    fn validate(&self) -> Result<(), InvalidError> {
        if self.ready {
            if !self.reasons.is_empty() || self.observed_at_micros == 0 {
                return Err(InvalidError::new(
                    "ready backend requires an observation time and no failure reasons",
                ));
            }
            return Ok(());
        }
        if self.reasons.is_empty() {
            return Err(InvalidError::new(
                "not-ready backend requires at least one stable reason",
            ));
        }
        if self.reasons.len() > 32 {
            return Err(InvalidError::new(
                "backend readiness reason count exceeds cap 32",
            ));
        }
        for reason in &self.reasons {
            if let Some(detail) = &reason.detail {
                validate_descriptor_text("backend readiness detail", detail)?;
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct BackendReadinessReason {
    pub code: BackendReadinessCode,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub detail: Option<String>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum BackendReadinessCode {
    Disabled,
    ConfigurationPending,
    ConfigurationRejected,
    CredentialUnavailable,
    ObjectStoreUnavailable,
    CatalogUnavailable,
    QueryRuntimeUnavailable,
    GenerationMismatch,
    ProbeFailed,
    /// The provisioned or configured raw SQL query identity failed its privilege check, so raw SQL is refused while materialization continues.
    QueryIdentityRejected,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct QueryCapabilities {
    pub dialects: Vec<SqlDialect>,
    pub consistency: Vec<Consistency>,
    pub logical_types: Vec<LogicalTypeKind>,
    pub raw_sql: bool,
    /// The aggregate functions the backend computes.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub aggregates: Vec<AggFunc>,
    #[serde(default)]
    pub vector_search: bool,
    #[serde(default)]
    pub text_search: bool,
}

impl Validate for QueryCapabilities {
    fn validate(&self) -> Result<(), InvalidError> {
        if self.dialects.is_empty() || self.consistency.is_empty() || self.logical_types.is_empty()
        {
            return Err(InvalidError::new(
                "query capabilities require dialect, consistency, and logical-type coverage",
            ));
        }
        ensure_unique(
            "query dialect",
            self.dialects.iter().map(|value| *value as u8),
        )?;
        ensure_unique(
            "query consistency",
            self.consistency.iter().map(|value| *value as u8),
        )?;
        ensure_unique(
            "query logical type",
            self.logical_types.iter().map(|value| *value as u8),
        )?;
        ensure_unique(
            "query aggregate",
            self.aggregates.iter().map(|value| *value as u8),
        )?;
        if self.raw_sql && self.dialects.is_empty() {
            return Err(InvalidError::new(
                "raw SQL capability requires an explicit dialect",
            ));
        }
        Ok(())
    }
}

fn ensure_unique(label: &str, values: impl IntoIterator<Item = u8>) -> Result<(), InvalidError> {
    let mut seen = BTreeSet::new();
    if values.into_iter().all(|value| seen.insert(value)) {
        Ok(())
    } else {
        Err(InvalidError::new(format!("backend repeats a {label}")))
    }
}

fn validate_descriptor_text(label: &str, value: &str) -> Result<(), InvalidError> {
    if value.is_empty() || value.len() > 4096 || value.chars().any(char::is_control) {
        return Err(InvalidError::new(format!(
            "{label} must contain 1..=4096 bytes without control characters"
        )));
    }
    Ok(())
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct SchemaCapabilities {
    pub logical_schema: bool,
    pub arrow_ipc_stream: bool,
    pub schema_evolution: bool,
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct BackendLimits {
    pub max_query_rows: u64,
    pub max_query_bytes: u64,
    pub max_scan_bytes: u64,
    pub max_query_micros: u64,
    pub max_concurrent_queries: u32,
    pub max_schema_fields: u32,
    /// The largest record body a whole-record row stores.
    #[serde(default)]
    pub max_payload_bytes: u64,
}

/// The managed backend's capability announcement to the streaming server, sent over their
/// private socket on connect (`AGDX_BACKEND_HELLO_CODE`). The streaming server caches the
/// `versions` and the advertised `backends`, and relays them verbatim when it answers a
/// client `AGDX_HELLO` / capabilities probe, so the streaming server never hardcodes feature
/// bits or backend identities the backend may or may not serve.
/// This makes the backend the single source of its own capability truth and
/// keeps the binary `features` bitset and the HTTP capability flags in agreement
/// with what is actually served. A separate type from [`HelloReply`] because the
/// direction and sender differ (backend to streaming server, not server to client).
const fn backend_ready_by_default() -> bool {
    true
}

const fn backend_is_ready(ready: &bool) -> bool {
    *ready
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[non_exhaustive]
pub struct BackendAnnounce {
    pub versions: OpVersions,
    #[serde(
        default = "backend_ready_by_default",
        skip_serializing_if = "backend_is_ready"
    )]
    pub ready: bool,
    /// Materialization backends the server currently exposes (the ones it has
    /// open). A client routes only to an advertised id. Empty (the default) is
    /// skipped on encode, so a pre-backends announce stays byte-identical and an
    /// older reader simply sees no advertised backends.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub backends: Vec<BackendDescriptor>,
    /// The stream/topic names this deployment uses. Absent (the default) is
    /// skipped on encode, so a pre-topology announce stays byte-identical and
    /// an older reader sees no advertised topology (falls back to its
    /// own [`WireTopology::default`]).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topology: Option<WireTopology>,
    /// The consumer-filter evaluation the streaming server serves, set with
    /// the `CONSUMER_FILTERS` bit. Absent from an older server, whose client
    /// then assumes its own evaluator.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub filters: Option<FilterAnnounce>,
}

/// The consumer-filter evaluation a server serves. A client refuses to run a
/// filter under another evaluator version or with a codec the server does not
/// list, so a version gap fails before the first page instead of filtering
/// differently.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct FilterAnnounce {
    pub evaluator_version: u32,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub codecs: Vec<FilterCodec>,
    /// Longest filtered wait the server holds. Zero means it serves no
    /// filtered wait, so a reader paces idle rounds itself.
    #[serde(default, skip_serializing_if = "is_zero_u32")]
    pub max_wait_ms: u32,
}

impl FilterAnnounce {
    /// What this build of the evaluator serves.
    pub fn served() -> Self {
        Self {
            evaluator_version: FILTER_EVALUATOR_VERSION,
            codecs: vec![
                FilterCodec::Json,
                FilterCodec::HeadersOnly,
                FilterCodec::Avro,
                FilterCodec::Protobuf,
                FilterCodec::Cbor,
            ],
            max_wait_ms: 0,
        }
    }

    /// Whether a filter with `evaluator_version` and `codec` evaluates here
    /// exactly as it would in the client that built it.
    pub fn evaluates(&self, evaluator_version: u32, codec: FilterCodec) -> bool {
        self.evaluator_version == evaluator_version && self.codecs.contains(&codec)
    }
}

impl BackendAnnounce {
    /// Constructor for the non-exhaustive wire struct.
    pub fn new(versions: OpVersions) -> Self {
        Self {
            versions,
            ready: true,
            backends: Vec::new(),
            topology: None,
            filters: None,
        }
    }

    /// Returns a copy announcing the served consumer-filter evaluation.
    #[must_use]
    pub fn with_filters(mut self, filters: FilterAnnounce) -> Self {
        self.filters = Some(filters);
        self
    }

    #[must_use]
    pub const fn unavailable(mut self) -> Self {
        self.ready = false;
        self
    }

    /// Returns a copy advertising `backends`.
    #[must_use]
    pub fn with_backends(mut self, backends: Vec<BackendDescriptor>) -> Self {
        self.backends = backends;
        self
    }

    /// Returns a copy advertising the deployment's `topology`.
    #[must_use]
    pub fn with_topology(mut self, topology: WireTopology) -> Self {
        self.topology = Some(topology);
        self
    }
}

impl Validate for BackendAnnounce {
    fn validate(&self) -> Result<(), InvalidError> {
        let mut resources = BTreeSet::new();
        for backend in &self.backends {
            backend.validate()?;
            if !resources.insert(backend.resource_id) {
                return Err(InvalidError::new(
                    "backend announcement repeats a resource id",
                ));
            }
        }
        if self.ready
            && !self.backends.is_empty()
            && !self.backends.iter().any(|backend| backend.readiness.ready)
        {
            return Err(InvalidError::new(
                "ready backend announcement must contain a ready backend",
            ));
        }
        Ok(())
    }
}

#[cfg(all(test, feature = "cbor"))]
mod tests {
    use super::*;
    use crate::codes::{CONTROL_OP_VERSION, FORK_OP_VERSION, KV_OP_VERSION, QUERY_OP_VERSION};
    use crate::framing::{decode_named, encode_named};

    fn backend(resource_id: u128, mode: BackendMode, label: &str, kind: &str) -> BackendDescriptor {
        BackendDescriptor::new(
            BackendResourceId::from_u128(resource_id),
            mode,
            label,
            BackendImplementation {
                kind: kind.to_owned(),
                version: "1.0.0".to_owned(),
            },
            1,
            1,
        )
    }

    #[test]
    fn given_a_hello_reply_when_round_tripped_then_should_preserve_versions() {
        // The pinned `HelloReply` shape (CBOR named fields). The connect-time
        // probe decodes exactly this shape.
        let reply = HelloReply::new(OpVersions::new(
            QUERY_OP_VERSION,
            CONTROL_OP_VERSION,
            KV_OP_VERSION,
            FORK_OP_VERSION,
        ));
        let bytes = encode_named(&reply).expect("hello reply serializes");
        let back: HelloReply = decode_named(&bytes).expect("hello reply deserializes");
        assert_eq!(back, reply);
    }

    #[test]
    fn given_a_backend_announce_when_round_tripped_then_should_preserve_features() {
        let announce = BackendAnnounce::new(
            OpVersions::new(
                QUERY_OP_VERSION,
                CONTROL_OP_VERSION,
                KV_OP_VERSION,
                FORK_OP_VERSION,
            )
            .with_features(feature::KV_CAS | feature::READ_YOUR_WRITES),
        );
        let bytes = encode_named(&announce).expect("serializes");
        let back: BackendAnnounce = decode_named(&bytes).expect("deserializes");
        assert_eq!(back, announce);
        assert!(back.versions.has_feature(feature::KV_CAS));
    }

    #[test]
    fn given_an_unavailable_backend_announce_when_round_tripped_then_should_stay_unavailable() {
        let announce = BackendAnnounce::new(OpVersions::new(
            QUERY_OP_VERSION,
            CONTROL_OP_VERSION,
            KV_OP_VERSION,
            FORK_OP_VERSION,
        ))
        .unavailable();
        let bytes = encode_named(&announce).expect("serializes");
        let back: BackendAnnounce = decode_named(&bytes).expect("deserializes");
        assert!(!back.ready);
    }

    #[test]
    fn given_an_empty_hello_body_when_decoded_then_should_yield_no_versions() {
        // Pre-versioned servers answer the probe with an empty body. The probe
        // treats a failed decode as "no versions advertised", never an error.
        assert!(decode_named::<HelloReply>(&[]).is_err());
    }

    #[test]
    fn given_advertised_backends_when_round_tripped_then_should_preserve_them_and_skip_empty() {
        let announce = BackendAnnounce::new(OpVersions::new(
            QUERY_OP_VERSION,
            CONTROL_OP_VERSION,
            KV_OP_VERSION,
            FORK_OP_VERSION,
        ))
        .with_backends(vec![
            backend(1, BackendMode::Operational, "Embedded", "embedded"),
            backend(
                2,
                BackendMode::Operational,
                "Analytics warehouse",
                "columnar",
            ),
        ]);
        let bytes = encode_named(&announce).expect("encodes");
        let back: BackendAnnounce = decode_named(&bytes).expect("decodes");
        assert_eq!(back, announce);
        assert_eq!(back.backends.len(), 2);
        assert_eq!(
            back.backends[1].resource_id,
            BackendResourceId::from_u128(2)
        );
        assert_eq!(back.backends[1].implementation.kind, "columnar");
        assert_eq!(back.backends[1].label, "Analytics warehouse");

        // No advertised backends (the default) is omitted on the wire, so a
        // pre-backends announce stays byte-identical.
        let plain = BackendAnnounce::new(OpVersions::new(1, 1, 1, 1));
        let json = serde_json::to_string(&plain).expect("json");
        assert!(!json.contains("backends"), "empty backends omitted: {json}");
    }

    #[test]
    fn given_announce_without_topology_when_decoded_then_should_default_none() {
        let announce = BackendAnnounce::new(OpVersions::new(
            QUERY_OP_VERSION,
            CONTROL_OP_VERSION,
            KV_OP_VERSION,
            FORK_OP_VERSION,
        ));
        let bytes = encode_named(&announce).expect("encodes");
        let back: BackendAnnounce = decode_named(&bytes).expect("decodes");
        assert_eq!(back.topology, None);
        let json = serde_json::to_string(&announce).expect("json");
        assert!(
            !json.contains("topology"),
            "absent topology omitted: {json}"
        );
    }

    #[test]
    fn given_announce_with_topology_when_round_tripped_then_should_preserve_it() {
        let custom = WireTopology {
            ops_stream: "custom-ops".to_owned(),
            ..WireTopology::default()
        };
        let announce = BackendAnnounce::new(OpVersions::new(
            QUERY_OP_VERSION,
            CONTROL_OP_VERSION,
            KV_OP_VERSION,
            FORK_OP_VERSION,
        ))
        .with_topology(custom.clone());
        let bytes = encode_named(&announce).expect("encodes");
        let back: BackendAnnounce = decode_named(&bytes).expect("decodes");
        assert_eq!(back.topology, Some(custom));
    }

    #[test]
    fn given_advertised_features_when_round_tripped_then_should_preserve_bits_and_skip_zero() {
        let versions = OpVersions::new(
            QUERY_OP_VERSION,
            CONTROL_OP_VERSION,
            KV_OP_VERSION,
            FORK_OP_VERSION,
        )
        .with_features(feature::KV_CAS | feature::READ_YOUR_WRITES);
        assert!(versions.has_feature(feature::KV_CAS));
        assert!(versions.has_feature(feature::READ_YOUR_WRITES));
        assert!(!versions.has_feature(feature::STRONG_CONSISTENCY));
        // has_feature on a combined mask requires every bit present.
        assert!(versions.has_feature(feature::KV_CAS | feature::READ_YOUR_WRITES));
        assert!(!versions.has_feature(feature::KV_CAS | feature::STRONG_CONSISTENCY));
        let reply = HelloReply::new(versions);
        let bytes = encode_named(&reply).expect("encodes");
        let back: HelloReply = decode_named(&bytes).expect("decodes");
        assert_eq!(back, reply);
        assert!(back.versions.has_feature(feature::READ_YOUR_WRITES));
        // No advertised feature (0) is omitted on the wire, so a pre-feature
        // hello reply stays byte-identical.
        let plain = HelloReply::new(OpVersions::new(1, 1, 1, 1));
        let json = serde_json::to_string(&plain).expect("json");
        assert!(!json.contains("features"), "zero features omitted: {json}");
    }

    #[test]
    fn given_announced_filter_evaluation_when_round_tripped_then_should_gate_codecs_and_versions() {
        let announce = BackendAnnounce::new(OpVersions::new(1, 1, 1, 1))
            .with_filters(FilterAnnounce::served());
        let bytes = encode_named(&announce).expect("serializes");
        let back: BackendAnnounce = decode_named(&bytes).expect("deserializes");
        let filters = back.filters.expect("the evaluation is announced");
        assert!(filters.evaluates(FILTER_EVALUATOR_VERSION, FilterCodec::Json));
        assert!(!filters.evaluates(FILTER_EVALUATOR_VERSION + 1, FilterCodec::Json));
        assert!(!filters.evaluates(FILTER_EVALUATOR_VERSION, FilterCodec::Unknown));
        let plain =
            encode_named(&BackendAnnounce::new(OpVersions::new(1, 1, 1, 1))).expect("serializes");
        let older: BackendAnnounce = decode_named(&plain).expect("deserializes");
        assert_eq!(older.filters, None);
    }

    #[test]
    fn given_the_feature_bits_when_read_then_should_match_the_pinned_values() {
        assert_eq!(
            [
                feature::KV_CAS,
                feature::READ_YOUR_WRITES,
                feature::STRONG_CONSISTENCY,
                feature::KV_CAS_FENCED,
                feature::AGENT_WORKFLOW,
                feature::KEYWORD_SEARCH,
                feature::WATCH,
                feature::AUTHZ,
                feature::DESTINATIONS,
                feature::KV_FENCED_LEASES,
                feature::CONSUMER_FILTERS,
                feature::GROUP_POLICY_READS,
                feature::SESSIONS,
                feature::STREAM_TENANCY,
                feature::MATERIALIZATIONS,
            ],
            [
                1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384
            ]
        );
    }

    #[test]
    fn given_op_versions_with_zero_surfaces_when_encoded_then_should_omit_them_and_keep_set_ones() {
        let advertised = OpVersions::new(1, 1, 1, 1)
            .with_agent(2)
            .with_graph(3)
            .with_filter(4);
        let json = serde_json::to_value(advertised).expect("encode advertised surfaces");
        let decoded: OpVersions = serde_json::from_value(json).expect("decode advertised surfaces");
        assert_eq!(decoded, advertised);
        let plain =
            serde_json::to_value(OpVersions::new(1, 1, 1, 1)).expect("encode base surfaces");
        assert_eq!(
            (plain.get("agent"), plain.get("graph"), plain.get("filter")),
            (None, None, None)
        );
    }

    #[test]
    fn given_a_ready_announce_with_a_filter_wait_when_encoded_then_should_keep_only_the_wait() {
        let waiting =
            BackendAnnounce::new(OpVersions::new(1, 1, 1, 1)).with_filters(FilterAnnounce {
                max_wait_ms: 250,
                ..FilterAnnounce::served()
            });
        let json = serde_json::to_value(&waiting).expect("encode a ready announce");
        assert!(
            json.get("ready").is_none(),
            "a ready announce omits ready: {json}"
        );
        let decoded: BackendAnnounce =
            serde_json::from_value(json).expect("decode a ready announce");
        assert_eq!(decoded, waiting);
        let served =
            serde_json::to_value(FilterAnnounce::served()).expect("encode the served evaluation");
        assert!(
            served.get("max_wait_ms").is_none(),
            "a zero wait is omitted: {served}"
        );
    }

    #[test]
    fn given_a_backend_descriptor_when_validated_then_should_refuse_each_broken_invariant() {
        let ready = ready_backend(1);
        ready.validate().expect("a ready backend with every limit");
        backend(1, BackendMode::Operational, "Embedded", "embedded")
            .validate()
            .expect("a backend that is not ready");
        BackendDescriptor {
            label: "l".repeat(4096),
            ..ready.clone()
        }
        .validate()
        .expect("a label at the cap");
        let refused = [
            (
                "another descriptor version",
                BackendDescriptor {
                    descriptor_version: BACKEND_DESCRIPTOR_VERSION + 1,
                    ..ready.clone()
                },
            ),
            (
                "a zero resource id",
                BackendDescriptor {
                    resource_id: BackendResourceId::from_u128(0),
                    ..ready.clone()
                },
            ),
            (
                "a zero backend generation",
                BackendDescriptor {
                    observed_backend_generation: 0,
                    ..ready.clone()
                },
            ),
            (
                "a zero configuration revision",
                BackendDescriptor {
                    runtime_configuration_revision: 0,
                    ..ready.clone()
                },
            ),
            (
                "an empty label",
                BackendDescriptor {
                    label: String::new(),
                    ..ready.clone()
                },
            ),
            (
                "a label past the cap",
                BackendDescriptor {
                    label: "l".repeat(4097),
                    ..ready.clone()
                },
            ),
            (
                "a ready backend still starting",
                BackendDescriptor {
                    observed_state: BackendObservedState::Starting,
                    ..ready.clone()
                },
            ),
            (
                "a logical schema without a field limit",
                BackendDescriptor {
                    limits: BackendLimits {
                        max_schema_fields: 0,
                        ..ready.limits.clone()
                    },
                    ..ready.clone()
                },
            ),
        ];
        for (case, descriptor) in refused {
            assert!(descriptor.validate().is_err(), "{case} must be refused");
        }
        let limits = &ready.limits;
        let zero_query_limits = [
            BackendLimits {
                max_query_rows: 0,
                ..limits.clone()
            },
            BackendLimits {
                max_query_bytes: 0,
                ..limits.clone()
            },
            BackendLimits {
                max_scan_bytes: 0,
                ..limits.clone()
            },
            BackendLimits {
                max_query_micros: 0,
                ..limits.clone()
            },
            BackendLimits {
                max_concurrent_queries: 0,
                ..limits.clone()
            },
        ];
        for limits in zero_query_limits {
            let descriptor = BackendDescriptor {
                limits,
                ..ready.clone()
            };
            assert!(
                descriptor.validate().is_err(),
                "a ready query backend needs every query limit"
            );
        }
    }

    #[test]
    fn given_backend_readiness_when_validated_then_should_need_a_time_or_a_bounded_reason_list() {
        let not_ready = |reasons: usize| BackendReadiness {
            ready: false,
            reasons: vec![
                BackendReadinessReason {
                    code: BackendReadinessCode::ProbeFailed,
                    detail: None,
                };
                reasons
            ],
            observed_at_micros: 0,
        };
        BackendReadiness::ready(5)
            .validate()
            .expect("ready with an observation time");
        not_ready(32).validate().expect("reasons at the cap");
        assert!(not_ready(33).validate().is_err(), "reasons past the cap");
        assert!(
            BackendReadiness::ready(0).validate().is_err(),
            "ready without an observation time"
        );
    }

    #[test]
    fn given_query_capabilities_when_validated_then_should_need_coverage_without_repeats() {
        capabilities()
            .validate()
            .expect("full coverage with raw SQL");
        let refused = [
            (
                "no dialect",
                QueryCapabilities {
                    dialects: Vec::new(),
                    raw_sql: false,
                    ..capabilities()
                },
            ),
            (
                "no consistency level",
                QueryCapabilities {
                    consistency: Vec::new(),
                    ..capabilities()
                },
            ),
            (
                "no logical type",
                QueryCapabilities {
                    logical_types: Vec::new(),
                    ..capabilities()
                },
            ),
            (
                "a repeated dialect",
                QueryCapabilities {
                    dialects: vec![SqlDialect::Postgres, SqlDialect::Postgres],
                    ..capabilities()
                },
            ),
        ];
        for (case, rejected) in refused {
            assert!(rejected.validate().is_err(), "{case} must be refused");
        }
    }

    #[test]
    fn given_a_backend_announce_when_validated_then_should_need_unique_ids_and_a_ready_backend() {
        let versions = OpVersions::new(1, 1, 1, 1);
        let not_ready = backend(2, BackendMode::Operational, "Analytics", "columnar");
        BackendAnnounce::new(versions)
            .validate()
            .expect("a ready announce without backends");
        BackendAnnounce::new(versions)
            .with_backends(vec![ready_backend(1), not_ready.clone()])
            .validate()
            .expect("a ready announce with a ready backend");
        assert!(
            BackendAnnounce::new(versions)
                .with_backends(vec![not_ready])
                .validate()
                .is_err(),
            "a ready announce without a ready backend"
        );
        assert!(
            BackendAnnounce::new(versions)
                .with_backends(vec![ready_backend(1), ready_backend(1)])
                .validate()
                .is_err(),
            "a repeated resource id"
        );
    }

    fn ready_backend(resource_id: u128) -> BackendDescriptor {
        backend(
            resource_id,
            BackendMode::Operational,
            "Embedded",
            "embedded",
        )
        .with_state(
            BackendDesiredState::Enabled,
            BackendObservedState::Ready,
            BackendReadiness::ready(5),
        )
        .with_query(capabilities())
        .with_schema(SchemaCapabilities {
            logical_schema: true,
            arrow_ipc_stream: false,
            schema_evolution: false,
        })
        .with_limits(BackendLimits {
            max_query_rows: 1000,
            max_query_bytes: 1_048_576,
            max_scan_bytes: 1_073_741_824,
            max_query_micros: 1_000_000,
            max_concurrent_queries: 8,
            max_schema_fields: 64,
            max_payload_bytes: 0,
        })
    }

    fn capabilities() -> QueryCapabilities {
        QueryCapabilities {
            dialects: vec![SqlDialect::Postgres],
            consistency: vec![Consistency::Eventual],
            logical_types: vec![LogicalTypeKind::String],
            raw_sql: true,
            aggregates: vec![AggFunc::Count],
            vector_search: false,
            text_search: false,
        }
    }
}
const fn is_zero_u32(value: &u32) -> bool {
    *value == 0
}