laser_wire/

http.rs

use crate::checkpoint::{
    CheckpointRequestEnvelope, DestinationCheckpointStatus, PreparedAttemptSummary, RetentionGap,
};
use crate::destination::{
    DestinationId, DestinationOperationId, MaterializationDestination, QueryRoute,
};
use crate::fork::{ForkError, ForkKind};
use crate::graph::SourceRef;
use crate::hello::{BackendDescriptor, OpVersions};
use crate::kv::KvError;
use crate::query::{Consistency, QueryError, QueryExecutionId, QueryExecutionStatus, QueryResult};
use crate::result::ResultCode;
use crate::schema::{Digest32, LogicalSchema, TypedValue, UuidValue};
use crate::topology::WireTopology;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// `GET /agdx/capabilities`: the feature-detection probe.
pub const CAPABILITIES_PATH: &str = "/agdx/capabilities";
/// `POST /agdx/query` (and `GET` with the query as a parameter).
pub const QUERY_PATH: &str = "/agdx/query";
/// `GET` and `POST /agdx/destinations`.
pub const DESTINATIONS_PATH: &str = "/agdx/destinations";
/// `GET /agdx/query-routes`.
pub const QUERY_ROUTES_PATH: &str = "/agdx/query-routes";
/// `GET /agdx/projections` to list, `POST` to register.
pub const PROJECTIONS_PATH: &str = "/agdx/projections";
/// `POST /agdx/bindings` to apply, `DELETE` to remove.
pub const BINDINGS_PATH: &str = "/agdx/bindings";
/// `GET /agdx/schemas` to list, `POST` to register.
pub const SCHEMAS_PATH: &str = "/agdx/schemas";
/// `GET /agdx/kv` to list the caller's namespaces.
pub const KV_PATH: &str = "/agdx/kv";
/// `GET /agdx/forks` to list, `POST` to create.
pub const FORKS_PATH: &str = "/agdx/forks";
/// `GET /agdx/graphs` to list graph projections, `POST` to register.
pub const GRAPHS_PATH: &str = "/agdx/graphs";
/// `GET /agdx/clients` to list live connections with their advertised metadata,
/// filtered and paginated by query parameters. The HTTP face of the
/// `AGDX_GET_CLIENTS_METADATA` discovery read.
pub const CLIENTS_PATH: &str = "/agdx/clients";
/// `GET /agdx/runs` to list runs (filtered and paged, [`RunsQuery`]), `POST`
/// to submit one (a JSON `AgentSubmit` body). The HTTP face of the
/// `AGDX_AGENT_*` run-registry band.
pub const RUNS_PATH: &str = "/agdx/runs";
/// `GET /agdx/authz/whoami`: the caller's effective governance roles and grants.
pub const AUTHZ_WHOAMI_PATH: &str = "/agdx/authz/whoami";
/// `GET /agdx/authz/roles`: list governance roles.
pub const AUTHZ_ROLES_PATH: &str = "/agdx/authz/roles";

pub fn destination_path(id: DestinationId) -> String {
    format!("{DESTINATIONS_PATH}/{id}")
}

pub fn destination_enable_path(id: DestinationId) -> String {
    format!("{}/{id}/enable", DESTINATIONS_PATH)
}

pub fn destination_disable_path(id: DestinationId) -> String {
    format!("{}/{id}/disable", DESTINATIONS_PATH)
}

pub fn destination_status_path(id: DestinationId) -> String {
    format!("{}/{id}/status", DESTINATIONS_PATH)
}

pub fn destination_checkpoint_path(id: DestinationId) -> String {
    format!("{}/{id}/checkpoint", DESTINATIONS_PATH)
}

pub fn destination_retention_gap_path(id: DestinationId) -> String {
    format!("{}/{id}/retention-gap", DESTINATIONS_PATH)
}

pub fn destination_prepared_attempt_path(id: DestinationId) -> String {
    format!("{}/{id}/prepared-attempt", DESTINATIONS_PATH)
}

pub fn destination_table_path(id: DestinationId) -> String {
    format!("{}/{id}/table", DESTINATIONS_PATH)
}

pub fn destination_table_schema_path(id: DestinationId) -> String {
    format!("{}/{id}/table/schema", DESTINATIONS_PATH)
}

pub fn destination_snapshots_path(id: DestinationId) -> String {
    format!("{}/{id}/table/snapshots", DESTINATIONS_PATH)
}

pub fn destination_current_snapshot_path(id: DestinationId) -> String {
    format!("{}/{id}/table/current-snapshot", DESTINATIONS_PATH)
}

pub fn destination_snapshot_path(id: DestinationId, snapshot_id: i64) -> String {
    format!("{}/{id}/table/snapshots/{snapshot_id}", DESTINATIONS_PATH)
}

pub fn destination_files_path(id: DestinationId) -> String {
    format!("{}/{id}/table/files", DESTINATIONS_PATH)
}

pub fn destination_metrics_path(id: DestinationId) -> String {
    format!("{}/{id}/table/metrics", DESTINATIONS_PATH)
}

pub fn query_execution_path(id: QueryExecutionId) -> String {
    format!("{QUERY_PATH}/{id}")
}

pub fn query_page_path(id: QueryExecutionId) -> String {
    format!("{QUERY_PATH}/{id}/pages")
}

pub fn query_cancel_path(id: QueryExecutionId) -> String {
    format!("{QUERY_PATH}/{id}/cancel")
}

