laser-wire 0.7.0

use crate::checkpoint::{
    CheckpointMutationResult, CheckpointReadConsistency, CheckpointRequestEnvelope,
    DestinationCheckpointStatus, PreparedAttemptSummary, RetentionGap,
};
use crate::destination::{
    DestinationId, DestinationOperationId, MaterializationDestination, QueryRoute,
};
use crate::fork::{ForkError, ForkKind};
use crate::graph::SourceRef;
use crate::hello::{BackendDescriptor, FilterAnnounce, 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";
/// Prefix of the graph traversal and neighbor routes, `/agdx/graph/{name}/...`.
pub const GRAPH_PATH: &str = "/agdx/graph";
/// `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/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";
/// `GET /agdx/filters`: list saved consumer filters.
pub const FILTERS_PATH: &str = "/agdx/filters";
pub const FILTER_METRICS_PATH: &str = "/agdx/filters/metrics";
/// `POST /agdx/filters/mutations`: one catalog mutation with its outcome.
pub const FILTER_MUTATIONS_PATH: &str = "/agdx/filters/mutations";
/// `POST /agdx/filters/validate`: validate and compile a filter.
pub const FILTER_VALIDATE_PATH: &str = "/agdx/filters/validate";
/// `POST /agdx/filters/test`: evaluate a filter against a supplied sample.
pub const FILTER_TEST_PATH: &str = "/agdx/filters/test";
/// `POST /agdx/filters/preview`: preview a filter over stored records.
pub const FILTER_PREVIEW_PATH: &str = "/agdx/filters/preview";
/// `GET /agdx/filter-bindings`: list consumer-group filter bindings.
pub const FILTER_BINDINGS_PATH: &str = "/agdx/filter-bindings";
/// `GET /agdx/group-policies/{stream}/{topic}/{group}`: confirmed group policy.
pub const GROUP_POLICIES_PATH: &str = "/agdx/group-policies";
/// Base path for reads of sessions in one named stream.
pub const SESSIONS_PATH: &str = "/agdx/sessions";

pub fn sessions_path(stream: &str) -> String {
    format!("{SESSIONS_PATH}/{}", path_segment(stream))
}

pub fn session_changes_path(stream: &str) -> String {
    format!("{}/changes", sessions_path(stream))
}

pub fn session_path(stream: &str, id: crate::agent::ConversationId) -> String {
    format!(
        "{}/{}",
        sessions_path(stream),
        path_segment(&id.to_string())
    )
}

pub fn session_events_path(stream: &str, id: crate::agent::ConversationId) -> String {
    format!("{}/events", session_path(stream, id))
}

pub fn session_state_path(stream: &str, id: crate::agent::ConversationId) -> String {
    format!("{}/state", session_path(stream, id))
}

pub fn session_links_path(stream: &str, id: crate::agent::ConversationId) -> String {
    format!("{}/links", session_path(stream, id))
}

pub fn session_sources_path(stream: &str, id: crate::agent::ConversationId) -> String {
    format!("{}/sources", session_path(stream, id))
}

/// `GET /agdx/filters/{id}`.
pub fn filter_path(id: u32) -> String {
    format!("{FILTERS_PATH}/{id}")
}

/// `GET /agdx/filters/{id}/revisions`.
pub fn filter_revisions_path(id: u32) -> String {
    format!("{FILTERS_PATH}/{id}/revisions")
}

/// `GET /agdx/filters/operations/{operation_id}`.
pub fn filter_operation_path(operation_id: u128) -> String {
    format!("{FILTERS_PATH}/operations/{operation_id}")
}

/// `GET /agdx/filter-bindings/{stream}/{topic}/{group}`. Each name is
/// percent-encoded as one path segment, because Iggy names can hold `/`, `?`,
/// or `#`.
pub fn filter_binding_path(stream: &str, topic: &str, group: &str) -> String {
    format!(
        "{FILTER_BINDINGS_PATH}/{}/{}/{}",
        path_segment(stream),
        path_segment(topic),
        path_segment(group)
    )
}

pub fn group_policy_path(stream: &str, topic: &str, group: &str) -> String {
    format!(
        "{GROUP_POLICIES_PATH}/{}/{}/{}",
        path_segment(stream),
        path_segment(topic),
        path_segment(group)
    )
}

/// Percent-encode `text` as one URL path segment: every byte outside the
/// RFC 3986 unreserved set becomes `%XX`.
pub fn path_segment(text: &str) -> String {
    let mut out = String::with_capacity(text.len());
    for byte in text.bytes() {
        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
            out.push(char::from(byte));
        } else {
            out.push_str(&format!("%{byte:02X}"));
        }
    }
    out
}

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, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CheckpointReadQuery {
    pub consistency: CheckpointReadConsistency,
}

