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

laser_wire/
graph.rs