pub fn destination_operation_path(id: DestinationOperationId) -> String {
    format!("{DESTINATIONS_PATH}/operations/{id}")
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct DestinationListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_stream: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source_topic: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name_contains: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<usize>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DestinationPageView {
    pub destinations: Vec<DestinationView>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,
    pub global_state_revision: u64,
    pub consistency: crate::checkpoint::CheckpointReadConsistency,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DestinationView {
    pub destination: MaterializationDestination,
    pub status: DestinationCheckpointStatus,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct DestinationMutationBody {
    pub checkpoint: CheckpointRequestEnvelope,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct AcceptedOperationView {
    pub operation_id: DestinationOperationId,
    pub request_id: crate::checkpoint::CheckpointRequestId,
    pub state: OperationState,
    pub submitted_at_micros: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub completed_at_micros: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error: Option<OperationErrorView>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct OperationErrorView {
    pub code: ResultCode,
    pub message: String,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum OperationState {
    Accepted,
    Running,
    Succeeded,
    Failed,
    Cancelled,
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueryRouteListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name_contains: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<usize>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueryRoutePageView {
    pub routes: Vec<QueryRoute>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,
    pub definition_revision: u64,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TableView {
    pub table_uuid: UuidValue,
    pub destination_id: DestinationId,
    pub destination_generation: u64,
    pub namespace: Vec<String>,
    pub table: String,
    pub current_snapshot_id: i64,
    pub current_schema_id: i32,
    pub current_partition_spec_id: i32,
    pub metadata_identity: String,
    pub properties: BTreeMap<String, String>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TableSchemaView {
    pub table_uuid: UuidValue,
    pub iceberg_schema_id: i32,
    pub logical_schema: LogicalSchema,
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SnapshotListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before_snapshot_id: Option<i64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<usize>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SnapshotPageView {
    pub snapshots: Vec<TableSnapshotView>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_before_snapshot_id: Option<i64>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TableSnapshotView {
    pub snapshot_id: i64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parent_snapshot_id: Option<i64>,
    pub sequence_number: i64,
    pub committed_at_micros: u64,
    pub schema_id: i32,
    pub partition_spec_id: i32,
    pub materialization_boundary_digest: Digest32,
    pub checkpoint_revision: u64,
    pub summary: BTreeMap<String, String>,
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct TableFileListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub snapshot_id: Option<i64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<usize>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct TableFilePageView {
    pub files: Vec<TableFileView>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<String>,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct TableFileView {
    pub object_identity: String,
    pub file_size_bytes: u64,
    pub row_count: u64,
    pub partition: BTreeMap<String, TypedValue>,
    pub lower_bounds: BTreeMap<u32, TypedValue>,
    pub upper_bounds: BTreeMap<u32, TypedValue>,
    pub null_value_counts: BTreeMap<u32, u64>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct TableMetricsView {
    pub snapshot_id: i64,
    pub data_file_count: u64,
    pub delete_file_count: u64,
    pub total_rows: u64,
    pub total_bytes: u64,
    pub partition_count: u64,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueryPageBody {
    pub cursor: String,
    pub deadline_micros: u64,
}

#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct QueryExecutionView {
    pub status: QueryExecutionStatus,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result: Option<QueryResult>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DestinationIssueView {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retention_gap: Option<RetentionGap>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prepared_attempt: Option<PreparedAttemptSummary>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub block: Option<crate::checkpoint::DestinationBlock>,
    pub checkpoint_revision: u64,
    pub consistency: crate::checkpoint::CheckpointReadConsistency,
}

/// `GET`/`PUT`/`DELETE /agdx/authz/roles/{name}`.
pub fn authz_role_path(name: &str) -> String {
    format!("{AUTHZ_ROLES_PATH}/{name}")
}

/// `GET`/`PUT /agdx/authz/users/{id}/roles`: read or replace one user's role set.
pub fn authz_user_roles_path(user_id: u32) -> String {
    format!("/agdx/authz/users/{user_id}/roles")
}

/// `DELETE`/`GET /agdx/graphs/{id}`: drop or read a graph projection.
pub fn graph_path(id: &str) -> String {
    format!("{GRAPHS_PATH}/{id}")
}

/// `POST /agdx/graph/{name}/query`: run a traversal (a `GraphQuery` body).
pub fn graph_query_path(name: &str) -> String {
    format!("/agdx/graph/{name}/query")
}

/// `GET /agdx/graph/{name}/neighbors/{node}`: one-hop neighbor read.
pub fn graph_neighbors_path(name: &str, node: &str) -> String {
    format!("/agdx/graph/{name}/neighbors/{node}")
}

/// `GET`/`DELETE /agdx/projections/{id}`.
pub fn projection_path(id: &str) -> String {
    format!("{PROJECTIONS_PATH}/{id}")
}

/// `GET`/`DELETE /agdx/schemas/{id}`.
pub fn schema_path(id: u32) -> String {
    format!("{SCHEMAS_PATH}/{id}")
}

/// `POST /agdx/schemas/{id}/decode`.
pub fn schema_decode_path(id: u32) -> String {
    format!("{SCHEMAS_PATH}/{id}/decode")
}

/// `GET /agdx/kv/{namespace}` to scan, `DELETE` to bulk-delete.
pub fn kv_namespace_path(namespace: &str) -> String {
    format!("{KV_PATH}/{namespace}")
}

/// `GET`/`PUT`/`DELETE /agdx/kv/{namespace}/{key}`. `key` is the URL-safe
/// unpadded base64 form of the key bytes, the encoding this surface uses for
/// every binary body.
pub fn kv_entry_path(namespace: &str, key_b64: &str) -> String {
    format!("{KV_PATH}/{namespace}/{key_b64}")
}

/// `PUT /agdx/kv/{namespace}/{key}/cas`: a conditional write (compare-and-swap).
/// The precondition rides the query string (`expect_version` or `expect_absent`)
/// and the value rides the raw body, like the plain `PUT`. A success replies
/// `CasCommittedView` with the new version, a precondition miss replies `409`
/// with an `ErrorBody` of code `conflict` whose `detail` carries the current
/// version.
pub fn kv_cas_path(namespace: &str, key_b64: &str) -> String {
    format!("{KV_PATH}/{namespace}/{key_b64}/cas")
}

/// `DELETE /agdx/forks/{id}`.
pub fn fork_path(id: &str) -> String {
    format!("{FORKS_PATH}/{id}")
}

/// `POST /agdx/forks/{id}/promote`.
pub fn fork_promote_path(id: &str) -> String {
    format!("{FORKS_PATH}/{id}/promote")
}

/// `PUT /agdx/forks/{id}/rows`.
pub fn fork_rows_path(id: &str) -> String {
    format!("{FORKS_PATH}/{id}/rows")
}

/// `GET /agdx/runs/{id}`: read one run's status.
pub fn run_path(id: &str) -> String {
    format!("{RUNS_PATH}/{id}")
}

/// `POST /agdx/runs/{id}/cancel`: record the cancel intent on a run.
pub fn run_cancel_path(id: &str) -> String {
    format!("{RUNS_PATH}/{id}/cancel")
}

/// `GET /agdx/capabilities` reply: what the `/agdx/*` surface offers on this
/// server, so a browser client can feature-detect before showing the
/// projections / query / KV / fork views. Richer than the binary `AGDX_HELLO`
/// probe (per-surface flags plus the wire op versions the JSON bodies must
/// match), and it answers truthfully even when the managed backend is disabled (200
/// with `managed: false`).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[non_exhaustive]
pub struct Capabilities {
    /// Connected to a managed plane at all (the root: with no plane every managed
    /// surface below is off, and the reply still answers `200` with `managed:
    /// false`).
    pub managed: bool,
    /// The managed query surface, its registry browse views, and the strongest
    /// read-consistency it serves.
    pub query: QueryCapsView,
    /// Destination declarations, checkpoint status, and table inspection.
    pub destinations: DestinationCapsView,
    /// The managed key-value surface and its conditional-write support.
    pub kv: KvCapsView,
    /// Whether the knowledge-graph ops (traversal, neighbors) are served. The
    /// agentic-memory API composes the query and graph surfaces, so it has no
    /// flag of its own: a client reads `query` and `graph`.
    #[serde(default)]
    pub graph: bool,
    /// Whether copy-on-write forks are served.
    pub fork: bool,
    /// Whether the agent and workflow control band is served. Off until the plane
    /// serves it (the engine is a later phase).
    #[serde(default)]
    pub agent_workflow: bool,
    /// Whether the change feed is published (one change record per committed
    /// notifying projector batch on the changes topic). Off by default: a
    /// client that waits on an unpublished feed would wait forever.
    #[serde(default)]
    pub watch: bool,
    /// Whether the authorization control band (`AGDX_AUTHZ_*`) is served, so a
    /// console can show its roles and bindings surface. Off by default.
    #[serde(default)]
    pub authz: bool,
    pub versions: OpVersions,
    /// Materialization backends the server currently exposes, so a client can
    /// show what it may route to. Identity only (id + engine kind), no settings
    /// or secrets. Empty (the default) is skipped on encode, so a pre-backends
    /// capabilities reply stays byte-identical.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub backends: Vec<BackendDescriptor>,
    /// The stream/topic names this deployment uses, mirroring
    /// [`crate::hello::BackendAnnounce::topology`]. Absent (the default) is
    /// skipped on encode, so a pre-topology capabilities reply stays
    /// byte-identical and an older reader falls back to
    /// [`WireTopology::default`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topology: Option<WireTopology>,
}

/// The managed query surface on the HTTP capabilities reply: whether it is
/// served, its registry browse views, and the consistency it honors.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueryCapsView {
    /// Whether `POST /agdx/query` is served.
    pub available: bool,
    /// Whether the projection registry browse routes are served.
    pub projections: bool,
    /// Whether the schema registry browse routes are served.
    pub schemas: bool,
    /// The strongest read-consistency the surface serves (the ladder
    /// `eventual < read_your_writes < strong`, so a level implies the weaker
    /// ones). Defaults to `eventual`, which every query surface serves.
    #[serde(default)]
    pub consistency: Consistency,
    /// Whether lexical relevance search (`Query.text`) is served. Defaults
    /// off, like every sub-feature: over-advertising would turn the clean
    /// unsupported into a silent wrong answer.
    #[serde(default)]
    pub keyword: bool,
    /// Whether opaque cursor page retrieval is served.
    #[serde(default)]
    pub cursor_paging: bool,
    /// Whether an executing query can be cancelled by its execution identity.
    #[serde(default)]
    pub cancellation: bool,
    /// Whether execution status can be polled independently of the first page.
    #[serde(default)]
    pub execution_status: bool,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DestinationCapsView {
    pub available: bool,
    #[serde(default)]
    pub lifecycle: bool,
    #[serde(default)]
    pub checkpoint_status: bool,
    #[serde(default)]
    pub query_routes: bool,
    #[serde(default)]
    pub table_schema: bool,
    #[serde(default)]
    pub snapshots: bool,
    #[serde(default)]
    pub files: bool,
    #[serde(default)]
    pub metrics: bool,
    #[serde(default)]
    pub strongest_consistency: crate::checkpoint::CheckpointReadConsistency,
}

/// The managed key-value surface on the HTTP capabilities reply.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvCapsView {
    /// Whether the get/set/scan routes are served.
    pub available: bool,
    /// Whether compare-and-swap (`AGDX_KV_CAS`) is served. Independent of plain
    /// get/set: a backend that cannot do a conditional write leaves it off and a
    /// `cas` returns a clean unsupported error.
    #[serde(default)]
    pub cas: bool,
    /// Whether fenced compare-and-swap (`AGDX_KV_CAS_FENCED`) is served. Independent
    /// of plain `cas`: a backend leaves it off when it cannot gate a write on a live
    /// fence sequence.
    #[serde(default)]
    pub cas_fenced: bool,
    /// Whether the full revocable fenced-lease contract is served: holder-scoped
    /// acquire, renewal, release, live-lease fenced CAS, and barriered reads.
    #[serde(default)]
    pub fenced_leases: bool,
}

impl Capabilities {
    /// Constructor for the non-exhaustive wire struct. The core surfaces track
    /// `enabled`, the way the binary `AGDX_HELLO` probe answers. The per-surface
    /// sub-features (`kv.cas`, `query.consistency` above `eventual`, `graph`) all
    /// start off: a server must opt into each, never over-advertising (which would
    /// turn a clean unsupported error into a silent wrong answer). Build them with
    /// [`from_versions`](Self::from_versions) or the setters.
    pub fn new(enabled: bool, versions: OpVersions) -> Self {
        Self {
            managed: enabled,
            query: QueryCapsView {
                available: enabled,
                projections: enabled,
                schemas: enabled,
                consistency: Consistency::Eventual,
                keyword: false,
                cursor_paging: false,
                cancellation: false,
                execution_status: false,
            },
            destinations: DestinationCapsView {
                available: false,
                lifecycle: false,
                checkpoint_status: false,
                query_routes: false,
                table_schema: false,
                snapshots: false,
                files: false,
                metrics: false,
                strongest_consistency:
                    crate::checkpoint::CheckpointReadConsistency::PotentiallyStale,
            },
            kv: KvCapsView {
                available: enabled,
                cas: false,
                cas_fenced: false,
                fenced_leases: false,
            },
            graph: false,
            fork: enabled,
            agent_workflow: false,
            watch: false,
            authz: false,
            versions,
            backends: Vec::new(),
            topology: None,
        }
    }

    /// Advertise that the knowledge-graph ops are served. Off by default: a
    /// server sets it only when a backend implements the graph surface.
    #[must_use]
    pub fn with_graph(mut self, value: bool) -> Self {
        self.graph = value;
        self
    }

    /// Advertise that the agent and workflow control band is served. Off by
    /// default: a server sets it only when it serves the band.
    #[must_use]
    pub fn with_agent_workflow(mut self, value: bool) -> Self {
        self.agent_workflow = value;
        self
    }

    /// Set whether lexical keyword search is served on the query surface.
    #[must_use]
    pub fn with_query_keyword(mut self, value: bool) -> Self {
        self.query.keyword = value;
        self
    }

    #[must_use]
    pub fn with_query_execution(mut self, paging: bool, cancellation: bool, status: bool) -> Self {
        self.query.cursor_paging = paging;
        self.query.cancellation = cancellation;
        self.query.execution_status = status;
        self
    }

    #[must_use]
    pub fn with_destinations(mut self, destinations: DestinationCapsView) -> Self {
        self.destinations = destinations;
        self
    }

    /// Advertise that the change feed is published. Only a deployment whose
    /// projector emits change records may set it.
    #[must_use]
    pub fn with_watch(mut self, value: bool) -> Self {
        self.watch = value;
        self
    }

    /// Advertise that the authorization control band is served.
    #[must_use]
    pub fn with_authz(mut self, value: bool) -> Self {
        self.authz = value;
        self
    }

    /// Advertise the materialization backends the server exposes. The wire pins
    /// no engine, so a server lists whatever it has open by id and kind.
    #[must_use]
    pub fn with_backends(mut self, backends: Vec<BackendDescriptor>) -> Self {
        self.backends = backends;
        self
    }

    /// Advertise the deployment's `topology`, mirroring
    /// [`crate::hello::BackendAnnounce::with_topology`].
    #[must_use]
    pub fn with_topology(mut self, topology: WireTopology) -> Self {
        self.topology = Some(topology);
        self
    }

    /// Advertise compare-and-swap on the KV surface (`AGDX_KV_CAS`). Only a
    /// backend that does a genuine conditional write may set it.
    #[must_use]
    pub fn with_kv_cas(mut self, on: bool) -> Self {
        self.kv.cas = on;
        self
    }

    /// Advertise fenced compare-and-swap on the KV surface (`AGDX_KV_CAS_FENCED`).
    /// Only a backend that gates a write on a live fence sequence may set it.
    #[must_use]
    pub fn with_kv_cas_fenced(mut self, on: bool) -> Self {
        self.kv.cas_fenced = on;
        self
    }

    /// Advertise the full revocable fenced-lease contract. This subsumes the
    /// older fence-only `cas_fenced` capability.
    #[must_use]
    pub fn with_kv_fenced_leases(mut self, on: bool) -> Self {
        self.kv.fenced_leases = on;
        self.kv.cas_fenced |= on;
        self
    }

    /// Advertise the strongest read-consistency the query surface serves.
    #[must_use]
    pub fn with_query_consistency(mut self, level: Consistency) -> Self {
        self.query.consistency = level;
        self
    }

    /// Build the HTTP capabilities from the same `OpVersions` the binary
    /// `AGDX_HELLO` probe answers with, reading the per-surface sub-features
    /// straight off its `features` bitset and `graph` op version. A server SHOULD
    /// use this so its two capability carriages (the binary `features` bits and
    /// these HTTP fields) cannot disagree: the one source drives both.
    pub fn from_versions(enabled: bool, versions: OpVersions) -> Self {
        use crate::hello::feature;
        let consistency = if versions.has_feature(feature::STRONG_CONSISTENCY) {
            Consistency::Strong
        } else if versions.has_feature(feature::READ_YOUR_WRITES) {
            Consistency::ReadYourWrites
        } else {
            Consistency::Eventual
        };
        let destinations_available =
            enabled && versions.checkpoint > 0 && versions.has_feature(feature::DESTINATIONS);
        Self::new(enabled, versions)
            .with_kv_cas(versions.has_feature(feature::KV_CAS))
            .with_kv_cas_fenced(versions.has_feature(feature::KV_CAS_FENCED))
            .with_kv_fenced_leases(versions.has_feature(feature::KV_FENCED_LEASES))
            .with_agent_workflow(versions.has_feature(feature::AGENT_WORKFLOW))
            .with_query_keyword(versions.has_feature(feature::KEYWORD_SEARCH))
            .with_watch(versions.has_feature(feature::WATCH))
            .with_authz(versions.has_feature(feature::AUTHZ))
            .with_query_consistency(consistency)
            .with_destinations(DestinationCapsView {
                available: destinations_available,
                lifecycle: destinations_available,
                checkpoint_status: destinations_available,
                query_routes: destinations_available,
                table_schema: false,
                snapshots: false,
                files: false,
                metrics: false,
                strongest_consistency:
                    crate::checkpoint::CheckpointReadConsistency::PotentiallyStale,
            })
            // The graph surface needs a backend that serves it, advertised as a
            // non-zero graph op version, so it is gated on that rather than implied.
            .with_graph(enabled && versions.graph > 0)
    }
}

/// One KV entry on the HTTP surface. `key` and `value` are URL-safe unpadded
/// base64, because keys and values are arbitrary bytes that JSON strings
/// cannot carry raw.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvEntryView {
    pub key: String,
    pub value: String,
    pub expires_at_micros: Option<u64>,
    /// The memory scope, present only on a memory read-view row (kind, agent,
    /// user, app, conversation, and a `SourceRef` source pointer to the origin
    /// log record). Absent on a generic entry, so a UI can fold a recalled item
    /// back to its source message. Skipped on the wire when absent.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub scope: Option<crate::kv::MemoryRowScope>,
    /// The origin log record this entry was folded from, mirroring
    /// [`crate::kv::KvEntry::source`]: every managed write is log-first, so a
    /// stored entry points back to the record that wrote it. Skipped when the
    /// store did not stamp one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source: Option<crate::graph::SourceRef>,
}

/// One KV scan page on the HTTP surface.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvPageView {
    pub entries: Vec<KvEntryView>,
    pub cursor: Option<String>,
}

/// One page of runs on the HTTP surface: the rows (the binary `AgentRunInfo`
/// is already JSON-safe) plus the next-page cursor as URL-safe unpadded
/// base64, like every binary value on this surface. Absent cursor means the
/// last page.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunPageView {
    pub runs: Vec<crate::agent_workflow::AgentRunInfo>,
    pub cursor: Option<String>,
}

/// `DELETE /agdx/kv/{namespace}` reply: the number of entries removed.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DeletedManyView {
    pub deleted: usize,
}

/// One connection on the `GET /agdx/clients` discovery surface. `metadata` is
/// URL-safe unpadded base64 of the opaque advertised bytes (a JSON string cannot
/// carry raw bytes), or `None` when the connection advertised none. The console
/// decodes and interprets it per producer kind (an agent card or an app blob).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClientMetadataView {
    pub client_id: u32,
    pub user_id: Option<u32>,
    pub transport: u8,
    pub address: String,
    pub consumer_groups_count: u32,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metadata: Option<String>,
}

/// One page of the `GET /agdx/clients` discovery read. `next_cursor` is the
/// `after` query parameter for the next page, or `None` on the last page.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClientMetadataListView {
    pub clients: Vec<ClientMetadataView>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub next_cursor: Option<u32>,
}

/// `GET /agdx/clients` query parameters: the discovery filters and the page
/// window, all optional. Shared by the server handler and the typed client so the
/// query string cannot drift between them.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ClientsQuery {
    /// Only return connections that advertised metadata.
    #[serde(default, skip_serializing_if = "core::ops::Not::not")]
    pub with_metadata_only: bool,
    /// Only return connections authenticated as this principal.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub user_id: Option<u32>,
    /// Page cursor: only connections with `client_id` greater than this.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after: Option<u32>,
    /// Max entries per page (clamped server-side to the page cap).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<u32>,
}

/// `POST /agdx/forks/{id}/promote` reply: the number of rows applied.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PromotedView {
    pub rows: usize,
}

/// One graph node on the HTTP surface: id as a string, its labels, and its
/// attributes rendered as strings (e.g. the entity `value`) for a browser or
/// wasm client that has no access to the typed `Value`.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct GraphNodeView {
    pub id: String,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub labels: Vec<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub attrs: Vec<(String, String)>,
    /// The source this node was first observed in, if known. See [`SourceRef`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source: Option<SourceRef>,
}

/// One graph edge on the HTTP surface: endpoint ids as strings, the type, and the
/// weight.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct GraphEdgeView {
    pub id: String,
    pub from: String,
    pub to: String,
    pub edge_type: String,
    pub weight: f32,
    /// Valid-time window (epoch micros) for a bitemporal edge, omitted when open.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub valid_from: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub valid_to: Option<u64>,
    /// The source that most recently asserted this relationship, if known. See
    /// [`SourceRef`].
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub source: Option<SourceRef>,
}

/// `POST /agdx/graph/{name}/query` reply: the reachable nodes, traversed edges,
/// and (for a `paths` return) the reconstructed paths as id sequences.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct GraphResultView {
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub nodes: Vec<GraphNodeView>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub edges: Vec<GraphEdgeView>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub paths: Vec<PathView>,
}

/// One path in a `GraphResultView`: parallel node and edge id sequences, ids as
/// Crockford-base32 strings (the JSON view of [`crate::graph::Path`]).
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct PathView {
    pub nodes: Vec<String>,
    pub edges: Vec<String>,
}

/// `POST /agdx/schemas` body: the register request without an id. The managed
/// backend allocates it and the reply carries it back as
/// `{"SchemaRegistered":id}`.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct RegisterSchemaBody {
    pub source: crate::control::SchemaSource,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub version: Option<u32>,
}

/// `POST /agdx/schemas/{id}/decode` body: the record payload as URL-safe
/// unpadded base64.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DecodeRecordBody {
    pub payload: String,
}

/// `POST /agdx/forks` body.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ForkCreateBody {
    pub fork_id: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parent: Option<String>,
    #[serde(default)]
    pub kind: ForkKind,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub tables: Vec<String>,
}