impl CheckpointReadQuery {
    pub const fn new(consistency: CheckpointReadConsistency) -> Self {
        Self { consistency }
    }
}

#[derive(Clone, Debug, 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>,
    pub consistency: CheckpointReadConsistency,
}

#[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>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub result: Option<CheckpointMutationResult>,
}

#[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, 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>,
    pub consistency: CheckpointReadConsistency,
}

#[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,
    pub global_state_revision: u64,
    pub consistency: CheckpointReadConsistency,
}

#[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 global_state_revision: u64,
    pub checkpoint_revision: u64,
    pub consistency: CheckpointReadConsistency,
}

/// One member of the closed destination HTTP operation set.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
#[non_exhaustive]
pub enum DestinationHttpOperation {
    ListDestinations,
    GetDestination,
    CreateDestination,
    EnableDestination,
    DisableDestination,
    GetOperation,
    GetStatus,
    GetCheckpoint,
    GetRetentionGap,
    GetPreparedAttempt,
    ListQueryRoutes,
    GetTable,
    GetTableSchema,
    GetCurrentSnapshot,
    ListSnapshots,
    GetSnapshot,
    ListFiles,
    GetMetrics,
}

/// One closed destination HTTP operation forwarded from Iggy to Plane.
///
/// The operation enum prevents arbitrary paths or methods from crossing the
/// sidecar boundary. Path parameters stay textual, while query and body carry
/// the JSON encodings of the operation's already-declared HTTP types.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DestinationHttpRequest {
    pub v: u32,
    pub operation: DestinationHttpOperation,
    pub parameters: Vec<String>,
    pub query: Vec<u8>,
    pub body: Vec<u8>,
}

/// Plane's HTTP-shaped response for one [`DestinationHttpRequest`].
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct DestinationHttpReply {
    pub status: u16,
    pub body: Vec<u8>,
}

/// `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. The id
/// is percent-encoded as one path segment, like every name below, because a
/// stream-scoped name holds `/`.
pub fn graph_path(id: &str) -> String {
    format!("{GRAPHS_PATH}/{}", path_segment(id))
}

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

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

/// `GET`/`DELETE /agdx/projections/{id}`.
pub fn projection_path(id: &str) -> String {
    format!("{PROJECTIONS_PATH}/{}", path_segment(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}/{}", path_segment(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_namespace_path(namespace), path_segment(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!("{}/cas", kv_entry_path(namespace, key_b64))
}

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

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

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

/// `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 managed session reads are served.
    #[serde(default)]
    pub sessions: 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,
    /// Consumer filters: server-side evaluation and the saved-filter catalog,
    /// reported separately.
    #[serde(default)]
    pub filters: FilterCapsView,
    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,
}

