laser_wire/
validate.rs

1use crate::batch::BatchRequest;
2use crate::browse::RegisterSchema;
3use crate::clients::ClientMetadata;
4use crate::codes::AGDX_BATCH_CODE;
5use crate::codes::KV_LEASE_OP_VERSION;
6use crate::error::InvalidError;
7use crate::fork::ForkPut;
8use crate::graph::{GraphEdge, GraphNeighbors, GraphNode, GraphQuery, GraphUpsert, SourceRef};
9use crate::kv::{KvCasFenced, KvLease, KvLeaseRenew, KvRelease, KvScan, KvSet};
10use crate::limits::{
11    MAX_BATCH_OPS, MAX_CLIENT_METADATA, MAX_FRAME_BYTES, MAX_GRAPH_LABEL_BYTES,
12    MAX_GRAPH_NODE_LABELS, MAX_GRAPH_RESULT_ELEMENTS, MAX_GRAPH_TRAVERSE_DEPTH,
13    MAX_HOLDER_ID_BYTES, MAX_KEY_BYTES, MAX_LEASE_TTL_MICROS, MAX_MEMORY_BODY_BYTES,
14    MAX_MEMORY_ID_BYTES, MAX_METADATA_ENTRIES, MAX_METADATA_KEY_BYTES, MAX_PAGE_SIZE,
15    MAX_QUERY_NAME_BYTES, MAX_SCAN_LIMIT, MAX_SCHEMA_NAME_BYTES, MAX_SOURCE_REF_BYTES,
16    MAX_TEXT_QUERY_BYTES, MAX_VALUE_BYTES, MAX_VECTOR_DIMENSIONS, MIN_LEASE_TTL_MICROS,
17};
18use crate::memory::MemoryRecord;
19use crate::query::{TextQuery, Value, VectorQuery};
20
21/// A capped request type that enforces its own size and shape limits, so the
22/// cap logic lives once in the wire crate and every port and both servers get
23/// the identical check by construction rather than each remembering to compare
24/// against [`crate::limits`]. The SDK calls it before encoding, the servers call
25/// it after decode and before execution.
26pub trait Validate {
27    /// Reject a request that violates a pinned cap or a structural rule.
28    fn validate(&self) -> Result<(), InvalidError>;
29}
30
31/// The shared rule for caller-chosen names that flow into matching, filtering,
32/// or storage identifiers: non-empty, within `cap` bytes, and made only of
33/// ASCII letters, digits, `-`, `_`, and `.`. A strict safelist, not just a
34/// length bound, because these names get inlined into queries, filters, and
35/// rendered views.
36pub(crate) fn validate_safelisted_name(
37    label: &str,
38    value: &str,
39    cap: usize,
40) -> Result<(), InvalidError> {
41    if value.is_empty() {
42        return Err(InvalidError::new(format!("{label} must not be empty")));
43    }
44    if value.len() > cap {
45        return Err(InvalidError::new(format!(
46            "{label} is {}B, exceeds cap {cap}B",
47            value.len()
48        )));
49    }
50    if let Some(bad) = value
51        .bytes()
52        .find(|byte| !matches!(byte, b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' | b'.'))
53    {
54        return Err(InvalidError::new(format!(
55            "{label} has a disallowed byte {bad:#04x}: allowed are ASCII letters, digits, '-', '_', '.'"
56        )));
57    }
58    Ok(())
59}
60
61/// The shared rule for caller-chosen text that names a resource without a
62/// safelist: non-empty, within `cap` bytes, and free of control characters.
63pub(crate) fn validate_bounded_text(
64    label: &str,
65    value: &str,
66    cap: usize,
67) -> Result<(), InvalidError> {
68    if value.is_empty() {
69        return Err(InvalidError::new(format!("{label} must not be empty")));
70    }
71    if value.len() > cap {
72        return Err(InvalidError::new(format!(
73            "{label} is {}B, exceeds cap {cap}B",
74            value.len()
75        )));
76    }
77    if value.chars().any(char::is_control) {
78        return Err(InvalidError::new(format!(
79            "{label} contains a control character"
80        )));
81    }
82    Ok(())
83}
84
85/// The shared rule for stored text that keeps its existing caps: no NUL, which
86/// no text column of a relational store can hold.
87pub(crate) fn validate_stored_text(label: &str, value: &str) -> Result<(), InvalidError> {
88    if value.contains('\0') {
89        return Err(InvalidError::new(format!(
90            "{label} contains a NUL character"
91        )));
92    }
93    Ok(())
94}
95
96/// Whether JSON text nests no deeper than `max_depth`, scanned before parsing.
97pub(crate) fn json_depth_within(payload: &[u8], max_depth: usize) -> bool {
98    let mut depth = 0usize;
99    let mut index = 0;
100    while index < payload.len() {
101        match payload[index] {
102            b'"' => {
103                index += 1;
104                loop {
105                    let Some(found) = memchr::memchr2(b'"', b'\\', &payload[index..]) else {
106                        return true;
107                    };
108                    index += found;
109                    if payload[index] == b'"' {
110                        break;
111                    }
112                    index += 2;
113                    if index >= payload.len() {
114                        return true;
115                    }
116                }
117            }
118            b'[' | b'{' => {
119                depth += 1;
120                if depth > max_depth {
121                    return false;
122                }
123            }
124            b']' | b'}' => depth = depth.saturating_sub(1),
125            _ => {}
126        }
127        index += 1;
128    }
129    true
130}
131
132impl Validate for BatchRequest {
133    fn validate(&self) -> Result<(), InvalidError> {
134        if self.ops.len() > MAX_BATCH_OPS {
135            return Err(InvalidError::new(format!(
136                "batch has {} ops, exceeds cap {MAX_BATCH_OPS}",
137                self.ops.len()
138            )));
139        }
140        let mut total = 0usize;
141        for item in &self.ops {
142            // A batch never nests a batch: an op that is itself a batch would let
143            // one request fan out without bound past the op cap.
144            if item.code == AGDX_BATCH_CODE {
145                return Err(InvalidError::new(
146                    "a batch may not contain a batch op".to_owned(),
147                ));
148            }
149            if item.payload.len() > MAX_VALUE_BYTES {
150                return Err(InvalidError::new(format!(
151                    "batch op payload is {}B, exceeds cap {MAX_VALUE_BYTES}B",
152                    item.payload.len()
153                )));
154            }
155            total = total.saturating_add(item.payload.len());
156        }
157        if total > MAX_FRAME_BYTES {
158            return Err(InvalidError::new(format!(
159                "batch total payload is {total}B, exceeds cap {MAX_FRAME_BYTES}B"
160            )));
161        }
162        Ok(())
163    }
164}
165
166impl Validate for KvSet {
167    fn validate(&self) -> Result<(), InvalidError> {
168        crate::kv::validate_namespace(&self.namespace)?;
169        crate::kv::resolve_expiry(self.expires_at_micros, self.ttl_micros, 0)?;
170        if self.key.is_empty() {
171            return Err(InvalidError::new("key-value key is empty".to_owned()));
172        }
173        if self.key.len() > MAX_KEY_BYTES {
174            return Err(InvalidError::new(format!(
175                "key is {}B, exceeds cap {MAX_KEY_BYTES}B",
176                self.key.len()
177            )));
178        }
179        if self.value.len() > MAX_VALUE_BYTES {
180            return Err(InvalidError::new(format!(
181                "value is {}B, exceeds cap {MAX_VALUE_BYTES}B",
182                self.value.len()
183            )));
184        }
185        Ok(())
186    }
187}
188
189impl Validate for KvScan {
190    fn validate(&self) -> Result<(), InvalidError> {
191        crate::kv::validate_namespace(&self.namespace)?;
192        if self.results as usize > MAX_SCAN_LIMIT {
193            return Err(InvalidError::new(format!(
194                "scan page size {} exceeds cap {MAX_SCAN_LIMIT}",
195                self.results
196            )));
197        }
198        Ok(())
199    }
200}
201
202// The fenced-lease family shares its structural rules, so they live once: the
203// exact op version (any other `v` is a typed rejection, the fail-closed half
204// of the in-place reshape), a bounded key, a bounded non-empty holder.
205fn validate_lease_shape(
206    v: u32,
207    namespace: &str,
208    key: &[u8],
209    holder_id: &str,
210) -> Result<(), InvalidError> {
211    if v != KV_LEASE_OP_VERSION {
212        return Err(InvalidError::new(format!(
213            "unsupported fenced-lease op version (expected {KV_LEASE_OP_VERSION}, got {v})"
214        )));
215    }
216    crate::kv::validate_namespace(namespace)?;
217    validate_kv_key("lease key", key)?;
218    if holder_id.is_empty() {
219        return Err(InvalidError::new("lease holder id must not be empty"));
220    }
221    if holder_id.len() > MAX_HOLDER_ID_BYTES {
222        return Err(InvalidError::new(format!(
223            "lease holder id is {}B, exceeds cap {MAX_HOLDER_ID_BYTES}B",
224            holder_id.len()
225        )));
226    }
227    if holder_id.chars().any(char::is_control) {
228        return Err(InvalidError::new(
229            "lease holder id contains a control character",
230        ));
231    }
232    Ok(())
233}
234
235// A requested lease lifetime is bounded on both sides, and the bound is the
236// contract's, not each backend's: a floor so a grant survives one round trip
237// plus the holder's renewal cadence, a ceiling so a crashed holder's ownership
238// always expires. The store may still grant less than the request
239// (`granted_ttl_micros`), so this rejects only what no deployment would serve,
240// and it rejects it locally instead of after a round trip.
241fn validate_lease_ttl(ttl_micros: u64) -> Result<(), InvalidError> {
242    if !(MIN_LEASE_TTL_MICROS..=MAX_LEASE_TTL_MICROS).contains(&ttl_micros) {
243        return Err(InvalidError::new(format!(
244            "lease ttl is {ttl_micros}\u{b5}s, outside the supported range \
245             {MIN_LEASE_TTL_MICROS}..={MAX_LEASE_TTL_MICROS}\u{b5}s"
246        )));
247    }
248    Ok(())
249}
250
251fn validate_kv_key(label: &str, key: &[u8]) -> Result<(), InvalidError> {
252    if key.is_empty() {
253        return Err(InvalidError::new(format!("{label} is empty")));
254    }
255    if key.len() > MAX_KEY_BYTES {
256        return Err(InvalidError::new(format!(
257            "{label} is {}B, exceeds cap {MAX_KEY_BYTES}B",
258            key.len()
259        )));
260    }
261    Ok(())
262}
263
264impl Validate for KvLease {
265    fn validate(&self) -> Result<(), InvalidError> {
266        validate_lease_shape(self.v, &self.namespace, &self.key, &self.holder_id)?;
267        validate_lease_ttl(self.lease_ttl_micros)?;
268        Ok(())
269    }
270}
271
272impl Validate for KvLeaseRenew {
273    fn validate(&self) -> Result<(), InvalidError> {
274        validate_lease_shape(self.v, &self.namespace, &self.key, &self.holder_id)?;
275        validate_lease_ttl(self.lease_ttl_micros)?;
276        if self.lease_token == 0 {
277            return Err(InvalidError::new("lease token must not be zero"));
278        }
279        Ok(())
280    }
281}
282
283impl Validate for KvRelease {
284    fn validate(&self) -> Result<(), InvalidError> {
285        validate_lease_shape(self.v, &self.namespace, &self.key, &self.holder_id)?;
286        if self.lease_token == 0 {
287            return Err(InvalidError::new("lease token must not be zero"));
288        }
289        Ok(())
290    }
291}
292
293impl Validate for KvCasFenced {
294    fn validate(&self) -> Result<(), InvalidError> {
295        if self.v != KV_LEASE_OP_VERSION {
296            return Err(InvalidError::new(format!(
297                "unsupported fenced-lease op version (expected {KV_LEASE_OP_VERSION}, got {})",
298                self.v
299            )));
300        }
301        crate::kv::validate_namespace(&self.namespace)?;
302        crate::kv::validate_namespace(&self.fence_namespace)?;
303        crate::kv::resolve_expiry(self.expires_at_micros, self.ttl_micros, 0)?;
304        validate_kv_key("key-value key", &self.key)?;
305        validate_kv_key("fence key", &self.fence_key)?;
306        if self.value.len() > MAX_VALUE_BYTES {
307            return Err(InvalidError::new(format!(
308                "value is {}B, exceeds cap {MAX_VALUE_BYTES}B",
309                self.value.len()
310            )));
311        }
312        // Zero is the reserved "unversioned" sentinel, never a granted token.
313        if self.fence_token == 0 {
314            return Err(InvalidError::new("fence token must not be zero"));
315        }
316        Ok(())
317    }
318}
319
320impl Validate for GraphQuery {
321    fn validate(&self) -> Result<(), InvalidError> {
322        crate::graph::validate_graph_name(&self.graph)?;
323        let mut depth = 0_u32;
324        for hop in &self.traverse {
325            if hop.max == 0 || hop.max > MAX_GRAPH_TRAVERSE_DEPTH {
326                return Err(InvalidError::new(format!(
327                    "graph hop max {} must be between 1 and {MAX_GRAPH_TRAVERSE_DEPTH}",
328                    hop.max
329                )));
330            }
331            depth = depth.saturating_add(hop.max);
332        }
333        if depth > MAX_GRAPH_TRAVERSE_DEPTH {
334            return Err(InvalidError::new(format!(
335                "graph traversal depth {depth} exceeds cap {MAX_GRAPH_TRAVERSE_DEPTH}"
336            )));
337        }
338        validate_graph_limit(self.limit)
339    }
340}
341
342impl Validate for GraphNeighbors {
343    fn validate(&self) -> Result<(), InvalidError> {
344        crate::graph::validate_graph_name(&self.graph)?;
345        if self.depth > MAX_GRAPH_TRAVERSE_DEPTH {
346            return Err(InvalidError::new(format!(
347                "graph neighbor depth {} exceeds cap {MAX_GRAPH_TRAVERSE_DEPTH}",
348                self.depth
349            )));
350        }
351        validate_graph_limit(self.limit)
352    }
353}
354
355fn validate_graph_limit(limit: u32) -> Result<(), InvalidError> {
356    if limit as usize > MAX_GRAPH_RESULT_ELEMENTS {
357        return Err(InvalidError::new(format!(
358            "graph result limit {limit} exceeds cap {MAX_GRAPH_RESULT_ELEMENTS}"
359        )));
360    }
361    Ok(())
362}
363
364impl Validate for SourceRef {
365    fn validate(&self) -> Result<(), InvalidError> {
366        if let SourceRef::Message {
367            conversation: Some(conversation),
368            ..
369        } = self
370        {
371            validate_stored_text("source conversation", conversation)?;
372        }
373        let size = match self {
374            // A log pointer is fixed-width numerics plus an optional
375            // conversation id, so only the id needs a bound.
376            SourceRef::Message { conversation, .. } => conversation.as_ref().map_or(0, String::len),
377            SourceRef::Kv { namespace, key } => namespace.len() + key.len(),
378            SourceRef::Memory { id } => id.len(),
379        };
380        if size > MAX_SOURCE_REF_BYTES {
381            return Err(InvalidError::new(format!(
382                "source reference is {size}B, exceeds cap {MAX_SOURCE_REF_BYTES}B"
383            )));
384        }
385        Ok(())
386    }
387}
388
389// Attribute lists ride on both nodes and edges, so their bound lives once.
390fn validate_attrs(label: &str, attrs: &[(String, Value)]) -> Result<(), InvalidError> {
391    if attrs.len() > MAX_METADATA_ENTRIES {
392        return Err(InvalidError::new(format!(
393            "{label} carries {} attributes, exceeds cap {MAX_METADATA_ENTRIES}",
394            attrs.len()
395        )));
396    }
397    for (key, _) in attrs {
398        if key.len() > MAX_METADATA_KEY_BYTES {
399            return Err(InvalidError::new(format!(
400                "{label} attribute name is {}B, exceeds cap {MAX_METADATA_KEY_BYTES}B",
401                key.len()
402            )));
403        }
404    }
405    Ok(())
406}
407
408impl Validate for GraphNode {
409    fn validate(&self) -> Result<(), InvalidError> {
410        if self.labels.len() > MAX_GRAPH_NODE_LABELS {
411            return Err(InvalidError::new(format!(
412                "graph node carries {} labels, exceeds cap {MAX_GRAPH_NODE_LABELS}",
413                self.labels.len()
414            )));
415        }
416        for label in &self.labels {
417            validate_bounded_text("graph node label", label, MAX_GRAPH_LABEL_BYTES)?;
418        }
419        validate_attrs("graph node", &self.attrs)?;
420        if let Some(source) = &self.source {
421            source.validate()?;
422        }
423        Ok(())
424    }
425}
426
427impl Validate for GraphEdge {
428    fn validate(&self) -> Result<(), InvalidError> {
429        validate_bounded_text("graph edge type", &self.edge_type, MAX_GRAPH_LABEL_BYTES)?;
430        validate_attrs("graph edge", &self.attrs)?;
431        if let Some(source) = &self.source {
432            source.validate()?;
433        }
434        Ok(())
435    }
436}
437
438impl Validate for GraphUpsert {
439    fn validate(&self) -> Result<(), InvalidError> {
440        crate::graph::validate_graph_name(&self.graph)?;
441        let elements = self.nodes.len() + self.edges.len();
442        if elements > MAX_GRAPH_RESULT_ELEMENTS {
443            return Err(InvalidError::new(format!(
444                "graph upsert carries {elements} elements, exceeds cap {MAX_GRAPH_RESULT_ELEMENTS}"
445            )));
446        }
447        for node in &self.nodes {
448            node.validate()?;
449        }
450        for edge in &self.edges {
451            edge.validate()?;
452        }
453        Ok(())
454    }
455}
456
457impl Validate for ForkPut {
458    fn validate(&self) -> Result<(), InvalidError> {
459        crate::fork::validate_fork_id(&self.fork_id)?;
460        validate_stored_text("fork table", &self.table)?;
461        validate_stored_text("fork projection id", &self.projection_id)?;
462        for (name, value) in &self.fields {
463            validate_stored_text("fork field name", name)?;
464            validate_stored_text("fork field value", value)?;
465        }
466        for (name, value) in &self.metadata {
467            validate_stored_text("fork metadata name", name)?;
468            validate_stored_text("fork metadata value", value)?;
469        }
470        if let Some(embedding) = &self.embedding {
471            validate_stored_text("fork embedding", embedding)?;
472        }
473        Ok(())
474    }
475}
476
477impl Validate for TextQuery {
478    fn validate(&self) -> Result<(), InvalidError> {
479        if self.query.trim().is_empty() || self.query.len() > MAX_TEXT_QUERY_BYTES {
480            return Err(InvalidError::new(format!(
481                "text query is {}B, expected 1..={MAX_TEXT_QUERY_BYTES}B",
482                self.query.len()
483            )));
484        }
485        if self.query.chars().any(char::is_control) {
486            return Err(InvalidError::new("text query contains a control character"));
487        }
488        if let Some(field) = &self.field
489            && (field.is_empty()
490                || field.len() > MAX_QUERY_NAME_BYTES
491                || field.chars().any(char::is_control))
492        {
493            return Err(InvalidError::new("text query field is invalid"));
494        }
495        Ok(())
496    }
497}
498
499impl Validate for VectorQuery {
500    fn validate(&self) -> Result<(), InvalidError> {
501        if self.field.is_empty()
502            || self.field.len() > MAX_QUERY_NAME_BYTES
503            || self.field.chars().any(char::is_control)
504        {
505            return Err(InvalidError::new("vector field is invalid"));
506        }
507        if self.embedding.is_empty() || self.embedding.len() > MAX_VECTOR_DIMENSIONS {
508            return Err(InvalidError::new(format!(
509                "vector dimensions must be in 1..={MAX_VECTOR_DIMENSIONS}"
510            )));
511        }
512        if self
513            .embedding
514            .iter()
515            .any(|value| !value.is_finite() || (*value == 0.0 && value.is_sign_negative()))
516        {
517            return Err(InvalidError::new(
518                "vector values must be finite canonical floats",
519            ));
520        }
521        if self.top_k == 0 || self.top_k > MAX_PAGE_SIZE as u32 {
522            return Err(InvalidError::new(format!(
523                "vector top_k {} is outside 1..={MAX_PAGE_SIZE}",
524                self.top_k
525            )));
526        }
527        Ok(())
528    }
529}
530
531impl Validate for MemoryRecord {
532    fn validate(&self) -> Result<(), InvalidError> {
533        match self {
534            MemoryRecord::Item { id, kind, .. } => {
535                validate_bounded_text("memory item id", id, MAX_MEMORY_ID_BYTES)?;
536                validate_bounded_text("memory item kind", kind, MAX_MEMORY_ID_BYTES)?;
537            }
538            MemoryRecord::Forget { target, .. } | MemoryRecord::Feedback { target, .. } => {
539                validate_bounded_text("memory target", target, MAX_MEMORY_ID_BYTES)?;
540            }
541        }
542        if let MemoryRecord::Item { body, .. } = self
543            && body.len() > MAX_MEMORY_BODY_BYTES
544        {
545            return Err(InvalidError::new(format!(
546                "memory body is {}B, exceeds cap {MAX_MEMORY_BODY_BYTES}B",
547                body.len()
548            )));
549        }
550        if let MemoryRecord::Forget {
551            conversation: Some(conversation),
552            ..
553        }
554        | MemoryRecord::Feedback {
555            conversation: Some(conversation),
556            ..
557        } = self
558        {
559            conversation
560                .parse::<crate::agent::ConversationId>()
561                .map_err(|_| InvalidError::new("memory record conversation is not a valid id"))?;
562        }
563        Ok(())
564    }
565}
566
567impl Validate for RegisterSchema {
568    fn validate(&self) -> Result<(), InvalidError> {
569        if let Some(name) = &self.name {
570            validate_bounded_text("schema name", name, MAX_SCHEMA_NAME_BYTES)?;
571        }
572        self.source.validate()
573    }
574}
575
576impl Validate for ClientMetadata {
577    fn validate(&self) -> Result<(), InvalidError> {
578        let size = self.metadata.as_ref().map_or(0, Vec::len);
579        if size > MAX_CLIENT_METADATA {
580            return Err(InvalidError::new(format!(
581                "client metadata is {size}B, exceeds cap {MAX_CLIENT_METADATA}B"
582            )));
583        }
584        Ok(())
585    }
586}
587
588#[cfg(test)]
589mod tests {
590    use super::*;
591    use crate::batch::BatchItem;
592    use crate::codes::{
593        AGDX_KV_GET_CODE, BATCH_OP_VERSION, KV_LEASE_OP_VERSION, KV_OP_VERSION, QUERY_OP_VERSION,
594    };
595    use crate::control::SchemaSource;
596    use crate::limits::{MAX_GRAPH_LABEL_BYTES, MAX_MEMORY_ID_BYTES, MAX_SCHEMA_NAME_BYTES};
597    use std::collections::BTreeMap;
598
599    #[test]
600    fn given_an_invalid_graph_label_or_edge_type_when_upserted_then_should_refuse_it() {
601        let at_cap = "l".repeat(MAX_GRAPH_LABEL_BYTES);
602        let alice = node_with(vec![at_cap.clone()], Vec::new());
603        let acme = node_with(vec!["Company".to_owned()], Vec::new());
604        alice.validate().expect("a node label at the cap is valid");
605        GraphEdge::relate(&alice, at_cap, &acme)
606            .validate()
607            .expect("an edge type at the cap is valid");
608        let long = "l".repeat(MAX_GRAPH_LABEL_BYTES + 1);
609        for invalid in [String::new(), long, "Per\nson".to_owned()] {
610            let node = node_with(vec![invalid.clone()], Vec::new());
611            assert!(node.validate().is_err(), "{invalid:?}");
612            assert!(
613                GraphEdge::relate(&alice, invalid.clone(), &acme)
614                    .validate()
615                    .is_err()
616            );
617            let upsert = GraphUpsert {
618                v: 1,
619                graph: "knowledge".to_owned(),
620                session: None,
621                nodes: vec![node],
622                edges: Vec::new(),
623            };
624            assert!(upsert.validate().is_err());
625        }
626    }
627
628    #[test]
629    fn given_an_invalid_memory_id_kind_or_target_when_validated_then_should_refuse_it() {
630        let item = |id: &str, kind: &str| MemoryRecord::Item {
631            origin: None,
632            producer: None,
633            id: id.to_owned(),
634            kind: kind.to_owned(),
635            body: b"auth is slow".to_vec(),
636        };
637        let at_cap = "m".repeat(MAX_MEMORY_ID_BYTES);
638        item(&at_cap, &at_cap)
639            .validate()
640            .expect("an id and a kind at the cap are valid");
641        for invalid in [
642            String::new(),
643            "m".repeat(MAX_MEMORY_ID_BYTES + 1),
644            "fa\u{0}ct".to_owned(),
645        ] {
646            assert!(item(&invalid, "fact").validate().is_err(), "{invalid:?}");
647            assert!(
648                item("01KWM3K3XEP3NP5TN850J17YBP", &invalid)
649                    .validate()
650                    .is_err()
651            );
652            assert!(
653                MemoryRecord::Forget {
654                    target: invalid.clone(),
655                    conversation: None,
656                }
657                .validate()
658                .is_err()
659            );
660            assert!(
661                MemoryRecord::Feedback {
662                    target: invalid,
663                    weight: 1.0,
664                    conversation: None,
665                }
666                .validate()
667                .is_err()
668            );
669        }
670    }
671
672    #[test]
673    fn given_a_fork_put_with_nul_text_when_validated_then_should_refuse_it() {
674        let valid = ForkPut {
675            v: crate::codes::FORK_OP_VERSION,
676            fork_id: "draft".to_owned(),
677            table: "orders".to_owned(),
678            partition_id: 0,
679            offset: 1,
680            projection_id: "orders.v1".to_owned(),
681            projection_version: 1,
682            fields: BTreeMap::from([("status".to_owned(), "ok".to_owned())]),
683            metadata: BTreeMap::from([("role".to_owned(), "user".to_owned())]),
684            payload: None,
685            embedding: None,
686            tombstone: false,
687        };
688        valid.validate().expect("valid fork put");
689        let mut bad_field = valid.clone();
690        bad_field
691            .fields
692            .insert("status".to_owned(), "o\u{0}k".to_owned());
693        let mut bad_metadata = valid.clone();
694        bad_metadata
695            .metadata
696            .insert("ro\u{0}le".to_owned(), "user".to_owned());
697        let mut bad_table = valid.clone();
698        bad_table.table = "ord\u{0}ers".to_owned();
699        let mut bad_projection = valid.clone();
700        bad_projection.projection_id = "orders\u{0}".to_owned();
701        let mut bad_embedding = valid.clone();
702        bad_embedding.embedding = Some("[1.0\u{0}]".to_owned());
703        let mut bad_fork = valid;
704        bad_fork.fork_id = "dr aft".to_owned();
705        for request in [
706            bad_field,
707            bad_metadata,
708            bad_table,
709            bad_projection,
710            bad_embedding,
711            bad_fork,
712        ] {
713            assert!(request.validate().is_err(), "{request:?}");
714        }
715    }
716
717    #[test]
718    fn given_a_message_source_with_a_nul_conversation_when_validated_then_should_refuse_it() {
719        let source = |conversation: &str| SourceRef::Message {
720            stream: 1,
721            topic: 2,
722            partition: 0,
723            offset: 3,
724            generation: None,
725            conversation: Some(conversation.to_owned()),
726        };
727        source("01KWM3K3XEP3NP5TN850J17YBP")
728            .validate()
729            .expect("a plain conversation is valid");
730        assert!(source("01KWM3K3\u{0}").validate().is_err());
731    }
732
733    #[test]
734    fn given_a_lease_holder_with_a_control_character_when_validated_then_should_refuse_it() {
735        let holder = "node\n1".to_owned();
736        let lease = KvLease {
737            v: KV_LEASE_OP_VERSION,
738            namespace: "connectors.coordination".to_owned(),
739            key: b"source-owner".to_vec(),
740            lease_ttl_micros: 30_000_000,
741            holder_id: holder.clone(),
742            subject_user_id: None,
743        };
744        assert!(lease.validate().is_err());
745        let renew = KvLeaseRenew {
746            v: KV_LEASE_OP_VERSION,
747            namespace: "connectors.coordination".to_owned(),
748            key: b"source-owner".to_vec(),
749            holder_id: holder.clone(),
750            subject_user_id: None,
751            lease_token: 7,
752            lease_ttl_micros: 30_000_000,
753        };
754        assert!(renew.validate().is_err());
755        let release = KvRelease {
756            v: KV_LEASE_OP_VERSION,
757            namespace: "connectors.coordination".to_owned(),
758            key: b"source-owner".to_vec(),
759            lease_token: 7,
760            holder_id: holder,
761        };
762        assert!(release.validate().is_err());
763    }
764
765    #[test]
766    fn given_an_invalid_schema_name_when_a_registration_is_validated_then_should_refuse_it() {
767        let register = |name: Option<String>| RegisterSchema {
768            v: QUERY_OP_VERSION,
769            source: SchemaSource::JsonSchema {
770                schema: "{}".to_owned(),
771            },
772            name,
773            version: None,
774            stream: Some("fleet".to_owned()),
775        };
776        register(None)
777            .validate()
778            .expect("a registration without a name is valid");
779        register(Some("reading".to_owned()))
780            .validate()
781            .expect("a short schema name is valid");
782        assert!(register(Some(String::new())).validate().is_err());
783        assert!(
784            register(Some("s".repeat(MAX_SCHEMA_NAME_BYTES + 1)))
785                .validate()
786                .is_err()
787        );
788        assert!(register(Some("read\ting".to_owned())).validate().is_err());
789    }
790
791    #[test]
792    fn given_a_batch_over_the_op_cap_when_validated_then_should_reject() {
793        let request = BatchRequest {
794            v: BATCH_OP_VERSION,
795            ops: (0..MAX_BATCH_OPS + 1)
796                .map(|_| BatchItem {
797                    code: AGDX_KV_GET_CODE,
798                    payload: Vec::new(),
799                })
800                .collect(),
801        };
802        assert!(request.validate().is_err());
803    }
804
805    #[test]
806    fn given_a_batch_nesting_a_batch_when_validated_then_should_reject() {
807        let request = BatchRequest {
808            v: BATCH_OP_VERSION,
809            ops: vec![BatchItem {
810                code: AGDX_BATCH_CODE,
811                payload: Vec::new(),
812            }],
813        };
814        assert!(request.validate().is_err());
815    }
816
817    #[test]
818    fn given_namespaces_when_validated_then_should_enforce_bounds() {
819        use crate::kv::validate_namespace;
820        use crate::limits::MAX_NAMESPACE_BYTES;
821        assert!(validate_namespace("default").is_ok());
822        assert!(validate_namespace("agent-abc/session").is_ok(), "hierarchy");
823        assert!(validate_namespace("").is_err(), "empty");
824        assert!(validate_namespace("bad\nns").is_err(), "control byte");
825        assert!(validate_namespace(&"n".repeat(MAX_NAMESPACE_BYTES)).is_ok());
826        assert!(validate_namespace(&"n".repeat(MAX_NAMESPACE_BYTES + 1)).is_err());
827    }
828
829    #[test]
830    fn given_graph_names_when_validated_then_should_enforce_bounds() {
831        use crate::graph::validate_graph_name;
832        use crate::limits::MAX_GRAPH_NAME_BYTES;
833        assert!(validate_graph_name("knowledge").is_ok());
834        assert!(validate_graph_name("").is_err(), "empty");
835        assert!(validate_graph_name("bad\tname").is_err(), "control byte");
836        assert!(validate_graph_name(&"g".repeat(MAX_GRAPH_NAME_BYTES + 1)).is_err());
837    }
838
839    #[test]
840    fn given_a_lease_request_when_validated_then_should_enforce_version_holder_and_ttl() {
841        use crate::codes::KV_LEASE_OP_VERSION;
842        use crate::kv::KvLease;
843        use crate::limits::MAX_HOLDER_ID_BYTES;
844        let valid = KvLease {
845            v: KV_LEASE_OP_VERSION,
846            namespace: "connectors.coordination".to_owned(),
847            key: b"source-owner".to_vec(),
848            lease_ttl_micros: 30_000_000,
849            holder_id: "node-1".to_owned(),
850            subject_user_id: Some(42),
851        };
852        assert!(valid.validate().is_ok());
853        let wrong_version = KvLease {
854            v: KV_LEASE_OP_VERSION + 1,
855            ..valid.clone()
856        };
857        assert!(
858            wrong_version.validate().is_err(),
859            "an unsupported version is rejected typed"
860        );
861        let empty_holder = KvLease {
862            holder_id: String::new(),
863            ..valid.clone()
864        };
865        assert!(empty_holder.validate().is_err());
866        let long_holder = KvLease {
867            holder_id: "h".repeat(MAX_HOLDER_ID_BYTES + 1),
868            ..valid.clone()
869        };
870        assert!(long_holder.validate().is_err());
871        let zero_ttl = KvLease {
872            lease_ttl_micros: 0,
873            ..valid
874        };
875        assert!(zero_ttl.validate().is_err());
876    }
877
878    // The TTL range is the contract's, so the same rejection happens on the
879    // acquire and the renewal, before any round trip, on every tier.
880    #[test]
881    fn given_out_of_range_lease_ttls_when_validated_then_should_reject_both_ends() {
882        use crate::codes::KV_LEASE_OP_VERSION;
883        use crate::kv::{KvLease, KvLeaseRenew};
884        use crate::limits::{MAX_LEASE_TTL_MICROS, MIN_LEASE_TTL_MICROS};
885        let acquire = KvLease {
886            v: KV_LEASE_OP_VERSION,
887            namespace: "connectors.coordination".to_owned(),
888            key: b"source-owner".to_vec(),
889            lease_ttl_micros: MIN_LEASE_TTL_MICROS,
890            holder_id: "node-1".to_owned(),
891            subject_user_id: None,
892        };
893        assert!(acquire.validate().is_ok(), "the floor itself is grantable");
894        assert!(
895            KvLease {
896                lease_ttl_micros: MAX_LEASE_TTL_MICROS,
897                ..acquire.clone()
898            }
899            .validate()
900            .is_ok(),
901            "the ceiling itself is grantable"
902        );
903        assert!(
904            KvLease {
905                lease_ttl_micros: MIN_LEASE_TTL_MICROS - 1,
906                ..acquire.clone()
907            }
908            .validate()
909            .is_err(),
910            "a lease too short to renew is rejected locally"
911        );
912        assert!(
913            KvLease {
914                lease_ttl_micros: MAX_LEASE_TTL_MICROS + 1,
915                ..acquire
916            }
917            .validate()
918            .is_err(),
919            "an unbounded ownership window is rejected locally"
920        );
921        let renew = KvLeaseRenew {
922            v: KV_LEASE_OP_VERSION,
923            namespace: "connectors.coordination".to_owned(),
924            key: b"source-owner".to_vec(),
925            holder_id: "node-1".to_owned(),
926            subject_user_id: None,
927            lease_token: 7,
928            lease_ttl_micros: MAX_LEASE_TTL_MICROS + 1,
929        };
930        assert!(renew.validate().is_err(), "renewal shares the same range");
931        assert!(
932            KvLeaseRenew {
933                lease_ttl_micros: MIN_LEASE_TTL_MICROS - 1,
934                ..renew
935            }
936            .validate()
937            .is_err()
938        );
939    }
940
941    #[test]
942    fn given_renew_and_release_requests_when_validated_then_should_require_a_nonzero_token() {
943        use crate::codes::KV_LEASE_OP_VERSION;
944        use crate::kv::{KvLeaseRenew, KvRelease};
945        let renew = KvLeaseRenew {
946            v: KV_LEASE_OP_VERSION,
947            namespace: "connectors.coordination".to_owned(),
948            key: b"source-owner".to_vec(),
949            holder_id: "node-1".to_owned(),
950            subject_user_id: None,
951            lease_token: 7,
952            lease_ttl_micros: 30_000_000,
953        };
954        assert!(renew.validate().is_ok());
955        let zero = KvLeaseRenew {
956            lease_token: 0,
957            ..renew
958        };
959        assert!(zero.validate().is_err());
960        let release = KvRelease {
961            v: KV_LEASE_OP_VERSION,
962            namespace: "connectors.coordination".to_owned(),
963            key: b"source-owner".to_vec(),
964            lease_token: 7,
965            holder_id: "node-1".to_owned(),
966        };
967        assert!(release.validate().is_ok());
968        let zero = KvRelease {
969            lease_token: 0,
970            ..release
971        };
972        assert!(zero.validate().is_err());
973    }
974
975    #[test]
976    fn given_a_fenced_cas_when_validated_then_should_enforce_both_namespaces_and_the_token() {
977        use crate::codes::KV_LEASE_OP_VERSION;
978        use crate::kv::{CasExpect, KvCasFenced};
979        let valid = KvCasFenced {
980            v: KV_LEASE_OP_VERSION,
981            namespace: "connectors.state".to_owned(),
982            session: None,
983            key: b"source_state/pg".to_vec(),
984            value: b"cursor".to_vec(),
985            expires_at_micros: None,
986            ttl_micros: None,
987            expect: CasExpect::Absent,
988            fence_namespace: "connectors.coordination".to_owned(),
989            fence_key: b"source-owner".to_vec(),
990            fence_token: 8,
991        };
992        assert!(valid.validate().is_ok());
993        let wrong_version = KvCasFenced {
994            v: KV_LEASE_OP_VERSION + 1,
995            ..valid.clone()
996        };
997        assert!(
998            wrong_version.validate().is_err(),
999            "an unsupported version is rejected typed"
1000        );
1001        let bad_fence_namespace = KvCasFenced {
1002            fence_namespace: "bad\nns".to_owned(),
1003            ..valid.clone()
1004        };
1005        assert!(bad_fence_namespace.validate().is_err());
1006        let zero_token = KvCasFenced {
1007            fence_token: 0,
1008            ..valid
1009        };
1010        assert!(zero_token.validate().is_err());
1011    }
1012
1013    #[test]
1014    fn given_an_oversized_key_when_validated_then_should_reject_and_a_valid_one_passes() {
1015        let over = KvSet {
1016            v: KV_OP_VERSION,
1017            namespace: "ns".to_owned(),
1018            session: None,
1019            key: vec![b'x'; MAX_KEY_BYTES + 1],
1020            value: vec![1, 2, 3],
1021            expires_at_micros: None,
1022            ttl_micros: None,
1023        };
1024        assert!(over.validate().is_err());
1025        let ok = KvSet {
1026            v: KV_OP_VERSION,
1027            namespace: "ns".to_owned(),
1028            session: None,
1029            key: vec![b'x'; 8],
1030            value: vec![1, 2, 3],
1031            expires_at_micros: None,
1032            ttl_micros: None,
1033        };
1034        assert!(ok.validate().is_ok());
1035    }
1036
1037    fn node_with(labels: Vec<String>, attrs: Vec<(String, Value)>) -> GraphNode {
1038        GraphNode {
1039            producer: None,
1040            id: crate::graph::NodeId::from_u128(1),
1041            labels,
1042            attrs,
1043            embedding: None,
1044            source: None,
1045        }
1046    }
1047
1048    #[test]
1049    fn given_a_node_over_the_label_cap_when_validated_then_should_reject() {
1050        let over = node_with(
1051            (0..=MAX_GRAPH_NODE_LABELS)
1052                .map(|index| format!("label-{index}"))
1053                .collect(),
1054            Vec::new(),
1055        );
1056        assert!(over.validate().is_err());
1057        assert!(
1058            node_with(vec!["Person".to_owned()], Vec::new())
1059                .validate()
1060                .is_ok()
1061        );
1062    }
1063
1064    #[test]
1065    fn given_a_node_over_the_attribute_cap_when_validated_then_should_reject() {
1066        let over = node_with(
1067            Vec::new(),
1068            (0..=MAX_METADATA_ENTRIES)
1069                .map(|index| (format!("attr-{index}"), Value::from("x")))
1070                .collect(),
1071        );
1072        assert!(over.validate().is_err());
1073    }
1074
1075    #[test]
1076    fn given_an_oversized_source_reference_when_validated_then_should_reject() {
1077        let over = SourceRef::Kv {
1078            namespace: "n".repeat(MAX_SOURCE_REF_BYTES),
1079            key: "k".repeat(MAX_SOURCE_REF_BYTES),
1080        };
1081        assert!(over.validate().is_err());
1082        assert!(
1083            SourceRef::Memory {
1084                id: "01KWM3K3XEP3NP5TN850J17YBP".to_owned()
1085            }
1086            .validate()
1087            .is_ok()
1088        );
1089    }
1090
1091    #[test]
1092    fn given_an_upsert_over_the_element_cap_when_validated_then_should_reject() {
1093        let over = GraphUpsert {
1094            v: 1,
1095            graph: "knowledge".to_owned(),
1096            session: None,
1097            nodes: (0..=MAX_GRAPH_RESULT_ELEMENTS)
1098                .map(|_| node_with(Vec::new(), Vec::new()))
1099                .collect(),
1100            edges: Vec::new(),
1101        };
1102        assert!(over.validate().is_err());
1103        let ok = GraphUpsert {
1104            v: 1,
1105            graph: "knowledge".to_owned(),
1106            session: None,
1107            nodes: vec![node_with(vec!["Person".to_owned()], Vec::new())],
1108            edges: Vec::new(),
1109        };
1110        assert!(ok.validate().is_ok());
1111    }
1112
1113    #[test]
1114    fn given_an_upsert_carrying_an_over_cap_node_when_validated_then_should_reject() {
1115        let upsert = GraphUpsert {
1116            v: 1,
1117            graph: "knowledge".to_owned(),
1118            session: None,
1119            nodes: vec![node_with(
1120                (0..=MAX_GRAPH_NODE_LABELS)
1121                    .map(|index| format!("label-{index}"))
1122                    .collect(),
1123                Vec::new(),
1124            )],
1125            edges: Vec::new(),
1126        };
1127        assert!(
1128            upsert.validate().is_err(),
1129            "an upsert must enforce its elements' caps, not just its own"
1130        );
1131    }
1132
1133    #[test]
1134    fn given_an_oversized_text_query_when_validated_then_should_reject() {
1135        let over = TextQuery {
1136            field: None,
1137            query: "q".repeat(MAX_TEXT_QUERY_BYTES + 1),
1138        };
1139        assert!(over.validate().is_err());
1140        let ok = TextQuery {
1141            field: None,
1142            query: "auth is slow".to_owned(),
1143        };
1144        assert!(ok.validate().is_ok());
1145    }
1146
1147    #[test]
1148    fn given_an_over_cap_vector_top_k_when_validated_then_should_reject() {
1149        let over = VectorQuery {
1150            field: "embedding".to_owned(),
1151            embedding: vec![0.0; 4],
1152            top_k: MAX_PAGE_SIZE as u32 + 1,
1153        };
1154        assert!(over.validate().is_err());
1155        let ok = VectorQuery {
1156            field: "embedding".to_owned(),
1157            embedding: vec![0.0; 4],
1158            top_k: 10,
1159        };
1160        assert!(ok.validate().is_ok());
1161    }
1162
1163    #[test]
1164    fn given_an_oversized_memory_body_when_validated_then_should_reject() {
1165        let over = MemoryRecord::Item {
1166            origin: None,
1167            producer: None,
1168            id: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1169            kind: "fact".to_owned(),
1170            body: vec![0u8; MAX_MEMORY_BODY_BYTES + 1],
1171        };
1172        assert!(over.validate().is_err());
1173        let ok = MemoryRecord::Item {
1174            origin: None,
1175            producer: None,
1176            id: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1177            kind: "fact".to_owned(),
1178            body: b"auth is slow".to_vec(),
1179        };
1180        assert!(ok.validate().is_ok());
1181        assert!(
1182            MemoryRecord::Forget {
1183                target: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1184                conversation: None,
1185            }
1186            .validate()
1187            .is_ok()
1188        );
1189        assert!(
1190            MemoryRecord::Forget {
1191                target: "01KWM3K3XEP3NP5TN850J17YBP".to_owned(),
1192                conversation: Some("not an id".to_owned()),
1193            }
1194            .validate()
1195            .is_err()
1196        );
1197    }
1198
1199    #[test]
1200    fn given_oversized_client_metadata_when_validated_then_should_reject() {
1201        let over = ClientMetadata {
1202            client_id: 1,
1203            user_id: None,
1204            transport: 1,
1205            address: "127.0.0.1:8090".to_owned(),
1206            consumer_groups_count: 0,
1207            metadata: Some(vec![0u8; MAX_CLIENT_METADATA + 1]),
1208        };
1209        assert!(over.validate().is_err());
1210        let ok = ClientMetadata {
1211            client_id: 1,
1212            user_id: None,
1213            transport: 1,
1214            address: "127.0.0.1:8090".to_owned(),
1215            consumer_groups_count: 0,
1216            metadata: Some(b"card".to_vec()),
1217        };
1218        assert!(ok.validate().is_ok());
1219    }
1220}
laser_wire/validate.rs

laser_wire/
validate.rs