/// `PUT /agdx/forks/{id}/rows` body. `payload_b64` is URL-safe unpadded base64,
/// like every binary body on this surface.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ForkPutBody {
    pub table: String,
    pub partition_id: u32,
    pub offset: u64,
    #[serde(default)]
    pub projection_id: String,
    #[serde(default)]
    pub projection_version: u32,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub fields: BTreeMap<String, String>,
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub metadata: BTreeMap<String, String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub payload_b64: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub embedding: Option<String>,
    #[serde(default)]
    pub tombstone: bool,
}

/// `DELETE /agdx/bindings` body: which binding to remove, by its source stream
/// and topic. `projection_ref` absent removes the whole binding for that source.
/// `projection_ref` present removes only that one projection from the binding,
/// leaving the rest. Mirrors `ControlCommand::RemoveBinding`.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RemoveBindingBody {
    pub stream: String,
    pub topic: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub projection_ref: Option<String>,
}

/// The canonical error body every `/agdx/*` route returns on a non-2xx status.
/// The HTTP binding's rule is "a 2xx carries the bare `Ok` payload, a failure
/// carries this": the status line gives the coarse class (from
/// [`ResultCode::http_status`]) and this body gives the machine-dispatchable
/// [`ResultCode`] plus a human `message`, so a client matches on `code` instead
/// of grepping the message text (which is for humans and may change). `detail`
/// carries optional structured context (e.g. the conflicting version on a CAS
/// miss) as free-form JSON.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ErrorBody {
    pub code: ResultCode,
    pub message: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub detail: Option<serde_json::Value>,
}