/// Consumer filters on the HTTP capabilities reply. Native evaluation is served
/// by the streaming server itself and holds with the managed backend off. The
/// catalog (saved filters and group bindings) needs the backend.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct FilterCapsView {
    /// Filtered reads, fenced acknowledgments, sample tests, and validation.
    #[serde(default)]
    pub native: bool,
    /// Group-aware reads, including pass-through when no policy is bound.
    #[serde(default)]
    pub group_policy_reads: bool,
    /// Bounded previews over stored records.
    #[serde(default)]
    pub preview: bool,
    /// Filtered reads from the partition primary through an attached session.
    #[serde(default)]
    pub primary_routing: bool,
    /// Saved filters, revisions, and mutation outcomes.
    #[serde(default)]
    pub catalog: bool,
    /// Consumer groups bound to a saved revision.
    #[serde(default)]
    pub managed_groups: bool,
    /// The evaluation rules served (`0` when native evaluation is off).
    #[serde(default)]
    pub evaluator_version: u32,
    /// The payload codecs a filter may select.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub codecs: Vec<crate::filter::FilterCodec>,
}

impl FilterCapsView {
    /// The view a server serving native evaluation and a ready catalog advertises.
    pub fn served(native: bool, catalog: bool) -> Self {
        let evaluation = native.then(FilterAnnounce::served);
        Self {
            native,
            group_policy_reads: false,
            preview: native,
            primary_routing: native,
            catalog,
            managed_groups: catalog,
            evaluator_version: evaluation
                .as_ref()
                .map_or(0, |evaluation| evaluation.evaluator_version),
            codecs: evaluation
                .map(|evaluation| evaluation.codecs)
                .unwrap_or_default(),
        }
    }

    #[must_use]
    pub const fn with_group_policy_reads(mut self, served: bool) -> Self {
        self.group_policy_reads = served;
        self
    }

    /// Returns a copy reporting the evaluation a server announced instead of
    /// this build's own.
    #[must_use]
    pub fn with_evaluation(mut self, evaluation: FilterAnnounce) -> Self {
        if self.native {
            self.evaluator_version = evaluation.evaluator_version;
            self.codecs = evaluation.codecs;
        }
        self
    }
}

/// 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,
            sessions: false,
            watch: false,
            authz: false,
            filters: FilterCapsView::default(),
            versions,
            backends: Vec::new(),
            topology: None,
        }
    }

    /// Advertise the consumer-filter surface.
    #[must_use]
    pub fn with_filters(mut self, filters: FilterCapsView) -> Self {
        self.filters = filters;
        self
    }

    /// 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
    }

    /// 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 managed session reads are served.
    #[must_use]
    pub fn with_sessions(mut self, value: bool) -> Self {
        self.sessions = value;
        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_query_keyword(versions.has_feature(feature::KEYWORD_SEARCH))
            .with_watch(versions.has_feature(feature::WATCH))
            .with_sessions(enabled && versions.has_feature(feature::SESSIONS))
            .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)
            .with_filters(
                FilterCapsView::served(
                    versions.has_feature(feature::CONSUMER_FILTERS),
                    enabled && versions.filter > 0,
                )
                .with_group_policy_reads(versions.has_feature(feature::GROUP_POLICY_READS)),
            )
    }
}

/// 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>,
}

/// `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>,
    /// The stream whose registry allocates the id. Absent for the
    /// deployment-wide registry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

/// `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<&crate::filter::FilterError> for ErrorBody {
    fn from(error: &crate::filter::FilterError) -> Self {
        Self::new(error.code, error.message.clone())
            .with_detail(serde_json::json!({ "reason": error.reason }))
    }
}

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::authz::AuthzError> for ErrorBody {
    fn from(error: &crate::authz::AuthzError) -> Self {
        Self::new(ResultCode::from(error), error.to_string())
    }
}

