laser_wire/
graph.rs

1use crate::agent::{IdParseError, crockford_decode, crockford_encode};
2use crate::error::InvalidError;
3use crate::limits::MAX_GRAPH_NAME_BYTES;
4use crate::query::{Consistency, Filter, Value};
5use serde::de::{self, Visitor};
6use serde::{Deserialize, Deserializer, Serialize, Serializer};
7use std::fmt;
8use std::str::FromStr;
9
10crate::agent::wire_id!(
11    /// A graph node's identity. Content-addressed (the hash of the node's label
12    /// and canonical value), so the same entity extracted from different messages
13    /// converges on one node. Minted SDK- or projector-side.
14    NodeId
15);
16crate::agent::wire_id!(
17    /// A graph edge's identity. Content-addressed over its endpoints and type, so
18    /// the same relationship is one edge however many times it is observed.
19    EdgeId
20);
21
22impl NodeId {
23    /// A content-addressed node id: the stable hash of the entity's `label` and
24    /// canonical `value`, so the same entity extracted from different records (or
25    /// upserted by different callers, in any SDK) converges on one node, which is
26    /// what makes a graph rather than disconnected pairs. The one canonical
27    /// [`content_id`](crate::hashing::content_id), so every SDK mints the same id
28    /// from the same segments (pinned by the golden vector below).
29    pub fn content(label: &str, value: &[u8]) -> Self {
30        Self::from_u128(crate::hashing::content_id(&[label.as_bytes(), &[0], value]))
31    }
32}
33
34impl EdgeId {
35    /// A content-addressed edge id over its endpoints and type, so the same
36    /// relationship observed any number of times is one edge. Idempotent upsert
37    /// keys off this id.
38    pub fn content(from: NodeId, edge_type: &str, to: NodeId) -> Self {
39        Self::from_u128(crate::hashing::content_id(&[
40            &from.to_bytes(),
41            &[0],
42            edge_type.as_bytes(),
43            &[0],
44            &to.to_bytes(),
45        ]))
46    }
47}
48
49/// Which way a hop follows edges from the current frontier.
50#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
51#[serde(rename_all = "snake_case")]
52pub enum EdgeDir {
53    /// Outgoing edges (from -> to). The default.
54    #[default]
55    Out,
56    /// Incoming edges (to -> from).
57    In,
58    /// Both directions.
59    Both,
60}
61
62/// What a graph query returns.
63#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
64#[serde(rename_all = "snake_case")]
65pub enum GraphReturn {
66    /// The reachable nodes. The default.
67    #[default]
68    Nodes,
69    /// The traversed edges.
70    Edges,
71    /// The full paths (node and edge id sequences).
72    Paths,
73    /// The triplets (source -> relationship -> target) along the traversal.
74    Triplets,
75}
76
77/// One traversal step: follow edges of an optional type in `dir`, up to `max`
78/// hops at this step.
79#[derive(Clone, Debug, Serialize, Deserialize)]
80pub struct Hop {
81    #[serde(default, skip_serializing_if = "Option::is_none")]
82    pub edge_type: Option<String>,
83    #[serde(default, skip_serializing_if = "EdgeDir::is_out")]
84    pub dir: EdgeDir,
85    pub max: u32,
86}
87
88impl EdgeDir {
89    /// Whether this is the default `Out` direction (omitted on the wire).
90    pub fn is_out(&self) -> bool {
91        matches!(self, EdgeDir::Out)
92    }
93}
94
95/// Where a traversal starts: explicit node ids, the nodes matching a predicate,
96/// or the nodes nearest an embedding (vector-seeded traversal).
97#[derive(Clone, Debug, Serialize, Deserialize)]
98pub enum GraphStart {
99    Ids(Vec<NodeId>),
100    Match(Filter),
101    Nearest { embedding: Vec<f32>, k: usize },
102}
103
104/// A graph traversal: start, hop spec, optional node and edge filters, and what
105/// to return. Reuses the query [`Filter`] predicate language, so there is one
106/// filter grammar across query and graph.
107#[derive(Clone, Debug, Serialize, Deserialize)]
108pub struct GraphQuery {
109    pub v: u32,
110    pub graph: String,
111    pub start: GraphStart,
112    #[serde(default, skip_serializing_if = "Vec::is_empty")]
113    pub traverse: Vec<Hop>,
114    #[serde(default, skip_serializing_if = "Option::is_none")]
115    pub node_filter: Option<Filter>,
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub edge_filter: Option<Filter>,
118    #[serde(default, skip_serializing_if = "GraphReturn::is_nodes")]
119    pub return_: GraphReturn,
120    pub limit: usize,
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub fork: Option<String>,
123    #[serde(default, skip_serializing_if = "Consistency::is_eventual")]
124    pub consistency: Consistency,
125    /// Valid-time "as of" read (epoch micros): keep only edges whose valid-time
126    /// window contains this instant. `None` traverses the current graph.
127    #[serde(default, skip_serializing_if = "Option::is_none")]
128    pub as_of: Option<u64>,
129    /// Restrict the traversal to elements a given conversation asserted (the
130    /// text form of its `gen_ai.conversation.id`), matched against each element's
131    /// [`SourceRef`] conversation. `None` reads the whole graph. The conversation
132    /// lens: "show me only what this conversation put in the graph."
133    #[serde(default, skip_serializing_if = "Option::is_none")]
134    pub conversation: Option<String>,
135}
136
137impl GraphReturn {
138    /// Whether this is the default `Nodes` return (omitted on the wire).
139    pub fn is_nodes(&self) -> bool {
140        matches!(self, GraphReturn::Nodes)
141    }
142}
143
144/// A one-hop neighbor read: the cheap, common traversal. `depth` follows the same
145/// hop repeatedly.
146#[derive(Clone, Debug, Serialize, Deserialize)]
147pub struct GraphNeighbors {
148    pub v: u32,
149    pub graph: String,
150    pub node: NodeId,
151    #[serde(default, skip_serializing_if = "EdgeDir::is_out")]
152    pub dir: EdgeDir,
153    #[serde(default, skip_serializing_if = "Option::is_none")]
154    pub edge_type: Option<String>,
155    pub depth: u32,
156    pub limit: usize,
157    /// Valid-time "as of" read (epoch micros): keep only edges whose valid-time
158    /// window contains this instant. `None` reads the current graph.
159    #[serde(default, skip_serializing_if = "Option::is_none")]
160    pub as_of: Option<u64>,
161    /// Restrict the neighborhood to elements a given conversation asserted (the
162    /// text form of its `gen_ai.conversation.id`), matched against each element's
163    /// [`SourceRef`] conversation. `None` reads the whole graph. See
164    /// [`GraphQuery::conversation`].
165    #[serde(default, skip_serializing_if = "Option::is_none")]
166    pub conversation: Option<String>,
167}
168
169/// Where a graph element was last observed: the source record an extraction came
170/// from, so a reader can navigate back to its origin. On an edge this is the
171/// record that asserted the relationship (the meaningful provenance, kept
172/// last-writer since the edge is rewritten on each observation to maintain its
173/// validity window). On a node it is the first record the entity was seen in
174/// (first-writer, so a re-observed node's stored bytes stay stable). Excluded
175/// from the content-addressed id, so it never affects identity or idempotent
176/// upsert. The complete history is the source log, which the projector can
177/// replay. Absent on the wire when unknown.
178#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
179pub enum SourceRef {
180    /// A record on the message log, by numeric stream id, topic id, partition, and
181    /// offset. Ids, not names: the pointer is route-ready and survives a rename.
182    /// The conversation is the record's `gen_ai.conversation.id`, kept for the
183    /// conversation lens but excluded from the content-addressed id, so the same
184    /// element re-observed from another conversation keeps its identity. Omitted
185    /// on the wire when unset.
186    Message {
187        stream: u32,
188        topic: u32,
189        partition: u32,
190        offset: u64,
191        #[serde(default, skip_serializing_if = "Option::is_none")]
192        conversation: Option<String>,
193    },
194    /// A key in the managed key-value store.
195    Kv { namespace: String, key: String },
196    /// A managed memory item, by its id.
197    Memory { id: String },
198}
199
200/// One node: its id, labels, attributes, optional embedding, and optional source.
201#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
202pub struct GraphNode {
203    pub id: NodeId,
204    #[serde(default, skip_serializing_if = "Vec::is_empty")]
205    pub labels: Vec<String>,
206    #[serde(default, skip_serializing_if = "Vec::is_empty")]
207    pub attrs: Vec<(String, Value)>,
208    #[serde(default, skip_serializing_if = "Option::is_none")]
209    pub embedding: Option<Vec<f32>>,
210    /// The source this node was first observed in, if known. See [`SourceRef`].
211    #[serde(default, skip_serializing_if = "Option::is_none")]
212    pub source: Option<SourceRef>,
213}
214
215impl GraphNode {
216    /// A node for the entity `value` labelled `label`. Its id is content-addressed
217    /// over the label and value (so re-observing the same entity converges on one
218    /// node), and the value is kept as a `value` attribute so a `label` or
219    /// attribute [`Match`](GraphStart::Match) start can find it. The ergonomic way
220    /// to build a node for [`GraphUpsert`] without hand-minting an id.
221    pub fn entity(label: impl Into<String>, value: impl Into<String>) -> Self {
222        let label = label.into();
223        let value = value.into();
224        let id = NodeId::content(&label, value.as_bytes());
225        Self {
226            id,
227            labels: vec![label],
228            attrs: vec![("value".to_owned(), Value::from(value))],
229            embedding: None,
230            source: None,
231        }
232    }
233}
234
235/// One edge: its id, endpoints, type, weight, attributes, and an optional
236/// valid-time window for bitemporal facts.
237#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
238pub struct GraphEdge {
239    pub id: EdgeId,
240    pub from: NodeId,
241    pub to: NodeId,
242    pub edge_type: String,
243    pub weight: f32,
244    #[serde(default, skip_serializing_if = "Vec::is_empty")]
245    pub attrs: Vec<(String, Value)>,
246    /// Valid-time start (epoch micros): when the relationship became true. `None`
247    /// is open-ended. The system-time axis (when observed) is the upsert's log
248    /// offset, so a fact can be superseded by closing `valid_to` and opening a new
249    /// edge rather than overwriting. Absent on the wire when unset.
250    #[serde(default, skip_serializing_if = "Option::is_none")]
251    pub valid_from: Option<u64>,
252    /// Valid-time end (epoch micros): when the relationship stopped being true.
253    /// `None` is still valid.
254    #[serde(default, skip_serializing_if = "Option::is_none")]
255    pub valid_to: Option<u64>,
256    /// The source that most recently asserted this relationship, if known. See
257    /// [`SourceRef`].
258    #[serde(default, skip_serializing_if = "Option::is_none")]
259    pub source: Option<SourceRef>,
260}
261
262impl GraphEdge {
263    /// An edge of `edge_type` from `from` to `to`, weight `1.0`. Its id is
264    /// content-addressed over the endpoints and type, so the same relationship is
265    /// one edge. The ergonomic way to relate two [`GraphNode`]s for an upsert.
266    pub fn relate(from: &GraphNode, edge_type: impl Into<String>, to: &GraphNode) -> Self {
267        let edge_type = edge_type.into();
268        Self {
269            id: EdgeId::content(from.id, &edge_type, to.id),
270            from: from.id,
271            to: to.id,
272            edge_type,
273            weight: 1.0,
274            attrs: Vec::new(),
275            valid_from: None,
276            valid_to: None,
277            source: None,
278        }
279    }
280
281    /// Set the source that asserted this relationship. See [`SourceRef`]. The
282    /// edge id is unchanged: provenance is metadata, not identity.
283    pub fn with_source(mut self, source: SourceRef) -> Self {
284        self.source = Some(source);
285        self
286    }
287
288    /// Set the valid-time window (epoch micros) on this edge, for a bitemporal
289    /// fact. Either bound may be `None` for open-ended. The edge id is unchanged:
290    /// validity is metadata on the relationship, not part of its identity, so
291    /// re-observing the same relationship with a new window updates the same edge.
292    pub fn valid(mut self, from: Option<u64>, to: Option<u64>) -> Self {
293        self.valid_from = from;
294        self.valid_to = to;
295        self
296    }
297
298    /// Whether this edge's valid-time window contains `at` (epoch micros). An
299    /// open bound is treated as unbounded, so an edge with no window always holds.
300    /// The half-open convention is `[valid_from, valid_to)`.
301    pub fn valid_at(&self, at: u64) -> bool {
302        self.valid_from.is_none_or(|from| at >= from) && self.valid_to.is_none_or(|to| at < to)
303    }
304}
305
306/// One path through the graph: parallel node and edge id sequences.
307#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
308pub struct Path {
309    pub nodes: Vec<NodeId>,
310    pub edges: Vec<EdgeId>,
311}
312
313/// The data a graph traversal returns. Which fields are populated depends on the
314/// query's [`GraphReturn`].
315#[derive(Clone, Debug, Default, Serialize, Deserialize)]
316pub struct GraphResult {
317    #[serde(default, skip_serializing_if = "Vec::is_empty")]
318    pub nodes: Vec<GraphNode>,
319    #[serde(default, skip_serializing_if = "Vec::is_empty")]
320    pub edges: Vec<GraphEdge>,
321    #[serde(default, skip_serializing_if = "Vec::is_empty")]
322    pub paths: Vec<Path>,
323}
324
325/// Upsert nodes and edges into a graph. The projector path: idempotent on
326/// content-addressed ids, so re-applying the same extraction is a no-op.
327#[derive(Clone, Debug, Default, Serialize, Deserialize)]
328pub struct GraphUpsert {
329    pub v: u32,
330    pub graph: String,
331    #[serde(default, skip_serializing_if = "Vec::is_empty")]
332    pub nodes: Vec<GraphNode>,
333    #[serde(default, skip_serializing_if = "Vec::is_empty")]
334    pub edges: Vec<GraphEdge>,
335}
336
337/// The result of a graph operation: `Ok` with the traversal data, or `Err` with
338/// a structured failure.
339#[derive(Clone, Debug, Serialize, Deserialize)]
340#[non_exhaustive]
341pub enum GraphReply {
342    Ok(GraphResult),
343    Err(GraphError),
344}
345
346/// Why a graph operation failed.
347#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
348#[non_exhaustive]
349pub enum GraphError {
350    #[error("graph not supported: {0}")]
351    Unsupported(String),
352    #[error("unauthorized: {0}")]
353    Unauthorized(String),
354    /// The request named a graph that fails [`validate_graph_name`].
355    #[error("invalid graph name: {0}")]
356    InvalidName(String),
357    #[error("graph not found: {0}")]
358    NotFound(String),
359    #[error("traversal too large: {what} is {size}, exceeds cap {cap}")]
360    TooLarge {
361        what: String,
362        size: usize,
363        cap: usize,
364    },
365    #[error("graph backend error: {0}")]
366    Backend(String),
367    #[error("unsupported graph op version (expected {expected}, got {got})")]
368    Version { expected: u32, got: u32 },
369}
370
371#[cfg(all(test, feature = "cbor"))]
372mod tests {
373    use super::*;
374    use crate::codes::GRAPH_OP_VERSION;
375    use crate::framing::{decode_named, encode_named};
376    use crate::query::CmpOp;
377
378    #[test]
379    fn given_a_graph_query_when_round_tripped_then_should_preserve_traversal() {
380        let query = GraphQuery {
381            v: GRAPH_OP_VERSION,
382            graph: "knowledge".to_owned(),
383            start: GraphStart::Match(Filter::pred("label", CmpOp::Eq, "Person")),
384            traverse: vec![
385                Hop {
386                    edge_type: Some("works_at".to_owned()),
387                    dir: EdgeDir::Out,
388                    max: 1,
389                },
390                Hop {
391                    edge_type: Some("located_in".to_owned()),
392                    dir: EdgeDir::Out,
393                    max: 1,
394                },
395            ],
396            node_filter: None,
397            edge_filter: None,
398            return_: GraphReturn::Paths,
399            limit: 100,
400            fork: None,
401            consistency: Consistency::Eventual,
402            as_of: Some(1_900_000_000_000_000),
403            conversation: None,
404        };
405        let bytes = encode_named(&query).expect("serializes");
406        let back: GraphQuery = decode_named(&bytes).expect("deserializes");
407        assert_eq!(back.graph, "knowledge");
408        assert_eq!(back.traverse.len(), 2);
409        assert_eq!(back.return_, GraphReturn::Paths);
410        assert_eq!(back.as_of, Some(1_900_000_000_000_000));
411    }
412
413    #[test]
414    fn given_a_graph_result_when_round_tripped_then_should_preserve_nodes_and_edges() {
415        let reply = GraphReply::Ok(GraphResult {
416            nodes: vec![GraphNode {
417                id: NodeId::from_u128(1),
418                labels: vec!["Person".to_owned()],
419                attrs: vec![("name".to_owned(), Value::from("Alice"))],
420                embedding: None,
421                source: None,
422            }],
423            edges: vec![GraphEdge {
424                id: EdgeId::from_u128(2),
425                from: NodeId::from_u128(1),
426                to: NodeId::from_u128(3),
427                edge_type: "works_at".to_owned(),
428                weight: 1.0,
429                attrs: Vec::new(),
430                valid_from: None,
431                valid_to: None,
432                source: None,
433            }],
434            paths: Vec::new(),
435        });
436        let bytes = encode_named(&reply).expect("serializes");
437        let back: GraphReply = decode_named(&bytes).expect("deserializes");
438        let GraphReply::Ok(result) = back else {
439            panic!("expected Ok");
440        };
441        assert_eq!(result.nodes.len(), 1);
442        assert_eq!(result.edges[0].edge_type, "works_at");
443    }
444
445    #[test]
446    fn given_a_nearest_start_when_round_tripped_then_should_preserve_the_seed() {
447        let query = GraphQuery {
448            v: GRAPH_OP_VERSION,
449            graph: "knowledge".to_owned(),
450            start: GraphStart::Nearest {
451                embedding: vec![0.1, 0.2, 0.3],
452                k: 5,
453            },
454            traverse: Vec::new(),
455            node_filter: None,
456            edge_filter: None,
457            return_: GraphReturn::Nodes,
458            limit: 10,
459            fork: None,
460            consistency: Consistency::Eventual,
461            as_of: None,
462            conversation: None,
463        };
464        let bytes = encode_named(&query).expect("serializes");
465        let back: GraphQuery = decode_named(&bytes).expect("deserializes");
466        match back.start {
467            GraphStart::Nearest { embedding, k } => {
468                assert_eq!(embedding, vec![0.1, 0.2, 0.3]);
469                assert_eq!(k, 5);
470            }
471            other => panic!("expected Nearest, got {other:?}"),
472        }
473    }
474
475    #[test]
476    fn given_a_node_id_when_round_tripped_through_a_string_then_should_be_equal() {
477        let id = NodeId::from_u128(987_654_321);
478        let parsed: NodeId = id.to_string().parse().expect("a node id parses");
479        assert_eq!(parsed, id);
480    }
481
482    #[test]
483    fn given_the_same_entity_when_addressed_twice_then_should_converge_on_one_node_id() {
484        let a = NodeId::content("Person", b"Alice");
485        let b = NodeId::content("Person", b"Alice");
486        assert_eq!(a, b, "the same entity is one node");
487        // A different label or value is a different node.
488        assert_ne!(a, NodeId::content("Company", b"Alice"));
489        assert_ne!(a, NodeId::content("Person", b"Bob"));
490    }
491
492    #[test]
493    fn given_the_pinned_entity_when_addressed_then_should_match_the_golden_id() {
494        // The cross-SDK golden vector: the Person entity "Alice". Every SDK renders
495        // this NodeId identically, so a graph shared across languages converges.
496        assert_eq!(
497            NodeId::content("Person", b"Alice").to_string(),
498            "13NCEPHNVFHHGNK9GD3MT0W1AB"
499        );
500    }
501
502    #[test]
503    fn given_two_nodes_when_related_then_should_content_address_the_edge() {
504        let alice = GraphNode::entity("Person", "Alice");
505        let acme = GraphNode::entity("Company", "Acme");
506        let one = GraphEdge::relate(&alice, "works_at", &acme);
507        let two = GraphEdge::relate(&alice, "works_at", &acme);
508        assert_eq!(one.id, two.id, "the same relationship is one edge");
509        assert_eq!(one.from, alice.id);
510        assert_eq!(one.to, acme.id);
511        // The direction is part of the identity: the reverse edge is a different id.
512        assert_ne!(one.id, GraphEdge::relate(&acme, "works_at", &alice).id);
513    }
514
515    #[test]
516    fn given_an_edge_validity_window_when_checked_then_should_hold_only_inside_it() {
517        let alice = GraphNode::entity("User", "alice");
518        let pro = GraphNode::entity("Plan", "pro");
519        let edge = GraphEdge::relate(&alice, "on_plan", &pro).valid(Some(100), Some(200));
520        assert!(!edge.valid_at(99), "before the window");
521        assert!(edge.valid_at(100), "the lower bound is inclusive");
522        assert!(edge.valid_at(150), "inside the window");
523        assert!(!edge.valid_at(200), "the upper bound is exclusive");
524        let open = GraphEdge::relate(&alice, "on_plan", &pro);
525        assert!(open.valid_at(0) && open.valid_at(u64::MAX));
526        assert_eq!(edge.id, open.id, "validity is not part of edge identity");
527    }
528
529    #[test]
530    fn given_an_edge_without_validity_when_serialized_then_should_omit_the_window() {
531        let edge = GraphEdge::relate(
532            &GraphNode::entity("A", "x"),
533            "rel",
534            &GraphNode::entity("B", "y"),
535        );
536        let json = serde_json::to_string(&edge).expect("serializes");
537        assert!(
538            !json.contains("valid_from") && !json.contains("valid_to"),
539            "an unset window must be omitted so a pre-bitemporal edge is byte-identical: {json}"
540        );
541    }
542
543    #[test]
544    fn given_a_node_without_a_source_when_serialized_then_should_omit_it() {
545        let node = GraphNode::entity("Person", "Alice");
546        let json = serde_json::to_string(&node).expect("serializes");
547        assert!(
548            !json.contains("source"),
549            "an unknown source must be omitted so a pre-provenance node is byte-identical: {json}"
550        );
551    }
552
553    #[test]
554    fn given_a_node_with_a_source_when_round_tripped_then_should_preserve_it_and_keep_identity() {
555        let mut node = GraphNode::entity("Component", "cache");
556        node.source = Some(SourceRef::Message {
557            stream: 7,
558            topic: 2,
559            partition: 3,
560            offset: 4096,
561            conversation: None,
562        });
563        let bytes = encode_named(&node).expect("serializes");
564        let back: GraphNode = decode_named(&bytes).expect("deserializes");
565        assert_eq!(back.source, node.source);
566        assert_eq!(
567            back.id,
568            GraphNode::entity("Component", "cache").id,
569            "source is not part of node identity"
570        );
571    }
572
573    #[test]
574    fn given_an_edge_with_a_source_when_round_tripped_then_should_preserve_it_and_keep_identity() {
575        let from = GraphNode::entity("A", "x");
576        let to = GraphNode::entity("B", "y");
577        let edge = GraphEdge::relate(&from, "rel", &to).with_source(SourceRef::Kv {
578            namespace: "ns".to_owned(),
579            key: "k".to_owned(),
580        });
581        let bytes = encode_named(&edge).expect("serializes");
582        let back: GraphEdge = decode_named(&bytes).expect("deserializes");
583        assert_eq!(back.source, edge.source);
584        assert_eq!(
585            back.id,
586            GraphEdge::relate(&from, "rel", &to).id,
587            "source is not part of edge identity"
588        );
589    }
590
591    #[test]
592    fn given_a_source_without_a_conversation_when_serialized_then_should_omit_it() {
593        let source = SourceRef::Message {
594            stream: 1,
595            topic: 1,
596            partition: 0,
597            offset: 0,
598            conversation: None,
599        };
600        let json = serde_json::to_string(&source).expect("serializes");
601        assert!(
602            !json.contains("conversation"),
603            "an unset conversation must be omitted so a pre-conversation source stays byte-identical: {json}"
604        );
605    }
606
607    #[test]
608    fn given_a_source_with_a_conversation_when_round_tripped_then_should_preserve_it() {
609        let mut node = GraphNode::entity("Ticket", "7");
610        node.source = Some(SourceRef::Message {
611            stream: 4,
612            topic: 6,
613            partition: 2,
614            offset: 99,
615            conversation: Some("01KWM3K3XEP3NP5TN850J17YBP".to_owned()),
616        });
617        let bytes = encode_named(&node).expect("serializes");
618        let back: GraphNode = decode_named(&bytes).expect("deserializes");
619        assert_eq!(
620            back.source, node.source,
621            "the conversation survives the round trip"
622        );
623        assert_eq!(
624            back.id,
625            GraphNode::entity("Ticket", "7").id,
626            "the conversation is provenance, not identity"
627        );
628    }
629
630    #[test]
631    fn given_a_conversation_filter_on_a_traversal_when_round_tripped_then_should_preserve_it() {
632        let query = GraphQuery {
633            v: GRAPH_OP_VERSION,
634            graph: "knowledge".to_owned(),
635            start: GraphStart::Ids(vec![NodeId::from_u128(1)]),
636            traverse: Vec::new(),
637            node_filter: None,
638            edge_filter: None,
639            return_: GraphReturn::Nodes,
640            limit: 10,
641            fork: None,
642            consistency: Consistency::Eventual,
643            as_of: None,
644            conversation: Some("01KWM3K3XEP3NP5TN850J17YBP".to_owned()),
645        };
646        let back: GraphQuery =
647            decode_named(&encode_named(&query).expect("serializes")).expect("deserializes");
648        assert_eq!(
649            back.conversation.as_deref(),
650            Some("01KWM3K3XEP3NP5TN850J17YBP")
651        );
652        // The default (no filter) stays omitted on the wire.
653        let unfiltered = GraphQuery {
654            conversation: None,
655            ..query
656        };
657        let json = serde_json::to_string(&unfiltered).expect("serializes");
658        assert!(
659            !json.contains("conversation"),
660            "an unset filter is omitted: {json}"
661        );
662    }
663
664    #[test]
665    fn given_a_max_element_reply_with_source_when_encoded_then_should_fit_one_frame() {
666        use crate::limits::{MAX_FRAME_BYTES, MAX_GRAPH_RESULT_ELEMENTS};
667        let source = SourceRef::Message {
668            stream: u32::MAX,
669            topic: u32::MAX,
670            partition: u32::MAX,
671            offset: u64::MAX,
672            conversation: Some("7ZZZZZZZZZZZZZZZZZZZZZZZZZ".to_owned()),
673        };
674        let half = (MAX_GRAPH_RESULT_ELEMENTS / 2) as u128;
675        let nodes = (0..half)
676            .map(|i| {
677                let mut node = GraphNode::entity("Component", format!("entity-{i}"));
678                node.source = Some(source.clone());
679                node
680            })
681            .collect();
682        let edges = (0..half)
683            .map(|i| GraphEdge {
684                id: EdgeId::from_u128(i),
685                from: NodeId::from_u128(i),
686                to: NodeId::from_u128(i + 1),
687                edge_type: "relates_to".to_owned(),
688                weight: 1.0,
689                attrs: Vec::new(),
690                valid_from: None,
691                valid_to: None,
692                source: Some(source.clone()),
693            })
694            .collect();
695        let reply = GraphReply::Ok(GraphResult {
696            nodes,
697            edges,
698            paths: Vec::new(),
699        });
700        let encoded = encode_named(&reply).expect("serializes");
701        assert!(
702            encoded.len() < MAX_FRAME_BYTES,
703            "a full {MAX_GRAPH_RESULT_ELEMENTS}-element reply with source is {} bytes, over the frame cap {MAX_FRAME_BYTES}",
704            encoded.len()
705        );
706    }
707}
708
709/// The canonical graph-name rule, shared by the SDK client edge and the
710/// serving plane: non-empty, at most [`MAX_GRAPH_NAME_BYTES`] bytes, no ASCII
711/// control characters.
712pub fn validate_graph_name(name: &str) -> Result<(), InvalidError> {
713    if name.is_empty() {
714        return Err(InvalidError::new("graph name must not be empty"));
715    }
716    if name.len() > MAX_GRAPH_NAME_BYTES {
717        return Err(InvalidError::new(format!(
718            "graph name is {}B, exceeds cap {MAX_GRAPH_NAME_BYTES}B",
719            name.len()
720        )));
721    }
722    if name.bytes().any(|byte| byte.is_ascii_control()) {
723        return Err(InvalidError::new(
724            "graph name must not contain ASCII control characters",
725        ));
726    }
727    Ok(())
728}
laser_wire/graph.rs

laser_wire/
graph.rs