impl ErrorBody {
    /// An error body from a classified code and a human message.
    pub fn new(code: ResultCode, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
            detail: None,
        }
    }

    /// Attach structured context.
    #[must_use]
    pub fn with_detail(mut self, detail: serde_json::Value) -> Self {
        self.detail = Some(detail);
        self
    }

    /// The HTTP status this body's `code` maps to, so a server sets the status
    /// line and the body from one value.
    pub fn http_status(&self) -> u16 {
        self.code.http_status()
    }
}

impl From<&QueryError> for ErrorBody {
    fn from(error: &QueryError) -> Self {
        Self::new(ResultCode::from(error), error.to_string())
    }
}

impl From<&KvError> for ErrorBody {
    fn from(error: &KvError) -> Self {
        Self::new(ResultCode::from(error), error.to_string())
    }
}

impl From<&ForkError> for ErrorBody {
    fn from(error: &ForkError) -> Self {
        Self::new(ResultCode::from(error), error.to_string())
    }
}

impl From<&crate::agent_workflow::AgentError> for ErrorBody {
    fn from(error: &crate::agent_workflow::AgentError) -> Self {
        Self::new(ResultCode::from(error), error.to_string())
    }
}

/// `?topic=` on `GET /agdx/projections`: filter to bindings off this source topic.
pub const PARAM_TOPIC: &str = "topic";
/// `?name_contains=` on a projection or schema list: substring over the name.
pub const PARAM_NAME_CONTAINS: &str = "name_contains";
/// `?id_prefix=` on `GET /agdx/projections`: keep ids starting with the prefix.
pub const PARAM_ID_PREFIX: &str = "id_prefix";
/// `?search=` on `GET /agdx/projections`: one substring matched against the
/// projection name OR id. A console with a single filter box maps to it. A
/// server matches it as `name_contains(name) OR id contains search`. Composes
/// (AND) with the narrower `name_contains` / `id_prefix` when several are set.
pub const PARAM_SEARCH: &str = "search";
/// `?prefix=` on a KV scan: base64url key prefix.
pub const PARAM_PREFIX: &str = "prefix";
/// `?start=` on a KV scan: base64url inclusive lower bound.
pub const PARAM_START: &str = "start";
/// `?end=` on a KV scan: base64url exclusive upper bound.
pub const PARAM_END: &str = "end";
/// `?key_contains=`: base64url substring the key must contain.
pub const PARAM_KEY_CONTAINS: &str = "key_contains";
/// `?limit=`: page size.
pub const PARAM_LIMIT: &str = "limit";
/// `?cursor=`: opaque continuation token from the prior page.
pub const PARAM_CURSOR: &str = "cursor";
/// `?expires_at_micros=` on a KV `PUT`: absolute expiry, epoch microseconds.
pub const PARAM_EXPIRES_AT_MICROS: &str = "expires_at_micros";
/// `?expect_version=` on a KV compare-and-swap: apply only if the key holds
/// this exact version.
pub const PARAM_EXPECT_VERSION: &str = "expect_version";
/// `?expect_absent=` on a KV compare-and-swap: apply only if the key is absent
/// (create-if-absent).
pub const PARAM_EXPECT_ABSENT: &str = "expect_absent";