impl From<&crate::session::SessionError> for ErrorBody {
    fn from(error: &crate::session::SessionError) -> 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";
/// `?stream=` on the schema routes: the stream whose writer-schema registry
/// the request addresses. Absent addresses the deployment-wide registry.
pub const PARAM_STREAM: &str = "stream";
/// `?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";
/// `?status=` on a session list: one session status.
pub const PARAM_STATUS: &str = "status";
/// `?agent=` on a session list: sessions this agent took part in.
pub const PARAM_AGENT: &str = "agent";
/// `?text=` on a session list: a substring of the label or id.
pub const PARAM_TEXT: &str = "text";
/// `?root=` on a session list: the root of the session tree.
pub const PARAM_ROOT: &str = "root";
/// `?label_prefix=` on a session list: the start of the label.
pub const PARAM_LABEL_PREFIX: &str = "label_prefix";
/// `?surface=` on session links: one link surface.
pub const PARAM_SURFACE: &str = "surface";
/// `?after=` on session changes: the last change sequence already seen.
pub const PARAM_AFTER: &str = "after";
/// `?total=` on a session list: also count every match.
pub const PARAM_TOTAL: &str = "total";
/// `?fixed_frontier=` on session events: pin the fold frontier of the first
/// page for a historical walk.
pub const PARAM_FIXED_FRONTIER: &str = "fixed_frontier";
/// `?history_limit=` on session state: the most history rows to return.
pub const PARAM_HISTORY_LIMIT: &str = "history_limit";
/// `?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>,
    /// The stream whose registry is listed. Absent lists the deployment-wide
    /// registry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

/// `?stream=` on `GET`/`DELETE /agdx/schemas/{id}` and
/// `POST /agdx/schemas/{id}/decode`: the stream whose registry holds the id.
/// Absent addresses the deployment-wide registry.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SchemaQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

/// `GET /agdx/filters` query parameters.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct FilterListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name_contains: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub state: Option<crate::filter::FilterState>,
    /// Only filters with a lower id, for stable paging.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub before_id: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub page: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub page_size: Option<u32>,
}

/// `GET /agdx/filters/{id}/revisions` query parameters.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct FilterRevisionListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub page: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub page_size: Option<u32>,
}

/// `GET /agdx/filter-bindings` query parameters.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct FilterBindingListQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub filter_id: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topic: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub page: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub page_size: Option<u32>,
}

/// `POST /agdx/filters/test` body. The sample payload is given either as
/// base64url bytes or as UTF-8 text, never both.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct FilterTestBody {
    pub filter: crate::filter::FilterRef,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub payload_base64: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub payload_text: Option<String>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub headers: Vec<crate::filter::FilterHeader>,
}

/// `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>,
}

/// Filters for `GET /agdx/sessions/{stream}`. The stream is the path
/// segment. A `limit` of zero or none leaves the page size to the server.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionsQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub status: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub root: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub label_prefix: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub agent: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub text: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cursor: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<u32>,
    #[serde(default, rename = "total")]
    pub want_total: bool,
}

/// Query of `GET /agdx/sessions/{stream}/{id}/events`. A `limit` of zero or
/// none leaves the page size to the server.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionEventsQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cursor: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<u32>,
    #[serde(default)]
    pub fixed_frontier: bool,
}

/// Query of `GET /agdx/sessions/{stream}/{id}/state`. A `history_limit` of
/// zero or none leaves the history size to the server.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionStateQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub history_limit: Option<u32>,
}

/// Query of `GET /agdx/sessions/{stream}/{id}/links`.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionLinksQuery {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub surface: Option<String>,
}

/// Query of `GET /agdx/sessions/{stream}/changes`. A `limit` of zero or none
/// leaves the page size to the server.
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionChangesQuery {
    #[serde(default)]
    pub after: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub limit: Option<u32>,
}

