laser_wire/
http.rs

1use crate::checkpoint::{
2    CheckpointMutationResult, CheckpointReadConsistency, CheckpointRequestEnvelope,
3    DestinationCheckpointStatus, PreparedAttemptSummary, RetentionGap,
4};
5use crate::destination::{
6    DestinationId, DestinationOperationId, MaterializationDestination, QueryRoute,
7};
8use crate::fork::{ForkError, ForkKind};
9use crate::graph::SourceRef;
10use crate::hello::{BackendDescriptor, FilterAnnounce, OpVersions};
11use crate::kv::KvError;
12use crate::query::{Consistency, QueryError, QueryExecutionId, QueryExecutionStatus, QueryResult};
13use crate::result::ResultCode;
14use crate::schema::{Digest32, LogicalSchema, TypedValue, UuidValue};
15use crate::topology::WireTopology;
16use serde::{Deserialize, Serialize};
17use std::collections::BTreeMap;
18
19/// `GET /agdx/capabilities`: the feature-detection probe.
20pub const CAPABILITIES_PATH: &str = "/agdx/capabilities";
21/// `POST /agdx/query` (and `GET` with the query as a parameter).
22pub const QUERY_PATH: &str = "/agdx/query";
23/// `GET` and `POST /agdx/destinations`.
24pub const DESTINATIONS_PATH: &str = "/agdx/destinations";
25/// `GET /agdx/query-routes`.
26pub const QUERY_ROUTES_PATH: &str = "/agdx/query-routes";
27/// `GET /agdx/projections` to list, `POST` to register.
28pub const PROJECTIONS_PATH: &str = "/agdx/projections";
29/// `POST /agdx/bindings` to apply, `DELETE` to remove.
30pub const BINDINGS_PATH: &str = "/agdx/bindings";
31/// `GET /agdx/schemas` to list, `POST` to register.
32pub const SCHEMAS_PATH: &str = "/agdx/schemas";
33/// `GET /agdx/kv` to list the caller's namespaces.
34pub const KV_PATH: &str = "/agdx/kv";
35/// `GET /agdx/forks` to list, `POST` to create.
36pub const FORKS_PATH: &str = "/agdx/forks";
37/// `GET /agdx/graphs` to list graph projections, `POST` to register.
38pub const GRAPHS_PATH: &str = "/agdx/graphs";
39/// Prefix of the graph traversal and neighbor routes, `/agdx/graph/{name}/...`.
40pub const GRAPH_PATH: &str = "/agdx/graph";
41/// `GET /agdx/clients` to list live connections with their advertised metadata,
42/// filtered and paginated by query parameters. The HTTP face of the
43/// `AGDX_GET_CLIENTS_METADATA` discovery read.
44pub const CLIENTS_PATH: &str = "/agdx/clients";
45/// `GET /agdx/authz/whoami`: the caller's effective governance roles and grants.
46pub const AUTHZ_WHOAMI_PATH: &str = "/agdx/authz/whoami";
47/// `GET /agdx/authz/roles`: list governance roles.
48pub const AUTHZ_ROLES_PATH: &str = "/agdx/authz/roles";
49/// `GET /agdx/filters`: list saved consumer filters.
50pub const FILTERS_PATH: &str = "/agdx/filters";
51pub const FILTER_METRICS_PATH: &str = "/agdx/filters/metrics";
52/// `POST /agdx/filters/mutations`: one catalog mutation with its outcome.
53pub const FILTER_MUTATIONS_PATH: &str = "/agdx/filters/mutations";
54/// `POST /agdx/filters/validate`: validate and compile a filter.
55pub const FILTER_VALIDATE_PATH: &str = "/agdx/filters/validate";
56/// `POST /agdx/filters/test`: evaluate a filter against a supplied sample.
57pub const FILTER_TEST_PATH: &str = "/agdx/filters/test";
58/// `POST /agdx/filters/preview`: preview a filter over stored records.
59pub const FILTER_PREVIEW_PATH: &str = "/agdx/filters/preview";
60/// `GET /agdx/filter-bindings`: list consumer-group filter bindings.
61pub const FILTER_BINDINGS_PATH: &str = "/agdx/filter-bindings";
62/// `GET /agdx/group-policies/{stream}/{topic}/{group}`: confirmed group policy.
63pub const GROUP_POLICIES_PATH: &str = "/agdx/group-policies";
64/// Base path for reads of sessions in one named stream.
65pub const SESSIONS_PATH: &str = "/agdx/sessions";
66
67pub fn sessions_path(stream: &str) -> String {
68    format!("{SESSIONS_PATH}/{}", path_segment(stream))
69}
70
71pub fn session_changes_path(stream: &str) -> String {
72    format!("{}/changes", sessions_path(stream))
73}
74
75pub fn session_path(stream: &str, id: crate::agent::ConversationId) -> String {
76    format!(
77        "{}/{}",
78        sessions_path(stream),
79        path_segment(&id.to_string())
80    )
81}
82
83pub fn session_events_path(stream: &str, id: crate::agent::ConversationId) -> String {
84    format!("{}/events", session_path(stream, id))
85}
86
87pub fn session_state_path(stream: &str, id: crate::agent::ConversationId) -> String {
88    format!("{}/state", session_path(stream, id))
89}
90
91pub fn session_links_path(stream: &str, id: crate::agent::ConversationId) -> String {
92    format!("{}/links", session_path(stream, id))
93}
94
95pub fn session_sources_path(stream: &str, id: crate::agent::ConversationId) -> String {
96    format!("{}/sources", session_path(stream, id))
97}
98
99/// `GET /agdx/filters/{id}`.
100pub fn filter_path(id: u32) -> String {
101    format!("{FILTERS_PATH}/{id}")
102}
103
104/// `GET /agdx/filters/{id}/revisions`.
105pub fn filter_revisions_path(id: u32) -> String {
106    format!("{FILTERS_PATH}/{id}/revisions")
107}
108
109/// `GET /agdx/filters/operations/{operation_id}`.
110pub fn filter_operation_path(operation_id: u128) -> String {
111    format!("{FILTERS_PATH}/operations/{operation_id}")
112}
113
114/// `GET /agdx/filter-bindings/{stream}/{topic}/{group}`. Each name is
115/// percent-encoded as one path segment, because Iggy names can hold `/`, `?`,
116/// or `#`.
117pub fn filter_binding_path(stream: &str, topic: &str, group: &str) -> String {
118    format!(
119        "{FILTER_BINDINGS_PATH}/{}/{}/{}",
120        path_segment(stream),
121        path_segment(topic),
122        path_segment(group)
123    )
124}
125
126pub fn group_policy_path(stream: &str, topic: &str, group: &str) -> String {
127    format!(
128        "{GROUP_POLICIES_PATH}/{}/{}/{}",
129        path_segment(stream),
130        path_segment(topic),
131        path_segment(group)
132    )
133}
134
135/// Percent-encode `text` as one URL path segment: every byte outside the
136/// RFC 3986 unreserved set becomes `%XX`.
137pub fn path_segment(text: &str) -> String {
138    let mut out = String::with_capacity(text.len());
139    for byte in text.bytes() {
140        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~') {
141            out.push(char::from(byte));
142        } else {
143            out.push_str(&format!("%{byte:02X}"));
144        }
145    }
146    out
147}
148
149pub fn destination_path(id: DestinationId) -> String {
150    format!("{DESTINATIONS_PATH}/{id}")
151}
152
153pub fn destination_enable_path(id: DestinationId) -> String {
154    format!("{}/{id}/enable", DESTINATIONS_PATH)
155}
156
157pub fn destination_disable_path(id: DestinationId) -> String {
158    format!("{}/{id}/disable", DESTINATIONS_PATH)
159}
160
161pub fn destination_status_path(id: DestinationId) -> String {
162    format!("{}/{id}/status", DESTINATIONS_PATH)
163}
164
165pub fn destination_checkpoint_path(id: DestinationId) -> String {
166    format!("{}/{id}/checkpoint", DESTINATIONS_PATH)
167}
168
169pub fn destination_retention_gap_path(id: DestinationId) -> String {
170    format!("{}/{id}/retention-gap", DESTINATIONS_PATH)
171}
172
173pub fn destination_prepared_attempt_path(id: DestinationId) -> String {
174    format!("{}/{id}/prepared-attempt", DESTINATIONS_PATH)
175}
176
177pub fn destination_table_path(id: DestinationId) -> String {
178    format!("{}/{id}/table", DESTINATIONS_PATH)
179}
180
181pub fn destination_table_schema_path(id: DestinationId) -> String {
182    format!("{}/{id}/table/schema", DESTINATIONS_PATH)
183}
184
185pub fn destination_snapshots_path(id: DestinationId) -> String {
186    format!("{}/{id}/table/snapshots", DESTINATIONS_PATH)
187}
188
189pub fn destination_current_snapshot_path(id: DestinationId) -> String {
190    format!("{}/{id}/table/current-snapshot", DESTINATIONS_PATH)
191}
192
193pub fn destination_snapshot_path(id: DestinationId, snapshot_id: i64) -> String {
194    format!("{}/{id}/table/snapshots/{snapshot_id}", DESTINATIONS_PATH)
195}
196
197pub fn destination_files_path(id: DestinationId) -> String {
198    format!("{}/{id}/table/files", DESTINATIONS_PATH)
199}
200
201pub fn destination_metrics_path(id: DestinationId) -> String {
202    format!("{}/{id}/table/metrics", DESTINATIONS_PATH)
203}
204
205pub fn query_execution_path(id: QueryExecutionId) -> String {
206    format!("{QUERY_PATH}/{id}")
207}
208
209pub fn query_page_path(id: QueryExecutionId) -> String {
210    format!("{QUERY_PATH}/{id}/pages")
211}
212
213pub fn query_cancel_path(id: QueryExecutionId) -> String {
214    format!("{QUERY_PATH}/{id}/cancel")
215}
216
217pub fn destination_operation_path(id: DestinationOperationId) -> String {
218    format!("{DESTINATIONS_PATH}/operations/{id}")
219}
220
221#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
222pub struct CheckpointReadQuery {
223    pub consistency: CheckpointReadConsistency,
224}
225
226impl CheckpointReadQuery {
227    pub const fn new(consistency: CheckpointReadConsistency) -> Self {
228        Self { consistency }
229    }
230}
231
232#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
233pub struct DestinationListQuery {
234    #[serde(default, skip_serializing_if = "Option::is_none")]
235    pub source_stream: Option<String>,
236    #[serde(default, skip_serializing_if = "Option::is_none")]
237    pub source_topic: Option<String>,
238    #[serde(default, skip_serializing_if = "Option::is_none")]
239    pub name_contains: Option<String>,
240    #[serde(default, skip_serializing_if = "Option::is_none")]
241    pub after: Option<String>,
242    #[serde(default, skip_serializing_if = "Option::is_none")]
243    pub limit: Option<usize>,
244    pub consistency: CheckpointReadConsistency,
245}
246
247#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
248pub struct DestinationPageView {
249    pub destinations: Vec<DestinationView>,
250    #[serde(default, skip_serializing_if = "Option::is_none")]
251    pub next_cursor: Option<String>,
252    pub global_state_revision: u64,
253    pub consistency: crate::checkpoint::CheckpointReadConsistency,
254}
255
256#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
257pub struct DestinationView {
258    pub destination: MaterializationDestination,
259    pub status: DestinationCheckpointStatus,
260}
261
262#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
263pub struct DestinationMutationBody {
264    pub checkpoint: CheckpointRequestEnvelope,
265}
266
267#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
268pub struct AcceptedOperationView {
269    pub operation_id: DestinationOperationId,
270    pub request_id: crate::checkpoint::CheckpointRequestId,
271    pub state: OperationState,
272    pub submitted_at_micros: u64,
273    #[serde(default, skip_serializing_if = "Option::is_none")]
274    pub completed_at_micros: Option<u64>,
275    #[serde(default, skip_serializing_if = "Option::is_none")]
276    pub error: Option<OperationErrorView>,
277    #[serde(default, skip_serializing_if = "Option::is_none")]
278    pub result: Option<CheckpointMutationResult>,
279}
280
281#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
282pub struct OperationErrorView {
283    pub code: ResultCode,
284    pub message: String,
285}
286
287#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
288#[serde(rename_all = "snake_case")]
289#[non_exhaustive]
290pub enum OperationState {
291    Accepted,
292    Running,
293    Succeeded,
294    Failed,
295    Cancelled,
296}
297
298#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
299pub struct QueryRouteListQuery {
300    #[serde(default, skip_serializing_if = "Option::is_none")]
301    pub name_contains: Option<String>,
302    #[serde(default, skip_serializing_if = "Option::is_none")]
303    pub after: Option<String>,
304    #[serde(default, skip_serializing_if = "Option::is_none")]
305    pub limit: Option<usize>,
306    pub consistency: CheckpointReadConsistency,
307}
308
309#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
310pub struct QueryRoutePageView {
311    pub routes: Vec<QueryRoute>,
312    #[serde(default, skip_serializing_if = "Option::is_none")]
313    pub next_cursor: Option<String>,
314    pub definition_revision: u64,
315    pub global_state_revision: u64,
316    pub consistency: CheckpointReadConsistency,
317}
318
319#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
320pub struct TableView {
321    pub table_uuid: UuidValue,
322    pub destination_id: DestinationId,
323    pub destination_generation: u64,
324    pub namespace: Vec<String>,
325    pub table: String,
326    pub current_snapshot_id: i64,
327    pub current_schema_id: i32,
328    pub current_partition_spec_id: i32,
329    pub metadata_identity: String,
330    pub properties: BTreeMap<String, String>,
331}
332
333#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
334pub struct TableSchemaView {
335    pub table_uuid: UuidValue,
336    pub iceberg_schema_id: i32,
337    pub logical_schema: LogicalSchema,
338}
339
340#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
341pub struct SnapshotListQuery {
342    #[serde(default, skip_serializing_if = "Option::is_none")]
343    pub before_snapshot_id: Option<i64>,
344    #[serde(default, skip_serializing_if = "Option::is_none")]
345    pub limit: Option<usize>,
346}
347
348#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
349pub struct SnapshotPageView {
350    pub snapshots: Vec<TableSnapshotView>,
351    #[serde(default, skip_serializing_if = "Option::is_none")]
352    pub next_before_snapshot_id: Option<i64>,
353}
354
355#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
356pub struct TableSnapshotView {
357    pub snapshot_id: i64,
358    #[serde(default, skip_serializing_if = "Option::is_none")]
359    pub parent_snapshot_id: Option<i64>,
360    pub sequence_number: i64,
361    pub committed_at_micros: u64,
362    pub schema_id: i32,
363    pub partition_spec_id: i32,
364    pub materialization_boundary_digest: Digest32,
365    pub checkpoint_revision: u64,
366    pub summary: BTreeMap<String, String>,
367}
368
369#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
370pub struct TableFileListQuery {
371    #[serde(default, skip_serializing_if = "Option::is_none")]
372    pub snapshot_id: Option<i64>,
373    #[serde(default, skip_serializing_if = "Option::is_none")]
374    pub after: Option<String>,
375    #[serde(default, skip_serializing_if = "Option::is_none")]
376    pub limit: Option<usize>,
377}
378
379#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
380pub struct TableFilePageView {
381    pub files: Vec<TableFileView>,
382    #[serde(default, skip_serializing_if = "Option::is_none")]
383    pub next_cursor: Option<String>,
384}
385
386#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
387pub struct TableFileView {
388    pub object_identity: String,
389    pub file_size_bytes: u64,
390    pub row_count: u64,
391    pub partition: BTreeMap<String, TypedValue>,
392    pub lower_bounds: BTreeMap<u32, TypedValue>,
393    pub upper_bounds: BTreeMap<u32, TypedValue>,
394    pub null_value_counts: BTreeMap<u32, u64>,
395}
396
397#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
398pub struct TableMetricsView {
399    pub snapshot_id: i64,
400    pub data_file_count: u64,
401    pub delete_file_count: u64,
402    pub total_rows: u64,
403    pub total_bytes: u64,
404    pub partition_count: u64,
405}
406
407#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
408pub struct QueryPageBody {
409    pub cursor: String,
410    pub deadline_micros: u64,
411}
412
413#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
414pub struct QueryExecutionView {
415    pub status: QueryExecutionStatus,
416    #[serde(default, skip_serializing_if = "Option::is_none")]
417    pub result: Option<QueryResult>,
418}
419
420#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
421pub struct DestinationIssueView {
422    #[serde(default, skip_serializing_if = "Option::is_none")]
423    pub retention_gap: Option<RetentionGap>,
424    #[serde(default, skip_serializing_if = "Option::is_none")]
425    pub prepared_attempt: Option<PreparedAttemptSummary>,
426    #[serde(default, skip_serializing_if = "Option::is_none")]
427    pub block: Option<crate::checkpoint::DestinationBlock>,
428    pub global_state_revision: u64,
429    pub checkpoint_revision: u64,
430    pub consistency: CheckpointReadConsistency,
431}
432
433/// One member of the closed destination HTTP operation set.
434#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
435#[serde(rename_all = "snake_case")]
436#[non_exhaustive]
437pub enum DestinationHttpOperation {
438    ListDestinations,
439    GetDestination,
440    CreateDestination,
441    EnableDestination,
442    DisableDestination,
443    GetOperation,
444    GetStatus,
445    GetCheckpoint,
446    GetRetentionGap,
447    GetPreparedAttempt,
448    ListQueryRoutes,
449    GetTable,
450    GetTableSchema,
451    GetCurrentSnapshot,
452    ListSnapshots,
453    GetSnapshot,
454    ListFiles,
455    GetMetrics,
456}
457
458/// One closed destination HTTP operation forwarded from Iggy to Plane.
459///
460/// The operation enum prevents arbitrary paths or methods from crossing the
461/// sidecar boundary. Path parameters stay textual, while query and body carry
462/// the JSON encodings of the operation's already-declared HTTP types.
463#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
464pub struct DestinationHttpRequest {
465    pub v: u32,
466    pub operation: DestinationHttpOperation,
467    pub parameters: Vec<String>,
468    pub query: Vec<u8>,
469    pub body: Vec<u8>,
470}
471
472/// Plane's HTTP-shaped response for one [`DestinationHttpRequest`].
473#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
474pub struct DestinationHttpReply {
475    pub status: u16,
476    pub body: Vec<u8>,
477}
478
479/// `GET`/`PUT`/`DELETE /agdx/authz/roles/{name}`.
480pub fn authz_role_path(name: &str) -> String {
481    format!("{AUTHZ_ROLES_PATH}/{name}")
482}
483
484/// `GET`/`PUT /agdx/authz/users/{id}/roles`: read or replace one user's role set.
485pub fn authz_user_roles_path(user_id: u32) -> String {
486    format!("/agdx/authz/users/{user_id}/roles")
487}
488
489/// `DELETE`/`GET /agdx/graphs/{id}`: drop or read a graph projection. The id
490/// is percent-encoded as one path segment, like every name below, because a
491/// stream-scoped name holds `/`.
492pub fn graph_path(id: &str) -> String {
493    format!("{GRAPHS_PATH}/{}", path_segment(id))
494}
495
496/// `POST /agdx/graph/{name}/query`: run a traversal (a `GraphQuery` body).
497pub fn graph_query_path(name: &str) -> String {
498    format!("{GRAPH_PATH}/{}/query", path_segment(name))
499}
500
501/// `GET /agdx/graph/{name}/neighbors/{node}`: one-hop neighbor read.
502pub fn graph_neighbors_path(name: &str, node: &str) -> String {
503    format!(
504        "{GRAPH_PATH}/{}/neighbors/{}",
505        path_segment(name),
506        path_segment(node)
507    )
508}
509
510/// `GET`/`DELETE /agdx/projections/{id}`.
511pub fn projection_path(id: &str) -> String {
512    format!("{PROJECTIONS_PATH}/{}", path_segment(id))
513}
514
515/// `GET`/`DELETE /agdx/schemas/{id}`.
516pub fn schema_path(id: u32) -> String {
517    format!("{SCHEMAS_PATH}/{id}")
518}
519
520/// `POST /agdx/schemas/{id}/decode`.
521pub fn schema_decode_path(id: u32) -> String {
522    format!("{SCHEMAS_PATH}/{id}/decode")
523}
524
525/// `GET /agdx/kv/{namespace}` to scan, `DELETE` to bulk-delete.
526pub fn kv_namespace_path(namespace: &str) -> String {
527    format!("{KV_PATH}/{}", path_segment(namespace))
528}
529
530/// `GET`/`PUT`/`DELETE /agdx/kv/{namespace}/{key}`. `key` is the URL-safe
531/// unpadded base64 form of the key bytes, the encoding this surface uses for
532/// every binary body.
533pub fn kv_entry_path(namespace: &str, key_b64: &str) -> String {
534    format!("{}/{}", kv_namespace_path(namespace), path_segment(key_b64))
535}
536
537/// `PUT /agdx/kv/{namespace}/{key}/cas`: a conditional write (compare-and-swap).
538/// The precondition rides the query string (`expect_version` or `expect_absent`)
539/// and the value rides the raw body, like the plain `PUT`. A success replies
540/// `CasCommittedView` with the new version, a precondition miss replies `409`
541/// with an `ErrorBody` of code `conflict` whose `detail` carries the current
542/// version.
543pub fn kv_cas_path(namespace: &str, key_b64: &str) -> String {
544    format!("{}/cas", kv_entry_path(namespace, key_b64))
545}
546
547/// `DELETE /agdx/forks/{id}`.
548pub fn fork_path(id: &str) -> String {
549    format!("{FORKS_PATH}/{}", path_segment(id))
550}
551
552/// `POST /agdx/forks/{id}/promote`.
553pub fn fork_promote_path(id: &str) -> String {
554    format!("{}/promote", fork_path(id))
555}
556
557/// `PUT /agdx/forks/{id}/rows`.
558pub fn fork_rows_path(id: &str) -> String {
559    format!("{}/rows", fork_path(id))
560}
561
562/// `GET /agdx/capabilities` reply: what the `/agdx/*` surface offers on this
563/// server, so a browser client can feature-detect before showing the
564/// projections / query / KV / fork views. Richer than the binary `AGDX_HELLO`
565/// probe (per-surface flags plus the wire op versions the JSON bodies must
566/// match), and it answers truthfully even when the managed backend is disabled (200
567/// with `managed: false`).
568#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
569#[non_exhaustive]
570pub struct Capabilities {
571    /// Connected to a managed plane at all (the root: with no plane every managed
572    /// surface below is off, and the reply still answers `200` with `managed:
573    /// false`).
574    pub managed: bool,
575    /// The managed query surface, its registry browse views, and the strongest
576    /// read-consistency it serves.
577    pub query: QueryCapsView,
578    /// Destination declarations, checkpoint status, and table inspection.
579    pub destinations: DestinationCapsView,
580    /// The managed key-value surface and its conditional-write support.
581    pub kv: KvCapsView,
582    /// Whether the knowledge-graph ops (traversal, neighbors) are served. The
583    /// agentic-memory API composes the query and graph surfaces, so it has no
584    /// flag of its own: a client reads `query` and `graph`.
585    #[serde(default)]
586    pub graph: bool,
587    /// Whether copy-on-write forks are served.
588    pub fork: bool,
589    /// Whether managed session reads are served.
590    #[serde(default)]
591    pub sessions: bool,
592    /// Whether the change feed is published (one change record per committed
593    /// notifying projector batch on the changes topic). Off by default: a
594    /// client that waits on an unpublished feed would wait forever.
595    #[serde(default)]
596    pub watch: bool,
597    /// Whether the authorization control band (`AGDX_AUTHZ_*`) is served, so a
598    /// console can show its roles and bindings surface. Off by default.
599    #[serde(default)]
600    pub authz: bool,
601    /// Consumer filters: server-side evaluation and the saved-filter catalog,
602    /// reported separately.
603    #[serde(default)]
604    pub filters: FilterCapsView,
605    pub versions: OpVersions,
606    /// Materialization backends the server currently exposes, so a client can
607    /// show what it may route to. Identity only (id + engine kind), no settings
608    /// or secrets. Empty (the default) is skipped on encode, so a pre-backends
609    /// capabilities reply stays byte-identical.
610    #[serde(default, skip_serializing_if = "Vec::is_empty")]
611    pub backends: Vec<BackendDescriptor>,
612    /// The stream/topic names this deployment uses, mirroring
613    /// [`crate::hello::BackendAnnounce::topology`]. Absent (the default) is
614    /// skipped on encode, so a pre-topology capabilities reply stays
615    /// byte-identical and an older reader falls back to
616    /// [`WireTopology::default`].
617    #[serde(default, skip_serializing_if = "Option::is_none")]
618    pub topology: Option<WireTopology>,
619}
620
621/// The managed query surface on the HTTP capabilities reply: whether it is
622/// served, its registry browse views, and the consistency it honors.
623#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
624pub struct QueryCapsView {
625    /// Whether `POST /agdx/query` is served.
626    pub available: bool,
627    /// Whether the projection registry browse routes are served.
628    pub projections: bool,
629    /// Whether the schema registry browse routes are served.
630    pub schemas: bool,
631    /// The strongest read-consistency the surface serves (the ladder
632    /// `eventual < read_your_writes < strong`, so a level implies the weaker
633    /// ones). Defaults to `eventual`, which every query surface serves.
634    #[serde(default)]
635    pub consistency: Consistency,
636    /// Whether lexical relevance search (`Query.text`) is served. Defaults
637    /// off, like every sub-feature: over-advertising would turn the clean
638    /// unsupported into a silent wrong answer.
639    #[serde(default)]
640    pub keyword: bool,
641    /// Whether opaque cursor page retrieval is served.
642    #[serde(default)]
643    pub cursor_paging: bool,
644    /// Whether an executing query can be cancelled by its execution identity.
645    #[serde(default)]
646    pub cancellation: bool,
647    /// Whether execution status can be polled independently of the first page.
648    #[serde(default)]
649    pub execution_status: bool,
650}
651
652#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
653pub struct DestinationCapsView {
654    pub available: bool,
655    #[serde(default)]
656    pub lifecycle: bool,
657    #[serde(default)]
658    pub checkpoint_status: bool,
659    #[serde(default)]
660    pub query_routes: bool,
661    #[serde(default)]
662    pub table_schema: bool,
663    #[serde(default)]
664    pub snapshots: bool,
665    #[serde(default)]
666    pub files: bool,
667    #[serde(default)]
668    pub metrics: bool,
669    #[serde(default)]
670    pub strongest_consistency: crate::checkpoint::CheckpointReadConsistency,
671}
672
673/// Consumer filters on the HTTP capabilities reply. Native evaluation is served
674/// by the streaming server itself and holds with the managed backend off. The
675/// catalog (saved filters and group bindings) needs the backend.
676#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
677pub struct FilterCapsView {
678    /// Filtered reads, fenced acknowledgments, sample tests, and validation.
679    #[serde(default)]
680    pub native: bool,
681    /// Group-aware reads, including pass-through when no policy is bound.
682    #[serde(default)]
683    pub group_policy_reads: bool,
684    /// Bounded previews over stored records.
685    #[serde(default)]
686    pub preview: bool,
687    /// Filtered reads from the partition primary through an attached session.
688    #[serde(default)]
689    pub primary_routing: bool,
690    /// Saved filters, revisions, and mutation outcomes.
691    #[serde(default)]
692    pub catalog: bool,
693    /// Consumer groups bound to a saved revision.
694    #[serde(default)]
695    pub managed_groups: bool,
696    /// The evaluation rules served (`0` when native evaluation is off).
697    #[serde(default)]
698    pub evaluator_version: u32,
699    /// The payload codecs a filter may select.
700    #[serde(default, skip_serializing_if = "Vec::is_empty")]
701    pub codecs: Vec<crate::filter::FilterCodec>,
702}
703
704impl FilterCapsView {
705    /// The view a server serving native evaluation and a ready catalog advertises.
706    pub fn served(native: bool, catalog: bool) -> Self {
707        let evaluation = native.then(FilterAnnounce::served);
708        Self {
709            native,
710            group_policy_reads: false,
711            preview: native,
712            primary_routing: native,
713            catalog,
714            managed_groups: catalog,
715            evaluator_version: evaluation
716                .as_ref()
717                .map_or(0, |evaluation| evaluation.evaluator_version),
718            codecs: evaluation
719                .map(|evaluation| evaluation.codecs)
720                .unwrap_or_default(),
721        }
722    }
723
724    #[must_use]
725    pub const fn with_group_policy_reads(mut self, served: bool) -> Self {
726        self.group_policy_reads = served;
727        self
728    }
729
730    /// Returns a copy reporting the evaluation a server announced instead of
731    /// this build's own.
732    #[must_use]
733    pub fn with_evaluation(mut self, evaluation: FilterAnnounce) -> Self {
734        if self.native {
735            self.evaluator_version = evaluation.evaluator_version;
736            self.codecs = evaluation.codecs;
737        }
738        self
739    }
740}
741
742/// The managed key-value surface on the HTTP capabilities reply.
743#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
744pub struct KvCapsView {
745    /// Whether the get/set/scan routes are served.
746    pub available: bool,
747    /// Whether compare-and-swap (`AGDX_KV_CAS`) is served. Independent of plain
748    /// get/set: a backend that cannot do a conditional write leaves it off and a
749    /// `cas` returns a clean unsupported error.
750    #[serde(default)]
751    pub cas: bool,
752    /// Whether fenced compare-and-swap (`AGDX_KV_CAS_FENCED`) is served. Independent
753    /// of plain `cas`: a backend leaves it off when it cannot gate a write on a live
754    /// fence sequence.
755    #[serde(default)]
756    pub cas_fenced: bool,
757    /// Whether the full revocable fenced-lease contract is served: holder-scoped
758    /// acquire, renewal, release, live-lease fenced CAS, and barriered reads.
759    #[serde(default)]
760    pub fenced_leases: bool,
761}
762
763impl Capabilities {
764    /// Constructor for the non-exhaustive wire struct. The core surfaces track
765    /// `enabled`, the way the binary `AGDX_HELLO` probe answers. The per-surface
766    /// sub-features (`kv.cas`, `query.consistency` above `eventual`, `graph`) all
767    /// start off: a server must opt into each, never over-advertising (which would
768    /// turn a clean unsupported error into a silent wrong answer). Build them with
769    /// [`from_versions`](Self::from_versions) or the setters.
770    pub fn new(enabled: bool, versions: OpVersions) -> Self {
771        Self {
772            managed: enabled,
773            query: QueryCapsView {
774                available: enabled,
775                projections: enabled,
776                schemas: enabled,
777                consistency: Consistency::Eventual,
778                keyword: false,
779                cursor_paging: false,
780                cancellation: false,
781                execution_status: false,
782            },
783            destinations: DestinationCapsView {
784                available: false,
785                lifecycle: false,
786                checkpoint_status: false,
787                query_routes: false,
788                table_schema: false,
789                snapshots: false,
790                files: false,
791                metrics: false,
792                strongest_consistency:
793                    crate::checkpoint::CheckpointReadConsistency::PotentiallyStale,
794            },
795            kv: KvCapsView {
796                available: enabled,
797                cas: false,
798                cas_fenced: false,
799                fenced_leases: false,
800            },
801            graph: false,
802            fork: enabled,
803            sessions: false,
804            watch: false,
805            authz: false,
806            filters: FilterCapsView::default(),
807            versions,
808            backends: Vec::new(),
809            topology: None,
810        }
811    }
812
813    /// Advertise the consumer-filter surface.
814    #[must_use]
815    pub fn with_filters(mut self, filters: FilterCapsView) -> Self {
816        self.filters = filters;
817        self
818    }
819
820    /// Advertise that the knowledge-graph ops are served. Off by default: a
821    /// server sets it only when a backend implements the graph surface.
822    #[must_use]
823    pub fn with_graph(mut self, value: bool) -> Self {
824        self.graph = value;
825        self
826    }
827
828    /// Set whether lexical keyword search is served on the query surface.
829    #[must_use]
830    pub fn with_query_keyword(mut self, value: bool) -> Self {
831        self.query.keyword = value;
832        self
833    }
834
835    #[must_use]
836    pub fn with_query_execution(mut self, paging: bool, cancellation: bool, status: bool) -> Self {
837        self.query.cursor_paging = paging;
838        self.query.cancellation = cancellation;
839        self.query.execution_status = status;
840        self
841    }
842
843    #[must_use]
844    pub fn with_destinations(mut self, destinations: DestinationCapsView) -> Self {
845        self.destinations = destinations;
846        self
847    }
848
849    /// Advertise that managed session reads are served.
850    #[must_use]
851    pub fn with_sessions(mut self, value: bool) -> Self {
852        self.sessions = value;
853        self
854    }
855
856    /// Advertise that the change feed is published. Only a deployment whose
857    /// projector emits change records may set it.
858    #[must_use]
859    pub fn with_watch(mut self, value: bool) -> Self {
860        self.watch = value;
861        self
862    }
863
864    /// Advertise that the authorization control band is served.
865    #[must_use]
866    pub fn with_authz(mut self, value: bool) -> Self {
867        self.authz = value;
868        self
869    }
870
871    /// Advertise the materialization backends the server exposes. The wire pins
872    /// no engine, so a server lists whatever it has open by id and kind.
873    #[must_use]
874    pub fn with_backends(mut self, backends: Vec<BackendDescriptor>) -> Self {
875        self.backends = backends;
876        self
877    }
878
879    /// Advertise the deployment's `topology`, mirroring
880    /// [`crate::hello::BackendAnnounce::with_topology`].
881    #[must_use]
882    pub fn with_topology(mut self, topology: WireTopology) -> Self {
883        self.topology = Some(topology);
884        self
885    }
886
887    /// Advertise compare-and-swap on the KV surface (`AGDX_KV_CAS`). Only a
888    /// backend that does a genuine conditional write may set it.
889    #[must_use]
890    pub fn with_kv_cas(mut self, on: bool) -> Self {
891        self.kv.cas = on;
892        self
893    }
894
895    /// Advertise fenced compare-and-swap on the KV surface (`AGDX_KV_CAS_FENCED`).
896    /// Only a backend that gates a write on a live fence sequence may set it.
897    #[must_use]
898    pub fn with_kv_cas_fenced(mut self, on: bool) -> Self {
899        self.kv.cas_fenced = on;
900        self
901    }
902
903    /// Advertise the full revocable fenced-lease contract. This subsumes the
904    /// older fence-only `cas_fenced` capability.
905    #[must_use]
906    pub fn with_kv_fenced_leases(mut self, on: bool) -> Self {
907        self.kv.fenced_leases = on;
908        self.kv.cas_fenced |= on;
909        self
910    }
911
912    /// Advertise the strongest read-consistency the query surface serves.
913    #[must_use]
914    pub fn with_query_consistency(mut self, level: Consistency) -> Self {
915        self.query.consistency = level;
916        self
917    }
918
919    /// Build the HTTP capabilities from the same `OpVersions` the binary
920    /// `AGDX_HELLO` probe answers with, reading the per-surface sub-features
921    /// straight off its `features` bitset and `graph` op version. A server SHOULD
922    /// use this so its two capability carriages (the binary `features` bits and
923    /// these HTTP fields) cannot disagree: the one source drives both.
924    pub fn from_versions(enabled: bool, versions: OpVersions) -> Self {
925        use crate::hello::feature;
926        let consistency = if versions.has_feature(feature::STRONG_CONSISTENCY) {
927            Consistency::Strong
928        } else if versions.has_feature(feature::READ_YOUR_WRITES) {
929            Consistency::ReadYourWrites
930        } else {
931            Consistency::Eventual
932        };
933        let destinations_available =
934            enabled && versions.checkpoint > 0 && versions.has_feature(feature::DESTINATIONS);
935        Self::new(enabled, versions)
936            .with_kv_cas(versions.has_feature(feature::KV_CAS))
937            .with_kv_cas_fenced(versions.has_feature(feature::KV_CAS_FENCED))
938            .with_kv_fenced_leases(versions.has_feature(feature::KV_FENCED_LEASES))
939            .with_query_keyword(versions.has_feature(feature::KEYWORD_SEARCH))
940            .with_watch(versions.has_feature(feature::WATCH))
941            .with_sessions(enabled && versions.has_feature(feature::SESSIONS))
942            .with_authz(versions.has_feature(feature::AUTHZ))
943            .with_query_consistency(consistency)
944            .with_destinations(DestinationCapsView {
945                available: destinations_available,
946                lifecycle: destinations_available,
947                checkpoint_status: destinations_available,
948                query_routes: destinations_available,
949                table_schema: false,
950                snapshots: false,
951                files: false,
952                metrics: false,
953                strongest_consistency:
954                    crate::checkpoint::CheckpointReadConsistency::PotentiallyStale,
955            })
956            // The graph surface needs a backend that serves it, advertised as a
957            // non-zero graph op version, so it is gated on that rather than implied.
958            .with_graph(enabled && versions.graph > 0)
959            .with_filters(
960                FilterCapsView::served(
961                    versions.has_feature(feature::CONSUMER_FILTERS),
962                    enabled && versions.filter > 0,
963                )
964                .with_group_policy_reads(versions.has_feature(feature::GROUP_POLICY_READS)),
965            )
966    }
967}
968
969/// One KV entry on the HTTP surface. `key` and `value` are URL-safe unpadded
970/// base64, because keys and values are arbitrary bytes that JSON strings
971/// cannot carry raw.
972#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
973pub struct KvEntryView {
974    pub key: String,
975    pub value: String,
976    pub expires_at_micros: Option<u64>,
977    /// The memory scope, present only on a memory read-view row (kind, agent,
978    /// user, app, conversation, and a `SourceRef` source pointer to the origin
979    /// log record). Absent on a generic entry, so a UI can fold a recalled item
980    /// back to its source message. Skipped on the wire when absent.
981    #[serde(default, skip_serializing_if = "Option::is_none")]
982    pub scope: Option<crate::kv::MemoryRowScope>,
983    /// The origin log record this entry was folded from, mirroring
984    /// [`crate::kv::KvEntry::source`]: every managed write is log-first, so a
985    /// stored entry points back to the record that wrote it. Skipped when the
986    /// store did not stamp one.
987    #[serde(default, skip_serializing_if = "Option::is_none")]
988    pub source: Option<crate::graph::SourceRef>,
989}
990
991/// One KV scan page on the HTTP surface.
992#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
993pub struct KvPageView {
994    pub entries: Vec<KvEntryView>,
995    pub cursor: Option<String>,
996}
997
998/// `DELETE /agdx/kv/{namespace}` reply: the number of entries removed.
999#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1000pub struct DeletedManyView {
1001    pub deleted: usize,
1002}
1003
1004/// One connection on the `GET /agdx/clients` discovery surface. `metadata` is
1005/// URL-safe unpadded base64 of the opaque advertised bytes (a JSON string cannot
1006/// carry raw bytes), or `None` when the connection advertised none. The console
1007/// decodes and interprets it per producer kind (an agent card or an app blob).
1008#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1009pub struct ClientMetadataView {
1010    pub client_id: u32,
1011    pub user_id: Option<u32>,
1012    pub transport: u8,
1013    pub address: String,
1014    pub consumer_groups_count: u32,
1015    #[serde(default, skip_serializing_if = "Option::is_none")]
1016    pub metadata: Option<String>,
1017}
1018
1019/// One page of the `GET /agdx/clients` discovery read. `next_cursor` is the
1020/// `after` query parameter for the next page, or `None` on the last page.
1021#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1022pub struct ClientMetadataListView {
1023    pub clients: Vec<ClientMetadataView>,
1024    #[serde(default, skip_serializing_if = "Option::is_none")]
1025    pub next_cursor: Option<u32>,
1026}
1027
1028/// `GET /agdx/clients` query parameters: the discovery filters and the page
1029/// window, all optional. Shared by the server handler and the typed client so the
1030/// query string cannot drift between them.
1031#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1032pub struct ClientsQuery {
1033    /// Only return connections that advertised metadata.
1034    #[serde(default, skip_serializing_if = "core::ops::Not::not")]
1035    pub with_metadata_only: bool,
1036    /// Only return connections authenticated as this principal.
1037    #[serde(default, skip_serializing_if = "Option::is_none")]
1038    pub user_id: Option<u32>,
1039    /// Page cursor: only connections with `client_id` greater than this.
1040    #[serde(default, skip_serializing_if = "Option::is_none")]
1041    pub after: Option<u32>,
1042    /// Max entries per page (clamped server-side to the page cap).
1043    #[serde(default, skip_serializing_if = "Option::is_none")]
1044    pub limit: Option<u32>,
1045}
1046
1047/// `POST /agdx/forks/{id}/promote` reply: the number of rows applied.
1048#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1049pub struct PromotedView {
1050    pub rows: usize,
1051}
1052
1053/// One graph node on the HTTP surface: id as a string, its labels, and its
1054/// attributes rendered as strings (e.g. the entity `value`) for a browser or
1055/// wasm client that has no access to the typed `Value`.
1056#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1057pub struct GraphNodeView {
1058    pub id: String,
1059    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1060    pub labels: Vec<String>,
1061    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1062    pub attrs: Vec<(String, String)>,
1063    /// The source this node was first observed in, if known. See [`SourceRef`].
1064    #[serde(default, skip_serializing_if = "Option::is_none")]
1065    pub source: Option<SourceRef>,
1066}
1067
1068/// One graph edge on the HTTP surface: endpoint ids as strings, the type, and the
1069/// weight.
1070#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1071pub struct GraphEdgeView {
1072    pub id: String,
1073    pub from: String,
1074    pub to: String,
1075    pub edge_type: String,
1076    pub weight: f32,
1077    /// Valid-time window (epoch micros) for a bitemporal edge, omitted when open.
1078    #[serde(default, skip_serializing_if = "Option::is_none")]
1079    pub valid_from: Option<u64>,
1080    #[serde(default, skip_serializing_if = "Option::is_none")]
1081    pub valid_to: Option<u64>,
1082    /// The source that most recently asserted this relationship, if known. See
1083    /// [`SourceRef`].
1084    #[serde(default, skip_serializing_if = "Option::is_none")]
1085    pub source: Option<SourceRef>,
1086}
1087
1088/// `POST /agdx/graph/{name}/query` reply: the reachable nodes, traversed edges,
1089/// and (for a `paths` return) the reconstructed paths as id sequences.
1090#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
1091pub struct GraphResultView {
1092    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1093    pub nodes: Vec<GraphNodeView>,
1094    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1095    pub edges: Vec<GraphEdgeView>,
1096    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1097    pub paths: Vec<PathView>,
1098}
1099
1100/// One path in a `GraphResultView`: parallel node and edge id sequences, ids as
1101/// Crockford-base32 strings (the JSON view of [`crate::graph::Path`]).
1102#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
1103pub struct PathView {
1104    pub nodes: Vec<String>,
1105    pub edges: Vec<String>,
1106}
1107
1108/// `POST /agdx/schemas` body: the register request without an id. The managed
1109/// backend allocates it and the reply carries it back as
1110/// `{"SchemaRegistered":id}`.
1111#[derive(Clone, Debug, Serialize, Deserialize)]
1112pub struct RegisterSchemaBody {
1113    pub source: crate::control::SchemaSource,
1114    #[serde(default, skip_serializing_if = "Option::is_none")]
1115    pub name: Option<String>,
1116    #[serde(default, skip_serializing_if = "Option::is_none")]
1117    pub version: Option<u32>,
1118    /// The stream whose registry allocates the id. Absent for the
1119    /// deployment-wide registry.
1120    #[serde(default, skip_serializing_if = "Option::is_none")]
1121    pub stream: Option<String>,
1122}
1123
1124/// `POST /agdx/schemas/{id}/decode` body: the record payload as URL-safe
1125/// unpadded base64.
1126#[derive(Clone, Debug, Serialize, Deserialize)]
1127pub struct DecodeRecordBody {
1128    pub payload: String,
1129}
1130
1131/// `POST /agdx/forks` body.
1132#[derive(Clone, Debug, Serialize, Deserialize)]
1133pub struct ForkCreateBody {
1134    pub fork_id: String,
1135    #[serde(default, skip_serializing_if = "Option::is_none")]
1136    pub parent: Option<String>,
1137    #[serde(default)]
1138    pub kind: ForkKind,
1139    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1140    pub tables: Vec<String>,
1141}
1142
1143/// `PUT /agdx/forks/{id}/rows` body. `payload_b64` is URL-safe unpadded base64,
1144/// like every binary body on this surface.
1145#[derive(Clone, Debug, Serialize, Deserialize)]
1146pub struct ForkPutBody {
1147    pub table: String,
1148    pub partition_id: u32,
1149    pub offset: u64,
1150    #[serde(default)]
1151    pub projection_id: String,
1152    #[serde(default)]
1153    pub projection_version: u32,
1154    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
1155    pub fields: BTreeMap<String, String>,
1156    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
1157    pub metadata: BTreeMap<String, String>,
1158    #[serde(default, skip_serializing_if = "Option::is_none")]
1159    pub payload_b64: Option<String>,
1160    #[serde(default, skip_serializing_if = "Option::is_none")]
1161    pub embedding: Option<String>,
1162    #[serde(default)]
1163    pub tombstone: bool,
1164}
1165
1166/// `DELETE /agdx/bindings` body: which binding to remove, by its source stream
1167/// and topic. `projection_ref` absent removes the whole binding for that source.
1168/// `projection_ref` present removes only that one projection from the binding,
1169/// leaving the rest. Mirrors `ControlCommand::RemoveBinding`.
1170#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1171pub struct RemoveBindingBody {
1172    pub stream: String,
1173    pub topic: String,
1174    #[serde(default, skip_serializing_if = "Option::is_none")]
1175    pub projection_ref: Option<String>,
1176}
1177
1178/// The canonical error body every `/agdx/*` route returns on a non-2xx status.
1179/// The HTTP binding's rule is "a 2xx carries the bare `Ok` payload, a failure
1180/// carries this": the status line gives the coarse class (from
1181/// [`ResultCode::http_status`]) and this body gives the machine-dispatchable
1182/// [`ResultCode`] plus a human `message`, so a client matches on `code` instead
1183/// of grepping the message text (which is for humans and may change). `detail`
1184/// carries optional structured context (e.g. the conflicting version on a CAS
1185/// miss) as free-form JSON.
1186#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1187pub struct ErrorBody {
1188    pub code: ResultCode,
1189    pub message: String,
1190    #[serde(default, skip_serializing_if = "Option::is_none")]
1191    pub detail: Option<serde_json::Value>,
1192}
1193
1194impl ErrorBody {
1195    /// An error body from a classified code and a human message.
1196    pub fn new(code: ResultCode, message: impl Into<String>) -> Self {
1197        Self {
1198            code,
1199            message: message.into(),
1200            detail: None,
1201        }
1202    }
1203
1204    /// Attach structured context.
1205    #[must_use]
1206    pub fn with_detail(mut self, detail: serde_json::Value) -> Self {
1207        self.detail = Some(detail);
1208        self
1209    }
1210
1211    /// The HTTP status this body's `code` maps to, so a server sets the status
1212    /// line and the body from one value.
1213    pub fn http_status(&self) -> u16 {
1214        self.code.http_status()
1215    }
1216}
1217
1218impl From<&QueryError> for ErrorBody {
1219    fn from(error: &QueryError) -> Self {
1220        Self::new(ResultCode::from(error), error.to_string())
1221    }
1222}
1223
1224impl From<&crate::filter::FilterError> for ErrorBody {
1225    fn from(error: &crate::filter::FilterError) -> Self {
1226        Self::new(error.code, error.message.clone())
1227            .with_detail(serde_json::json!({ "reason": error.reason }))
1228    }
1229}
1230
1231impl From<&KvError> for ErrorBody {
1232    fn from(error: &KvError) -> Self {
1233        Self::new(ResultCode::from(error), error.to_string())
1234    }
1235}
1236
1237impl From<&ForkError> for ErrorBody {
1238    fn from(error: &ForkError) -> Self {
1239        Self::new(ResultCode::from(error), error.to_string())
1240    }
1241}
1242
1243impl From<&crate::authz::AuthzError> for ErrorBody {
1244    fn from(error: &crate::authz::AuthzError) -> Self {
1245        Self::new(ResultCode::from(error), error.to_string())
1246    }
1247}
1248
1249impl From<&crate::session::SessionError> for ErrorBody {
1250    fn from(error: &crate::session::SessionError) -> Self {
1251        Self::new(ResultCode::from(error), error.to_string())
1252    }
1253}
1254
1255/// `?topic=` on `GET /agdx/projections`: filter to bindings off this source topic.
1256pub const PARAM_TOPIC: &str = "topic";
1257/// `?name_contains=` on a projection or schema list: substring over the name.
1258pub const PARAM_NAME_CONTAINS: &str = "name_contains";
1259/// `?id_prefix=` on `GET /agdx/projections`: keep ids starting with the prefix.
1260pub const PARAM_ID_PREFIX: &str = "id_prefix";
1261/// `?stream=` on the schema routes: the stream whose writer-schema registry
1262/// the request addresses. Absent addresses the deployment-wide registry.
1263pub const PARAM_STREAM: &str = "stream";
1264/// `?search=` on `GET /agdx/projections`: one substring matched against the
1265/// projection name OR id. A console with a single filter box maps to it. A
1266/// server matches it as `name_contains(name) OR id contains search`. Composes
1267/// (AND) with the narrower `name_contains` / `id_prefix` when several are set.
1268pub const PARAM_SEARCH: &str = "search";
1269/// `?prefix=` on a KV scan: base64url key prefix.
1270pub const PARAM_PREFIX: &str = "prefix";
1271/// `?start=` on a KV scan: base64url inclusive lower bound.
1272pub const PARAM_START: &str = "start";
1273/// `?end=` on a KV scan: base64url exclusive upper bound.
1274pub const PARAM_END: &str = "end";
1275/// `?key_contains=`: base64url substring the key must contain.
1276pub const PARAM_KEY_CONTAINS: &str = "key_contains";
1277/// `?limit=`: page size.
1278pub const PARAM_LIMIT: &str = "limit";
1279/// `?cursor=`: opaque continuation token from the prior page.
1280pub const PARAM_CURSOR: &str = "cursor";
1281/// `?status=` on a session list: one session status.
1282pub const PARAM_STATUS: &str = "status";
1283/// `?agent=` on a session list: sessions this agent took part in.
1284pub const PARAM_AGENT: &str = "agent";
1285/// `?text=` on a session list: a substring of the label or id.
1286pub const PARAM_TEXT: &str = "text";
1287/// `?root=` on a session list: the root of the session tree.
1288pub const PARAM_ROOT: &str = "root";
1289/// `?label_prefix=` on a session list: the start of the label.
1290pub const PARAM_LABEL_PREFIX: &str = "label_prefix";
1291/// `?surface=` on session links: one link surface.
1292pub const PARAM_SURFACE: &str = "surface";
1293/// `?after=` on session changes: the last change sequence already seen.
1294pub const PARAM_AFTER: &str = "after";
1295/// `?total=` on a session list: also count every match.
1296pub const PARAM_TOTAL: &str = "total";
1297/// `?fixed_frontier=` on session events: pin the fold frontier of the first
1298/// page for a historical walk.
1299pub const PARAM_FIXED_FRONTIER: &str = "fixed_frontier";
1300/// `?history_limit=` on session state: the most history rows to return.
1301pub const PARAM_HISTORY_LIMIT: &str = "history_limit";
1302/// `?expires_at_micros=` on a KV `PUT`: absolute expiry, epoch microseconds.
1303pub const PARAM_EXPIRES_AT_MICROS: &str = "expires_at_micros";
1304/// `?expect_version=` on a KV compare-and-swap: apply only if the key holds
1305/// this exact version.
1306pub const PARAM_EXPECT_VERSION: &str = "expect_version";
1307/// `?expect_absent=` on a KV compare-and-swap: apply only if the key is absent
1308/// (create-if-absent).
1309pub const PARAM_EXPECT_ABSENT: &str = "expect_absent";
1310
1311/// Response header on `GET /agdx/kv/{namespace}/{key}` carrying the entry's
1312/// absolute expiry (epoch microseconds) as a decimal string. The value itself
1313/// rides the raw response body, so this header carries the one piece of
1314/// out-of-band metadata a single-key read needs. Owned here so the name is a
1315/// wire constant rather than an unscoped string. Absent means no expiry.
1316pub const KV_EXPIRES_AT_MICROS_HEADER: &str = "agdx-expires-at-micros";
1317
1318/// `GET /agdx/projections` filters. Every field is optional, and an absent field is
1319/// omitted from the query string (no empty `topic=`). Field names are the
1320/// `PARAM_*` consts verbatim, so the client serializer and the server parser
1321/// share one spelling.
1322#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1323pub struct ProjectionListQuery {
1324    #[serde(default, skip_serializing_if = "Option::is_none")]
1325    pub topic: Option<String>,
1326    #[serde(default, skip_serializing_if = "Option::is_none")]
1327    pub name_contains: Option<String>,
1328    #[serde(default, skip_serializing_if = "Option::is_none")]
1329    pub id_prefix: Option<String>,
1330    /// One substring matched against the projection name OR id, for a console
1331    /// with a single filter box. Composes (AND) with `name_contains`/`id_prefix`.
1332    #[serde(default, skip_serializing_if = "Option::is_none")]
1333    pub search: Option<String>,
1334}
1335
1336/// `GET /agdx/schemas` filters. `name_contains` is the substring filter on a
1337/// schema's optional name, the same spelling as the projection list, so the two
1338/// list surfaces share one filter vocabulary.
1339#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1340pub struct SchemaListQuery {
1341    #[serde(default, skip_serializing_if = "Option::is_none")]
1342    pub name_contains: Option<String>,
1343    /// The stream whose registry is listed. Absent lists the deployment-wide
1344    /// registry.
1345    #[serde(default, skip_serializing_if = "Option::is_none")]
1346    pub stream: Option<String>,
1347}
1348
1349/// `?stream=` on `GET`/`DELETE /agdx/schemas/{id}` and
1350/// `POST /agdx/schemas/{id}/decode`: the stream whose registry holds the id.
1351/// Absent addresses the deployment-wide registry.
1352#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1353pub struct SchemaQuery {
1354    #[serde(default, skip_serializing_if = "Option::is_none")]
1355    pub stream: Option<String>,
1356}
1357
1358/// `GET /agdx/filters` query parameters.
1359#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1360pub struct FilterListQuery {
1361    #[serde(default, skip_serializing_if = "Option::is_none")]
1362    pub name_contains: Option<String>,
1363    #[serde(default, skip_serializing_if = "Option::is_none")]
1364    pub state: Option<crate::filter::FilterState>,
1365    /// Only filters with a lower id, for stable paging.
1366    #[serde(default, skip_serializing_if = "Option::is_none")]
1367    pub before_id: Option<u32>,
1368    #[serde(default, skip_serializing_if = "Option::is_none")]
1369    pub page: Option<u32>,
1370    #[serde(default, skip_serializing_if = "Option::is_none")]
1371    pub page_size: Option<u32>,
1372}
1373
1374/// `GET /agdx/filters/{id}/revisions` query parameters.
1375#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1376pub struct FilterRevisionListQuery {
1377    #[serde(default, skip_serializing_if = "Option::is_none")]
1378    pub page: Option<u32>,
1379    #[serde(default, skip_serializing_if = "Option::is_none")]
1380    pub page_size: Option<u32>,
1381}
1382
1383/// `GET /agdx/filter-bindings` query parameters.
1384#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1385pub struct FilterBindingListQuery {
1386    #[serde(default, skip_serializing_if = "Option::is_none")]
1387    pub filter_id: Option<u32>,
1388    #[serde(default, skip_serializing_if = "Option::is_none")]
1389    pub stream: Option<String>,
1390    #[serde(default, skip_serializing_if = "Option::is_none")]
1391    pub topic: Option<String>,
1392    #[serde(default, skip_serializing_if = "Option::is_none")]
1393    pub page: Option<u32>,
1394    #[serde(default, skip_serializing_if = "Option::is_none")]
1395    pub page_size: Option<u32>,
1396}
1397
1398/// `POST /agdx/filters/test` body. The sample payload is given either as
1399/// base64url bytes or as UTF-8 text, never both.
1400#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
1401pub struct FilterTestBody {
1402    pub filter: crate::filter::FilterRef,
1403    #[serde(default, skip_serializing_if = "Option::is_none")]
1404    pub payload_base64: Option<String>,
1405    #[serde(default, skip_serializing_if = "Option::is_none")]
1406    pub payload_text: Option<String>,
1407    #[serde(default, skip_serializing_if = "Vec::is_empty")]
1408    pub headers: Vec<crate::filter::FilterHeader>,
1409}
1410
1411/// `GET /agdx/kv/{namespace}` scan filters. The byte-valued bounds
1412/// (`prefix`/`start`/`end`/`key_contains`) are base64url, like every binary
1413/// value on this surface.
1414#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1415pub struct KvScanQuery {
1416    #[serde(default, skip_serializing_if = "Option::is_none")]
1417    pub prefix: Option<String>,
1418    #[serde(default, skip_serializing_if = "Option::is_none")]
1419    pub start: Option<String>,
1420    #[serde(default, skip_serializing_if = "Option::is_none")]
1421    pub end: Option<String>,
1422    #[serde(default, skip_serializing_if = "Option::is_none")]
1423    pub key_contains: Option<String>,
1424    /// The conversation lens: keep only rows the given conversation wrote (the
1425    /// text form of its `gen_ai.conversation.id`). The memory read view stamps
1426    /// this on each record, so a scan of a memory namespace narrows to one
1427    /// conversation. Plain text, not base64url, since a conversation id is text.
1428    #[serde(default, skip_serializing_if = "Option::is_none")]
1429    pub conversation: Option<String>,
1430    #[serde(default, skip_serializing_if = "Option::is_none")]
1431    pub limit: Option<usize>,
1432    #[serde(default, skip_serializing_if = "Option::is_none")]
1433    pub cursor: Option<String>,
1434}
1435
1436/// `PUT /agdx/kv/{namespace}/{key}` query: an optional absolute expiry.
1437#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1438pub struct KvPutQuery {
1439    #[serde(default, skip_serializing_if = "Option::is_none")]
1440    pub expires_at_micros: Option<u64>,
1441}
1442
1443/// `GET /agdx/graph/{name}/neighbors/{node}` query: the traversal direction
1444/// (`out`, `in`, or `both`, omitted for the default `out`), an optional edge-type
1445/// filter, the hop depth (omitted for the default one hop), and a result limit
1446/// (omitted for the backend ceiling). One struct shared by the typed client and
1447/// the server route, so the two cannot drift.
1448#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1449pub struct GraphNeighborsQuery {
1450    #[serde(default, skip_serializing_if = "Option::is_none")]
1451    pub dir: Option<String>,
1452    #[serde(default, skip_serializing_if = "Option::is_none")]
1453    pub edge_type: Option<String>,
1454    #[serde(default, skip_serializing_if = "Option::is_none")]
1455    pub depth: Option<u32>,
1456    #[serde(default, skip_serializing_if = "Option::is_none")]
1457    pub limit: Option<usize>,
1458    /// Valid-time "as of" read (epoch micros): only edges valid at this instant.
1459    #[serde(default, skip_serializing_if = "Option::is_none")]
1460    pub as_of: Option<u64>,
1461    /// Restrict to elements a given conversation asserted (the text form of its
1462    /// `gen_ai.conversation.id`). Omitted reads the whole graph. The conversation
1463    /// lens over the neighbors route.
1464    #[serde(default, skip_serializing_if = "Option::is_none")]
1465    pub conversation: Option<String>,
1466}
1467
1468/// Filters for `GET /agdx/sessions/{stream}`. The stream is the path
1469/// segment. A `limit` of zero or none leaves the page size to the server.
1470#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1471pub struct SessionsQuery {
1472    #[serde(default, skip_serializing_if = "Option::is_none")]
1473    pub status: Option<String>,
1474    #[serde(default, skip_serializing_if = "Option::is_none")]
1475    pub root: Option<String>,
1476    #[serde(default, skip_serializing_if = "Option::is_none")]
1477    pub label_prefix: Option<String>,
1478    #[serde(default, skip_serializing_if = "Option::is_none")]
1479    pub agent: Option<String>,
1480    #[serde(default, skip_serializing_if = "Option::is_none")]
1481    pub text: Option<String>,
1482    #[serde(default, skip_serializing_if = "Option::is_none")]
1483    pub cursor: Option<String>,
1484    #[serde(default, skip_serializing_if = "Option::is_none")]
1485    pub limit: Option<u32>,
1486    #[serde(default, rename = "total")]
1487    pub want_total: bool,
1488}
1489
1490/// Query of `GET /agdx/sessions/{stream}/{id}/events`. A `limit` of zero or
1491/// none leaves the page size to the server.
1492#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1493pub struct SessionEventsQuery {
1494    #[serde(default, skip_serializing_if = "Option::is_none")]
1495    pub cursor: Option<String>,
1496    #[serde(default, skip_serializing_if = "Option::is_none")]
1497    pub limit: Option<u32>,
1498    #[serde(default)]
1499    pub fixed_frontier: bool,
1500}
1501
1502/// Query of `GET /agdx/sessions/{stream}/{id}/state`. A `history_limit` of
1503/// zero or none leaves the history size to the server.
1504#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1505pub struct SessionStateQuery {
1506    #[serde(default, skip_serializing_if = "Option::is_none")]
1507    pub history_limit: Option<u32>,
1508}
1509
1510/// Query of `GET /agdx/sessions/{stream}/{id}/links`.
1511#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1512pub struct SessionLinksQuery {
1513    #[serde(default, skip_serializing_if = "Option::is_none")]
1514    pub surface: Option<String>,
1515}
1516
1517/// Query of `GET /agdx/sessions/{stream}/changes`. A `limit` of zero or none
1518/// leaves the page size to the server.
1519#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1520pub struct SessionChangesQuery {
1521    #[serde(default)]
1522    pub after: u64,
1523    #[serde(default, skip_serializing_if = "Option::is_none")]
1524    pub limit: Option<u32>,
1525}
1526
1527/// `PUT /agdx/kv/{namespace}/{key}/cas` query: the compare-and-swap precondition
1528/// plus an optional expiry. Exactly one of `expect_version` (match the held
1529/// version) or `expect_absent` (create-if-absent) is set, mirroring the binary
1530/// `CasExpect`. The value rides the raw request body.
1531#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
1532pub struct KvCasQuery {
1533    #[serde(default, skip_serializing_if = "Option::is_none")]
1534    pub expect_version: Option<u64>,
1535    #[serde(default, skip_serializing_if = "Option::is_none")]
1536    pub expect_absent: Option<bool>,
1537    #[serde(default, skip_serializing_if = "Option::is_none")]
1538    pub expires_at_micros: Option<u64>,
1539}
1540
1541/// `PUT /agdx/kv/{namespace}/{key}/cas` reply on success: the new version the
1542/// committed write took.
1543#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
1544pub struct CasCommittedView {
1545    pub version: u64,
1546}
1547
1548#[cfg(test)]
1549mod tests {
1550    use super::*;
1551
1552    #[test]
1553    fn given_path_builders_when_rendered_then_should_match_the_router() {
1554        assert_eq!(CAPABILITIES_PATH, "/agdx/capabilities");
1555        assert_eq!(
1556            projection_path("reading.v1"),
1557            "/agdx/projections/reading.v1"
1558        );
1559        assert_eq!(schema_path(7), "/agdx/schemas/7");
1560        assert_eq!(schema_decode_path(7), "/agdx/schemas/7/decode");
1561        assert_eq!(kv_namespace_path("sessions"), "/agdx/kv/sessions");
1562        assert_eq!(
1563            kv_entry_path("sessions", "dXNlcjo0Mg"),
1564            "/agdx/kv/sessions/dXNlcjo0Mg"
1565        );
1566        assert_eq!(
1567            kv_cas_path("sessions", "dXNlcjo0Mg"),
1568            "/agdx/kv/sessions/dXNlcjo0Mg/cas"
1569        );
1570        assert_eq!(fork_path("f1"), "/agdx/forks/f1");
1571        assert_eq!(fork_promote_path("f1"), "/agdx/forks/f1/promote");
1572        assert_eq!(fork_rows_path("f1"), "/agdx/forks/f1/rows");
1573    }
1574
1575    #[test]
1576    fn given_stream_scoped_names_when_rendered_then_should_encode_each_as_one_segment() {
1577        assert_eq!(
1578            kv_namespace_path("stream:acme/support"),
1579            "/agdx/kv/stream%3Aacme%2Fsupport"
1580        );
1581        assert_eq!(
1582            kv_cas_path("stream:acme/support", "dXNlcjo0Mg"),
1583            "/agdx/kv/stream%3Aacme%2Fsupport/dXNlcjo0Mg/cas"
1584        );
1585        assert_eq!(
1586            graph_neighbors_path("stream:acme/kg", "a/b"),
1587            "/agdx/graph/stream%3Aacme%2Fkg/neighbors/a%2Fb"
1588        );
1589        assert_eq!(
1590            graph_query_path("stream:acme/kg"),
1591            "/agdx/graph/stream%3Aacme%2Fkg/query"
1592        );
1593        assert_eq!(
1594            graph_path("stream:acme/kg"),
1595            "/agdx/graphs/stream%3Aacme%2Fkg"
1596        );
1597        assert_eq!(
1598            projection_path("stream:acme/readings"),
1599            "/agdx/projections/stream%3Aacme%2Freadings"
1600        );
1601    }
1602
1603    #[test]
1604    fn given_the_sessions_feature_when_built_from_versions_then_should_advertise_sessions() {
1605        let mut versions = OpVersions::new(1, 1, 1, 1);
1606        assert!(!Capabilities::from_versions(true, versions).sessions);
1607        versions.features |= crate::hello::feature::SESSIONS;
1608        assert!(Capabilities::from_versions(true, versions).sessions);
1609        assert!(!Capabilities::from_versions(false, versions).sessions);
1610    }
1611
1612    #[test]
1613    fn given_capabilities_when_constructed_then_extended_features_default_off() {
1614        let caps = Capabilities::new(true, OpVersions::new(1, 1, 1, 1));
1615        assert!(
1616            caps.query.available && caps.kv.available && caps.fork,
1617            "core surfaces track enabled"
1618        );
1619        assert!(
1620            !caps.kv.cas
1621                && !caps.kv.cas_fenced
1622                && !caps.kv.fenced_leases
1623                && caps.query.consistency == Consistency::Eventual,
1624            "sub-features must be opt-in, never on by default"
1625        );
1626        let opted = caps
1627            .with_kv_cas(true)
1628            .with_query_consistency(Consistency::ReadYourWrites);
1629        assert!(opted.kv.cas && opted.query.consistency == Consistency::ReadYourWrites);
1630    }
1631
1632    #[test]
1633    fn given_the_fenced_lease_bit_when_building_http_capabilities_then_should_advertise_it() {
1634        let versions =
1635            OpVersions::new(1, 1, 1, 1).with_features(crate::hello::feature::KV_FENCED_LEASES);
1636        let caps = Capabilities::from_versions(true, versions);
1637        assert!(caps.kv.fenced_leases);
1638        assert!(
1639            caps.kv.cas_fenced,
1640            "the full fenced-lease contract subsumes fenced CAS"
1641        );
1642    }
1643
1644    #[test]
1645    fn given_group_reads_without_evaluation_when_building_http_capabilities_then_should_keep_catalog_available()
1646     {
1647        let versions = OpVersions::new(1, 1, 1, 1)
1648            .with_filter(crate::codes::FILTER_OP_VERSION)
1649            .with_features(crate::hello::feature::GROUP_POLICY_READS);
1650        let caps = Capabilities::from_versions(true, versions);
1651        assert!(caps.filters.group_policy_reads);
1652        assert!(caps.filters.catalog && caps.filters.managed_groups);
1653        assert!(!caps.filters.native && !caps.filters.preview);
1654        assert_eq!(caps.filters.evaluator_version, 0);
1655        assert!(caps.filters.codecs.is_empty());
1656        let absent = Capabilities::from_versions(false, OpVersions::new(1, 1, 1, 1));
1657        assert!(!absent.filters.group_policy_reads && !absent.filters.catalog);
1658    }
1659
1660    #[test]
1661    fn given_destination_versions_when_building_capabilities_then_the_core_surface_is_available() {
1662        use crate::codes::CHECKPOINT_OP_VERSION;
1663        use crate::hello::feature;
1664
1665        let versions = OpVersions::new(2, 2, 1, 1)
1666            .with_checkpoint(CHECKPOINT_OP_VERSION)
1667            .with_features(feature::DESTINATIONS);
1668        let caps = Capabilities::from_versions(true, versions);
1669
1670        assert!(caps.destinations.available);
1671        assert!(caps.destinations.lifecycle);
1672        assert!(caps.destinations.checkpoint_status);
1673        assert!(caps.destinations.query_routes);
1674        assert!(!caps.destinations.table_schema);
1675    }
1676
1677    #[test]
1678    fn given_capabilities_backends_when_json_round_tripped_then_should_preserve_and_omit_empty() {
1679        use crate::destination::BackendResourceId;
1680        use crate::hello::{BackendDescriptor, BackendImplementation, BackendMode};
1681        let caps = Capabilities::new(true, OpVersions::new(1, 1, 1, 1)).with_backends(vec![
1682            BackendDescriptor::new(
1683                BackendResourceId::from_u128(1),
1684                BackendMode::Operational,
1685                "Embedded",
1686                BackendImplementation {
1687                    kind: "embedded".to_owned(),
1688                    version: "1.0.0".to_owned(),
1689                },
1690                1,
1691                1,
1692            ),
1693            BackendDescriptor::new(
1694                BackendResourceId::from_u128(2),
1695                BackendMode::Lakehouse,
1696                "Warehouse",
1697                BackendImplementation {
1698                    kind: "columnar".to_owned(),
1699                    version: "1.0.0".to_owned(),
1700                },
1701                1,
1702                1,
1703            ),
1704        ]);
1705        let json = serde_json::to_string(&caps).expect("serializes");
1706        let back: Capabilities = serde_json::from_str(&json).expect("deserializes");
1707        assert_eq!(back.backends.len(), 2);
1708        assert_eq!(
1709            back.backends[1].resource_id,
1710            BackendResourceId::from_u128(2)
1711        );
1712        assert_eq!(back.backends[1].implementation.kind, "columnar");
1713
1714        // No advertised backends is omitted on the wire, so a pre-backends
1715        // capabilities reply stays byte-identical.
1716        let plain = Capabilities::new(true, OpVersions::new(1, 1, 1, 1));
1717        let json = serde_json::to_string(&plain).expect("json");
1718        assert!(!json.contains("backends"), "empty backends omitted: {json}");
1719    }
1720
1721    #[test]
1722    fn given_capabilities_topology_when_json_round_tripped_then_should_preserve_and_omit_absent() {
1723        let custom = WireTopology {
1724            ops_stream: "custom-ops".to_owned(),
1725            ..WireTopology::default()
1726        };
1727        let caps =
1728            Capabilities::new(true, OpVersions::new(1, 1, 1, 1)).with_topology(custom.clone());
1729        let json = serde_json::to_string(&caps).expect("serializes");
1730        let back: Capabilities = serde_json::from_str(&json).expect("deserializes");
1731        assert_eq!(back.topology, Some(custom));
1732
1733        // No advertised topology (the default) is omitted on the wire, so a
1734        // pre-topology capabilities reply stays byte-identical.
1735        let plain = Capabilities::new(true, OpVersions::new(1, 1, 1, 1));
1736        let json = serde_json::to_string(&plain).expect("json");
1737        assert!(
1738            !json.contains("topology"),
1739            "absent topology omitted: {json}"
1740        );
1741    }
1742
1743    #[test]
1744    fn given_a_typed_error_when_made_into_a_body_then_should_carry_code_and_message() {
1745        let body = ErrorBody::from(&QueryError::IndexNotFound("readings".to_owned()));
1746        assert_eq!(body.code, ResultCode::NotFound);
1747        assert_eq!(body.http_status(), 404);
1748        assert!(body.message.contains("readings"));
1749        // Round-trips as JSON, the form the HTTP surface serves it in.
1750        let json = serde_json::to_string(&body).expect("serializes");
1751        let back: ErrorBody = serde_json::from_str(&json).expect("deserializes");
1752        assert_eq!(back, body);
1753    }
1754
1755    #[test]
1756    #[cfg(feature = "http-client")]
1757    fn given_scan_filters_when_url_encoded_then_should_omit_absent_fields() {
1758        let query = KvScanQuery {
1759            prefix: Some("dXNlcjo".to_owned()),
1760            limit: Some(50),
1761            ..Default::default()
1762        };
1763        let encoded = serde_urlencoded::to_string(&query).expect("encodes");
1764        assert_eq!(encoded, "prefix=dXNlcjo&limit=50");
1765        // Field names are the PARAM_* consts verbatim.
1766        assert!(encoded.contains(&format!("{PARAM_PREFIX}=")));
1767        assert!(encoded.contains(&format!("{PARAM_LIMIT}=")));
1768    }
1769
1770    #[test]
1771    #[cfg(feature = "http-client")]
1772    fn given_names_with_reserved_characters_when_building_a_binding_path_then_should_encode_each_segment()
1773     {
1774        assert_eq!(
1775            filter_binding_path("orbit", "fleet/changes", "desk?x#1"),
1776            "/agdx/filter-bindings/orbit/fleet%2Fchanges/desk%3Fx%231"
1777        );
1778    }
1779
1780    #[cfg(feature = "http-client")]
1781    #[test]
1782    fn given_list_filters_when_url_encoded_then_field_names_match_the_param_consts() {
1783        let projections = ProjectionListQuery {
1784            name_contains: Some("reading".to_owned()),
1785            id_prefix: Some("reading.".to_owned()),
1786            ..Default::default()
1787        };
1788        let encoded = serde_urlencoded::to_string(&projections).expect("encodes");
1789        assert_eq!(encoded, "name_contains=reading&id_prefix=reading.");
1790        assert!(encoded.contains(&format!("{PARAM_NAME_CONTAINS}=")));
1791        assert!(encoded.contains(&format!("{PARAM_ID_PREFIX}=")));
1792
1793        let schemas = SchemaListQuery {
1794            name_contains: Some("Reading".to_owned()),
1795            stream: None,
1796        };
1797        assert_eq!(
1798            serde_urlencoded::to_string(&schemas).expect("encodes"),
1799            "name_contains=Reading"
1800        );
1801        let scoped = SchemaListQuery {
1802            name_contains: None,
1803            stream: Some("acme".to_owned()),
1804        };
1805        assert_eq!(
1806            serde_urlencoded::to_string(&scoped).expect("encodes"),
1807            format!("{PARAM_STREAM}=acme")
1808        );
1809        assert_eq!(
1810            serde_urlencoded::to_string(SchemaQuery::default()).expect("encodes"),
1811            ""
1812        );
1813        assert_eq!(
1814            serde_urlencoded::to_string(SchemaQuery {
1815                stream: Some("acme".to_owned()),
1816            })
1817            .expect("encodes"),
1818            format!("{PARAM_STREAM}=acme")
1819        );
1820    }
1821
1822    #[test]
1823    fn given_a_register_schema_body_without_a_stream_when_encoded_then_should_omit_it() {
1824        let body = RegisterSchemaBody {
1825            source: crate::control::SchemaSource::JsonSchema {
1826                schema: "{}".to_owned(),
1827            },
1828            name: None,
1829            version: None,
1830            stream: None,
1831        };
1832        let json = serde_json::to_string(&body).expect("encodes");
1833        assert!(
1834            !json.contains("stream"),
1835            "an unset stream is omitted: {json}"
1836        );
1837        let scoped: RegisterSchemaBody = serde_json::from_str(
1838            r#"{"source":{"kind":"json_schema","schema":"{}"},"stream":"acme"}"#,
1839        )
1840        .expect("decodes");
1841        assert_eq!(scoped.stream.as_deref(), Some("acme"));
1842    }
1843}
laser_wire/http.rs

laser_wire/
http.rs