/// Response header on `GET /agdx/kv/{namespace}/{key}` carrying the entry's
/// absolute expiry (epoch microseconds) as a decimal string. The value itself
/// rides the raw response body, so this header carries the one piece of
/// out-of-band metadata a single-key read needs. Owned here so the name is a
/// wire constant rather than an unscoped string. Absent means no expiry.
pub const KV_EXPIRES_AT_MICROS_HEADER: &str = "agdx-expires-at-micros";

/// `GET /agdx/projections` filters. Every field is optional, and an absent field is
/// omitted from the query string (no empty `topic=`). Field names are the
/// `PARAM_*` consts verbatim, so the client serializer and the server parser
/// share one spelling.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProjectionListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topic: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name_contains: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub id_prefix: Option<String>,
    /// One substring matched against the projection name OR id, for a console
    /// with a single filter box. Composes (AND) with `name_contains`/`id_prefix`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub search: Option<String>,
}

/// `GET /agdx/schemas` filters. `name_contains` is the substring filter on a
/// schema's optional name, the same spelling as the projection list, so the two
/// list surfaces share one filter vocabulary.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SchemaListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name_contains: Option<String>,
}

/// `GET /agdx/kv/{namespace}` scan filters. The byte-valued bounds
/// (`prefix`/`start`/`end`/`key_contains`) are base64url, like every binary
/// value on this surface.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvScanQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prefix: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub start: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub end: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key_contains: Option<String>,
    /// The conversation lens: keep only rows the given conversation wrote (the
    /// text form of its `gen_ai.conversation.id`). The memory read view stamps
    /// this on each record, so a scan of a memory namespace narrows to one
    /// conversation. Plain text, not base64url, since a conversation id is text.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub conversation: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<usize>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cursor: Option<String>,
}

