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

laser_wire/
graph.rs
