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

laser_wire/
http.rs