/// `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("reading.v1"),
            "/agdx/projections/reading.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_stream_scoped_names_when_rendered_then_should_encode_each_as_one_segment() {
        assert_eq!(
            kv_namespace_path("stream:acme/support"),
            "/agdx/kv/stream%3Aacme%2Fsupport"
        );
        assert_eq!(
            kv_cas_path("stream:acme/support", "dXNlcjo0Mg"),
            "/agdx/kv/stream%3Aacme%2Fsupport/dXNlcjo0Mg/cas"
        );
        assert_eq!(
            graph_neighbors_path("stream:acme/kg", "a/b"),
            "/agdx/graph/stream%3Aacme%2Fkg/neighbors/a%2Fb"
        );
        assert_eq!(
            graph_query_path("stream:acme/kg"),
            "/agdx/graph/stream%3Aacme%2Fkg/query"
        );
        assert_eq!(
            graph_path("stream:acme/kg"),
            "/agdx/graphs/stream%3Aacme%2Fkg"
        );
        assert_eq!(
            projection_path("stream:acme/readings"),
            "/agdx/projections/stream%3Aacme%2Freadings"
        );
    }

    #[test]
    fn given_the_sessions_feature_when_built_from_versions_then_should_advertise_sessions() {
        let mut versions = OpVersions::new(1, 1, 1, 1);
        assert!(!Capabilities::from_versions(true, versions).sessions);
        versions.features |= crate::hello::feature::SESSIONS;
        assert!(Capabilities::from_versions(true, versions).sessions);
        assert!(!Capabilities::from_versions(false, versions).sessions);
    }

    #[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_group_reads_without_evaluation_when_building_http_capabilities_then_should_keep_catalog_available()
     {
        let versions = OpVersions::new(1, 1, 1, 1)
            .with_filter(crate::codes::FILTER_OP_VERSION)
            .with_features(crate::hello::feature::GROUP_POLICY_READS);
        let caps = Capabilities::from_versions(true, versions);
        assert!(caps.filters.group_policy_reads);
        assert!(caps.filters.catalog && caps.filters.managed_groups);
        assert!(!caps.filters.native && !caps.filters.preview);
        assert_eq!(caps.filters.evaluator_version, 0);
        assert!(caps.filters.codecs.is_empty());
        let absent = Capabilities::from_versions(false, OpVersions::new(1, 1, 1, 1));
        assert!(!absent.filters.group_policy_reads && !absent.filters.catalog);
    }

    #[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("readings".to_owned()));
        assert_eq!(body.code, ResultCode::NotFound);
        assert_eq!(body.http_status(), 404);
        assert!(body.message.contains("readings"));
        // 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_names_with_reserved_characters_when_building_a_binding_path_then_should_encode_each_segment()
     {
        assert_eq!(
            filter_binding_path("orbit", "fleet/changes", "desk?x#1"),
            "/agdx/filter-bindings/orbit/fleet%2Fchanges/desk%3Fx%231"
        );
    }

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

        let schemas = SchemaListQuery {
            name_contains: Some("Reading".to_owned()),
            stream: None,
        };
        assert_eq!(
            serde_urlencoded::to_string(&schemas).expect("encodes"),
            "name_contains=Reading"
        );
        let scoped = SchemaListQuery {
            name_contains: None,
            stream: Some("acme".to_owned()),
        };
        assert_eq!(
            serde_urlencoded::to_string(&scoped).expect("encodes"),
            format!("{PARAM_STREAM}=acme")
        );
        assert_eq!(
            serde_urlencoded::to_string(SchemaQuery::default()).expect("encodes"),
            ""
        );
        assert_eq!(
            serde_urlencoded::to_string(SchemaQuery {
                stream: Some("acme".to_owned()),
            })
            .expect("encodes"),
            format!("{PARAM_STREAM}=acme")
        );
    }

    #[test]
    fn given_a_register_schema_body_without_a_stream_when_encoded_then_should_omit_it() {
        let body = RegisterSchemaBody {
            source: crate::control::SchemaSource::JsonSchema {
                schema: "{}".to_owned(),
            },
            name: None,
            version: None,
            stream: None,
        };
        let json = serde_json::to_string(&body).expect("encodes");
        assert!(
            !json.contains("stream"),
            "an unset stream is omitted: {json}"
        );
        let scoped: RegisterSchemaBody = serde_json::from_str(
            r#"{"source":{"kind":"json_schema","schema":"{}"},"stream":"acme"}"#,
        )
        .expect("decodes");
        assert_eq!(scoped.stream.as_deref(), Some("acme"));
    }
}