/// `PUT /agdx/kv/{namespace}/{key}` query: an optional absolute expiry.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvPutQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expires_at_micros: Option<u64>,
}

/// `GET /agdx/graph/{name}/neighbors/{node}` query: the traversal direction
/// (`out`, `in`, or `both`, omitted for the default `out`), an optional edge-type
/// filter, the hop depth (omitted for the default one hop), and a result limit
/// (omitted for the backend ceiling). One struct shared by the typed client and
/// the server route, so the two cannot drift.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct GraphNeighborsQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub dir: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub edge_type: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub depth: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<usize>,
    /// Valid-time "as of" read (epoch micros): only edges valid at this instant.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub as_of: Option<u64>,
    /// Restrict to elements a given conversation asserted (the text form of its
    /// `gen_ai.conversation.id`). Omitted reads the whole graph. The conversation
    /// lens over the neighbors route.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub conversation: Option<String>,
}

/// `GET /agdx/runs` filters: the binary `AgentList` rendered as query
/// parameters. `state` is the snake-case
/// [`AgentRunState`](crate::agent_workflow::AgentRunState) word, `cursor` is
/// the base64url form of the opaque page cursor, like every binary value on
/// this surface. One struct shared by the typed client and the server route,
/// so the two cannot drift.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunsQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub agent_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub state: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cursor: Option<String>,
}

