laser_wire/
graph.rs

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

laser_wire/
graph.rs
