laser_wire/
graph.rs

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

laser_wire/
graph.rs
