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

laser_wire/
http.rs
