laser_sdk/
graph.rs

1use crate::error::LaserError;
2use crate::laser::Laser;
3use crate::types::ConversationId;
4
5impl Laser {
6    /// A handle to the knowledge-graph surface `name`. Traversals require the
7    /// `graph` capability (LaserData Cloud) and ride the managed binary
8    /// transport. Against Apache Iggy a fetch returns
9    /// [`LaserError::Unsupported`]. The name is sent as
10    /// [`resource_name`](Self::resource_name) names it.
11    pub fn graph(&self, name: impl Into<String>) -> GraphHandle<'_> {
12        GraphHandle {
13            laser: self,
14            name: self.resource_name(&name.into()),
15            start: None,
16            hops: Vec::new(),
17            node_filter: None,
18            edge_filter: None,
19            return_: laser_wire::graph::GraphReturn::Nodes,
20            limit: laser_wire::limits::DEFAULT_RECALL_LIMIT,
21            as_of: None,
22            conversation: None,
23            stream: None,
24            session: None,
25            producer: None,
26            source: None,
27        }
28    }
29}
30
31/// A fluent knowledge-graph traversal, created by [`Laser::graph`]. Set a start
32/// (`start_ids`/`start_match`/`start_nearest`), add hops (`out`/`incoming`), pick
33/// what to return, and finish with `.fetch().await`. Gated on the `graph`
34/// feature: a traversal rides the managed binary transport.
35pub struct GraphHandle<'a> {
36    laser: &'a Laser,
37    name: String,
38    start: Option<laser_wire::graph::GraphStart>,
39    hops: Vec<laser_wire::graph::Hop>,
40    node_filter: Option<laser_wire::query::Filter>,
41    edge_filter: Option<laser_wire::query::Filter>,
42    return_: laser_wire::graph::GraphReturn,
43    limit: usize,
44    as_of: Option<u64>,
45    conversation: Option<String>,
46    stream: Option<String>,
47    session: Option<laser_wire::agent::SessionRef>,
48    producer: Option<laser_wire::graph::ProducerInfo>,
49    source: Option<laser_wire::graph::SourceRef>,
50}
51
52impl GraphHandle<'_> {
53    /// Link every upsert to `session`, so the deployment records which session
54    /// wrote each element.
55    #[must_use]
56    pub fn in_session(mut self, session: laser_wire::agent::SessionRef) -> Self {
57        self.session = Some(session);
58        self
59    }
60
61    /// Stamp `producer` on every upserted element that has none.
62    #[must_use]
63    pub fn produced_by(mut self, producer: laser_wire::graph::ProducerInfo) -> Self {
64        self.producer = Some(producer);
65        self
66    }
67
68    /// Stamp `source` on every upserted element that has none.
69    #[must_use]
70    pub fn sourced_from(mut self, source: laser_wire::graph::SourceRef) -> Self {
71        self.source = Some(source);
72        self
73    }
74
75    /// Narrow the traversal to elements a single conversation asserted (the
76    /// conversation lens): only nodes and edges whose source records that
77    /// `conversation` are traversed and returned. Applies to both `fetch` and
78    /// `neighbors`. Reads the whole graph when unset.
79    #[must_use]
80    pub fn conversation(mut self, conversation: ConversationId) -> Self {
81        self.conversation = Some(conversation.to_string());
82        self.stream = self.laser.resource_stream().map(str::to_owned);
83        self
84    }
85    /// Start the traversal from explicit node ids.
86    #[must_use]
87    pub fn start_ids(mut self, ids: Vec<laser_wire::graph::NodeId>) -> Self {
88        self.start = Some(laser_wire::graph::GraphStart::Ids(ids));
89        self
90    }
91
92    /// Start from the nodes matching a predicate.
93    #[must_use]
94    pub fn start_match(mut self, filter: laser_wire::query::Filter) -> Self {
95        self.start = Some(laser_wire::graph::GraphStart::Match(filter));
96        self
97    }
98
99    /// Start from the nodes nearest an embedding (vector-seeded traversal).
100    #[must_use]
101    pub fn start_nearest(mut self, embedding: Vec<f32>, k: usize) -> Self {
102        self.start = Some(laser_wire::graph::GraphStart::Nearest { embedding, k });
103        self
104    }
105
106    /// Follow outgoing edges of `edge_type` one hop.
107    #[must_use]
108    pub fn out(mut self, edge_type: impl Into<String>) -> Self {
109        self.hops.push(laser_wire::graph::Hop {
110            edge_type: Some(edge_type.into()),
111            dir: laser_wire::graph::EdgeDir::Out,
112            max: 1,
113        });
114        self
115    }
116
117    /// Follow incoming edges of `edge_type` one hop.
118    #[must_use]
119    pub fn incoming(mut self, edge_type: impl Into<String>) -> Self {
120        self.hops.push(laser_wire::graph::Hop {
121            edge_type: Some(edge_type.into()),
122            dir: laser_wire::graph::EdgeDir::In,
123            max: 1,
124        });
125        self
126    }
127
128    /// Follow edges of `edge_type` one hop in both directions.
129    #[must_use]
130    pub fn both(mut self, edge_type: impl Into<String>) -> Self {
131        self.hops.push(laser_wire::graph::Hop {
132            edge_type: Some(edge_type.into()),
133            dir: laser_wire::graph::EdgeDir::Both,
134            max: 1,
135        });
136        self
137    }
138
139    /// Return the traversed edges instead of the reachable nodes.
140    #[must_use]
141    pub fn return_edges(mut self) -> Self {
142        self.return_ = laser_wire::graph::GraphReturn::Edges;
143        self
144    }
145
146    /// Return the traversed edges as `(source, type, destination)` triplets
147    /// instead of the reachable nodes.
148    #[must_use]
149    pub fn return_triplets(mut self) -> Self {
150        self.return_ = laser_wire::graph::GraphReturn::Triplets;
151        self
152    }
153
154    /// Return whole paths (node and edge id sequences) instead of nodes.
155    #[must_use]
156    pub fn return_paths(mut self) -> Self {
157        self.return_ = laser_wire::graph::GraphReturn::Paths;
158        self
159    }
160
161    /// Cap the number of elements returned.
162    #[must_use]
163    pub fn limit(mut self, limit: usize) -> Self {
164        self.limit = limit;
165        self
166    }
167
168    /// Read the graph as of `micros` (valid-time, epoch micros): only edges whose
169    /// valid-time window contains that instant are traversed. Applies to both
170    /// `fetch` and `neighbors`.
171    #[must_use]
172    pub fn as_of(mut self, micros: u64) -> Self {
173        self.as_of = Some(micros);
174        self
175    }
176
177    /// Run the traversal. Requires `managed_graph`. Otherwise returns
178    /// [`LaserError::Unsupported`].
179    pub async fn fetch(self) -> Result<laser_wire::graph::GraphResult, LaserError> {
180        use laser_wire::graph::{GraphQuery, GraphStart};
181        self.require_graph()?;
182        let query = GraphQuery {
183            v: laser_wire::codes::GRAPH_OP_VERSION,
184            graph: self.name,
185            start: self.start.unwrap_or(GraphStart::Ids(Vec::new())),
186            traverse: self.hops,
187            node_filter: self.node_filter,
188            edge_filter: self.edge_filter,
189            return_: self.return_,
190            limit: self.limit,
191            fork: None,
192            consistency: laser_wire::query::Consistency::Eventual,
193            as_of: self.as_of,
194            conversation: self.conversation,
195            stream: self.stream,
196        };
197        let payload = laser_wire::framing::encode_named(&query)
198            .map_err(|error| LaserError::Codec(format!("encode graph query: {error}")))?;
199        let payload = self
200            .laser
201            .send_raw_with_response(laser_wire::codes::AGDX_GRAPH_QUERY_CODE, payload)
202            .await?;
203        decode_graph_reply(&payload)
204    }
205
206    /// Read a node's neighbors: the nodes reachable in `dir` over `edge_type` (or
207    /// any type when `None`), following the same hop `depth` times. The cheap,
208    /// common traversal. Requires `managed_graph`. Otherwise returns
209    /// [`LaserError::Unsupported`].
210    pub async fn neighbors(
211        self,
212        node: laser_wire::graph::NodeId,
213        dir: laser_wire::graph::EdgeDir,
214        edge_type: Option<String>,
215        depth: u32,
216    ) -> Result<laser_wire::graph::GraphResult, LaserError> {
217        use laser_wire::graph::GraphNeighbors;
218        self.require_graph()?;
219        let request = GraphNeighbors {
220            v: laser_wire::codes::GRAPH_OP_VERSION,
221            graph: self.name,
222            node,
223            dir,
224            edge_type,
225            depth,
226            limit: self.limit,
227            as_of: self.as_of,
228            conversation: self.conversation,
229            stream: self.stream,
230        };
231        let payload = laser_wire::framing::encode_named(&request)
232            .map_err(|error| LaserError::Codec(format!("encode graph neighbors: {error}")))?;
233        let payload = self
234            .laser
235            .send_raw_with_response(laser_wire::codes::AGDX_GRAPH_NEIGHBORS_CODE, payload)
236            .await?;
237        decode_graph_reply(&payload)
238    }
239
240    /// Relate two entities in one call: `link("operator:42", "opened",
241    /// "ticket:7")` upserts both content-addressed entity nodes and the typed
242    /// edge between them. Sugar over [`upsert`](Self::upsert), so re-linking
243    /// the same triple converges on the same nodes and edge. The entity ids
244    /// double as their labels' values (`kind:value` strings work well).
245    pub async fn link(
246        self,
247        from: impl Into<String>,
248        relation: impl Into<String>,
249        to: impl Into<String>,
250    ) -> Result<(), LaserError> {
251        use laser_wire::graph::GraphEdge;
252        let from = entity_node(&from.into());
253        let to = entity_node(&to.into());
254        let edge = GraphEdge::relate(&from, relation, &to);
255        self.upsert(vec![from, to], vec![edge]).await
256    }
257
258    /// Assert the latest value of a single-valued relationship: close every
259    /// live `relation` edge from `from` that points at a DIFFERENT target
260    /// (`valid_to` now, the bitemporal supersede), then link `from` to `to`.
261    /// The deterministic fact-invalidation mechanics (same subject, same
262    /// predicate, new object supersedes the old), returning how many edges
263    /// were closed. Contradiction detection beyond same-subject-same-predicate
264    /// is the caller's policy, and recording an `improve` feedback item that
265    /// links superseded to superseding is the caller's memory concern.
266    pub async fn relink(
267        self,
268        from: impl Into<String>,
269        relation: impl Into<String>,
270        to: impl Into<String>,
271    ) -> Result<usize, LaserError> {
272        use laser_wire::graph::EdgeDir;
273        let laser = self.laser;
274        let graph = self.name.clone();
275        let from = from.into();
276        let relation = relation.into();
277        let to = to.into();
278        let from_node = entity_node(&from);
279        let to_node = entity_node(&to);
280        let live = laser
281            .graph(&graph)
282            .neighbors(from_node.id, EdgeDir::Out, Some(relation.clone()), 1)
283            .await?;
284        let now = now_micros();
285        let superseded: Vec<laser_wire::graph::GraphEdge> = live
286            .edges
287            .into_iter()
288            .filter(|edge| edge.to != to_node.id && edge.valid_to.is_none())
289            .map(|mut edge| {
290                edge.valid_to = Some(now);
291                edge
292            })
293            .collect();
294        let closed = superseded.len();
295        if !superseded.is_empty() {
296            laser.graph(&graph).upsert(Vec::new(), superseded).await?;
297        }
298        self.link(from, relation, to).await?;
299        Ok(closed)
300    }
301
302    /// Close the relationship `link` opened: rewrite the edge with `valid_to`
303    /// now, so the fact is superseded without being destroyed (a bitemporal
304    /// close). The nodes stay.
305    pub async fn unlink(
306        self,
307        from: impl Into<String>,
308        relation: impl Into<String>,
309        to: impl Into<String>,
310    ) -> Result<(), LaserError> {
311        use laser_wire::graph::GraphEdge;
312        let from = entity_node(&from.into());
313        let to = entity_node(&to.into());
314        let mut edge = GraphEdge::relate(&from, relation, &to);
315        edge.valid_to = Some(now_micros());
316        self.upsert(Vec::new(), vec![edge]).await
317    }
318
319    /// Write `nodes` and `edges` into the graph: the projector path, surfaced for
320    /// callers that build the graph directly rather than through a `graph`
321    /// projection. Idempotent on content-addressed ids
322    /// ([`GraphNode::entity`](laser_wire::graph::GraphNode::entity),
323    /// [`GraphEdge::relate`](laser_wire::graph::GraphEdge::relate)), so re-applying
324    /// the same entities is a no-op. Requires `managed_graph`. Otherwise returns
325    /// [`LaserError::Unsupported`].
326    pub async fn upsert(
327        self,
328        nodes: Vec<laser_wire::graph::GraphNode>,
329        edges: Vec<laser_wire::graph::GraphEdge>,
330    ) -> Result<(), LaserError> {
331        use laser_wire::graph::GraphUpsert;
332        use laser_wire::validate::Validate;
333        self.require_graph()?;
334        let (mut nodes, mut edges) = (nodes, edges);
335        for node in &mut nodes {
336            if node.producer.is_none() {
337                node.producer.clone_from(&self.producer);
338            }
339            if node.source.is_none() {
340                node.source.clone_from(&self.source);
341            }
342        }
343        for edge in &mut edges {
344            if edge.producer.is_none() {
345                edge.producer.clone_from(&self.producer);
346            }
347            if edge.source.is_none() {
348                edge.source.clone_from(&self.source);
349            }
350        }
351        let request = GraphUpsert {
352            v: laser_wire::codes::GRAPH_OP_VERSION,
353            graph: self.name,
354            session: self.session,
355            nodes,
356            edges,
357        };
358        // Spend the cap failure locally rather than amplifying an over-cap
359        // upsert against the managed backend, which rejects it with the same
360        // reason.
361        request.validate()?;
362        let payload = laser_wire::framing::encode_named(&request)
363            .map_err(|error| LaserError::Codec(format!("encode graph upsert: {error}")))?;
364        let payload = self
365            .laser
366            .send_raw_with_response(laser_wire::codes::AGDX_GRAPH_UPSERT_CODE, payload)
367            .await?;
368        decode_graph_reply(&payload).map(|_| ())
369    }
370
371    // Every graph op rides the managed binary transport, so it is unavailable
372    // against Apache Iggy. Fail the same way before encoding any request.
373    fn require_graph(&self) -> Result<(), LaserError> {
374        laser_wire::graph::validate_graph_name(&self.name)?;
375        if self.laser.current_capabilities().graph {
376            Ok(())
377        } else {
378            Err(LaserError::unsupported(
379                "graph",
380                "graph traversal is not served by this deployment",
381            ))
382        }
383    }
384}
385
386// Decode a managed `GraphReply` into the `Ok` result or its typed error.
387fn decode_graph_reply(payload: &[u8]) -> Result<laser_wire::graph::GraphResult, LaserError> {
388    use laser_wire::graph::GraphReply;
389    match crate::error::decode_managed_reply::<GraphReply>(payload)? {
390        GraphReply::Ok(result) => Ok(result),
391        GraphReply::Err(error) => Err(error.into()),
392        _ => Err(LaserError::Protocol(
393            "graph: unknown reply variant".to_owned(),
394        )),
395    }
396}
397
398/// The entity node for a `kind:value` style id: the id string is both the
399/// label-ish identity and the `value` attribute, so `link`ed entities converge
400/// on one content-addressed node per id.
401fn entity_node(id: &str) -> laser_wire::graph::GraphNode {
402    let (label, value) = id.split_once(':').unwrap_or(("entity", id));
403    laser_wire::graph::GraphNode::entity(label, value)
404}
405
406fn now_micros() -> u64 {
407    use std::time::{SystemTime, UNIX_EPOCH};
408    SystemTime::now()
409        .duration_since(UNIX_EPOCH)
410        .map(|duration| u64::try_from(duration.as_micros()).unwrap_or(u64::MAX))
411        .unwrap_or(0)
412}
413
414#[cfg(test)]
415mod tests {
416    use crate::laser::{Laser, ResourceNaming};
417    use crate::types::ConversationId;
418
419    fn laser() -> Laser {
420        Laser::from_client(crate::iggy::prelude::IggyClient::default()).with_default_stream("acme")
421    }
422
423    #[test]
424    fn given_a_default_stream_when_opening_a_graph_then_should_scope_the_name_and_the_lens() {
425        let laser = laser();
426        let handle = laser
427            .graph("knowledge")
428            .conversation(ConversationId::derive("lens"));
429        assert_eq!(handle.name, "stream:acme/knowledge");
430        assert_eq!(handle.stream.as_deref(), Some("acme"));
431    }
432
433    #[test]
434    fn given_bare_naming_when_opening_a_graph_then_should_send_the_name_and_lens_unchanged() {
435        let laser = laser().with_resource_naming(ResourceNaming::Bare);
436        let handle = laser
437            .graph("knowledge")
438            .conversation(ConversationId::derive("lens"));
439        assert_eq!(handle.name, "knowledge");
440        assert_eq!(handle.stream, None);
441    }
442}
laser_sdk/graph.rs

laser_sdk/
graph.rs