/// `PUT /agdx/kv/{namespace}/{key}/cas` query: the compare-and-swap precondition
/// plus an optional expiry. Exactly one of `expect_version` (match the held
/// version) or `expect_absent` (create-if-absent) is set, mirroring the binary
/// `CasExpect`. The value rides the raw request body.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct KvCasQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expect_version: Option<u64>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expect_absent: Option<bool>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub expires_at_micros: Option<u64>,
}

/// `PUT /agdx/kv/{namespace}/{key}/cas` reply on success: the new version the
/// committed write took.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CasCommittedView {
    pub version: u64,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn given_path_builders_when_rendered_then_should_match_the_router() {
        assert_eq!(CAPABILITIES_PATH, "/agdx/capabilities");
        assert_eq!(projection_path("order.v1"), "/agdx/projections/order.v1");
        assert_eq!(schema_path(7), "/agdx/schemas/7");
        assert_eq!(schema_decode_path(7), "/agdx/schemas/7/decode");
        assert_eq!(kv_namespace_path("sessions"), "/agdx/kv/sessions");
        assert_eq!(
            kv_entry_path("sessions", "dXNlcjo0Mg"),
            "/agdx/kv/sessions/dXNlcjo0Mg"
        );
        assert_eq!(
            kv_cas_path("sessions", "dXNlcjo0Mg"),
            "/agdx/kv/sessions/dXNlcjo0Mg/cas"
        );
        assert_eq!(fork_path("f1"), "/agdx/forks/f1");
        assert_eq!(fork_promote_path("f1"), "/agdx/forks/f1/promote");
        assert_eq!(fork_rows_path("f1"), "/agdx/forks/f1/rows");
    }

    #[test]
    fn given_capabilities_when_constructed_then_extended_features_default_off() {
        let caps = Capabilities::new(true, OpVersions::new(1, 1, 1, 1));
        assert!(
            caps.query.available && caps.kv.available && caps.fork,
            "core surfaces track enabled"
        );
        assert!(
            !caps.kv.cas
                && !caps.kv.cas_fenced
                && !caps.kv.fenced_leases
                && caps.query.consistency == Consistency::Eventual,
            "sub-features must be opt-in, never on by default"
        );
        let opted = caps
            .with_kv_cas(true)
            .with_query_consistency(Consistency::ReadYourWrites);
        assert!(opted.kv.cas && opted.query.consistency == Consistency::ReadYourWrites);
    }

    #[test]
    fn given_the_fenced_lease_bit_when_building_http_capabilities_then_should_advertise_it() {
        let versions =
            OpVersions::new(1, 1, 1, 1).with_features(crate::hello::feature::KV_FENCED_LEASES);
        let caps = Capabilities::from_versions(true, versions);
        assert!(caps.kv.fenced_leases);
        assert!(
            caps.kv.cas_fenced,
            "the full fenced-lease contract subsumes fenced CAS"
        );
    }

    #[test]
    fn given_destination_versions_when_building_capabilities_then_the_core_surface_is_available() {
        use crate::codes::CHECKPOINT_OP_VERSION;
        use crate::hello::feature;

        let versions = OpVersions::new(2, 2, 1, 1)
            .with_checkpoint(CHECKPOINT_OP_VERSION)
            .with_features(feature::DESTINATIONS);
        let caps = Capabilities::from_versions(true, versions);

        assert!(caps.destinations.available);
        assert!(caps.destinations.lifecycle);
        assert!(caps.destinations.checkpoint_status);
        assert!(caps.destinations.query_routes);
        assert!(!caps.destinations.table_schema);
    }

    #[test]
    fn given_capabilities_backends_when_json_round_tripped_then_should_preserve_and_omit_empty() {
        use crate::destination::BackendResourceId;
        use crate::hello::{BackendDescriptor, BackendImplementation, BackendMode};
        let caps = Capabilities::new(true, OpVersions::new(1, 1, 1, 1)).with_backends(vec![
            BackendDescriptor::new(
                BackendResourceId::from_u128(1),
                BackendMode::Operational,
                "Embedded",
                BackendImplementation {
                    kind: "embedded".to_owned(),
                    version: "1.0.0".to_owned(),
                },
                1,
                1,
            ),
            BackendDescriptor::new(
                BackendResourceId::from_u128(2),
                BackendMode::Lakehouse,
                "Warehouse",
                BackendImplementation {
                    kind: "columnar".to_owned(),
                    version: "1.0.0".to_owned(),
                },
                1,
                1,
            ),
        ]);
        let json = serde_json::to_string(&caps).expect("serializes");
        let back: Capabilities = serde_json::from_str(&json).expect("deserializes");
        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");

        // No advertised backends is omitted on the wire, so a pre-backends
        // capabilities reply stays byte-identical.
        let plain = Capabilities::new(true, 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_capabilities_topology_when_json_round_tripped_then_should_preserve_and_omit_absent() {
        let custom = WireTopology {
            ops_stream: "custom-ops".to_owned(),
            ..WireTopology::default()
        };
        let caps =
            Capabilities::new(true, OpVersions::new(1, 1, 1, 1)).with_topology(custom.clone());
        let json = serde_json::to_string(&caps).expect("serializes");
        let back: Capabilities = serde_json::from_str(&json).expect("deserializes");
        assert_eq!(back.topology, Some(custom));

        // No advertised topology (the default) is omitted on the wire, so a
        // pre-topology capabilities reply stays byte-identical.
        let plain = Capabilities::new(true, OpVersions::new(1, 1, 1, 1));
        let json = serde_json::to_string(&plain).expect("json");
        assert!(
            !json.contains("topology"),
            "absent topology omitted: {json}"
        );
    }

    #[test]
    fn given_a_typed_error_when_made_into_a_body_then_should_carry_code_and_message() {
        let body = ErrorBody::from(&QueryError::IndexNotFound("orders".to_owned()));
        assert_eq!(body.code, ResultCode::NotFound);
        assert_eq!(body.http_status(), 404);
        assert!(body.message.contains("orders"));
        // Round-trips as JSON, the form the HTTP surface serves it in.
        let json = serde_json::to_string(&body).expect("serializes");
        let back: ErrorBody = serde_json::from_str(&json).expect("deserializes");
        assert_eq!(back, body);
    }

    #[test]
    #[cfg(feature = "http-client")]
    fn given_scan_filters_when_url_encoded_then_should_omit_absent_fields() {
        let query = KvScanQuery {
            prefix: Some("dXNlcjo".to_owned()),
            limit: Some(50),
            ..Default::default()
        };
        let encoded = serde_urlencoded::to_string(&query).expect("encodes");
        assert_eq!(encoded, "prefix=dXNlcjo&limit=50");
        // Field names are the PARAM_* consts verbatim.
        assert!(encoded.contains(&format!("{PARAM_PREFIX}=")));
        assert!(encoded.contains(&format!("{PARAM_LIMIT}=")));
    }

    #[test]
    #[cfg(feature = "http-client")]
    fn given_list_filters_when_url_encoded_then_field_names_match_the_param_consts() {
        let projections = ProjectionListQuery {
            name_contains: Some("order".to_owned()),
            id_prefix: Some("order.".to_owned()),
            ..Default::default()
        };
        let encoded = serde_urlencoded::to_string(&projections).expect("encodes");
        assert_eq!(encoded, "name_contains=order&id_prefix=order.");
        assert!(encoded.contains(&format!("{PARAM_NAME_CONTAINS}=")));
        assert!(encoded.contains(&format!("{PARAM_ID_PREFIX}=")));

        let schemas = SchemaListQuery {
            name_contains: Some("Order".to_owned()),
        };
        assert_eq!(
            serde_urlencoded::to_string(&schemas).expect("encodes"),
            "name_contains=Order"
        );
    }
}