semantic_memory/
graph_edges.rs

1//! First-class stored graph edges — durable, typed relationships between
2//! any two nodes in the knowledge graph.
3//!
4//! Unlike the derived edges in [`crate::graph`] (which are computed on-the-fly
5//! from SQLite state), these edges are explicitly created by agents or users
6//! and persisted in the `graph_edges` table (migration V27).
7//!
8//! ## Authority
9//!
10//! `semantic-memory` is authoritative for graph edge state. These edges are
11//! first-class knowledge relationships, not projection lineage (which lives in
12//! `derivation_edges`).
13//!
14//! ## Bitemporal semantics
15//!
16//! Edges are append-only. Invalidation is a separate operation that sets
17//! `is_invalidated = 1` with a timestamp and reason — the original row is
18//! never deleted. This follows the append-plus-supersession doctrine.
19//!
20//! ## Idempotent insertion
21//!
22//! Each edge carries a `content_digest` (blake3 of source + target + edge_type
23//! + weight + metadata). Inserting the same edge twice returns the existing
24//! row ID without creating a duplicate.
25
26use crate::error::MemoryError;
27use crate::types::{GraphEdge, GraphEdgeType};
28use chrono::Utc;
29use rusqlite::{params, Connection};
30use stack_ids::DigestBuilder;
31
32/// A stored graph edge row.
33#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
34pub struct StoredGraphEdge {
35    /// Unique edge ID (UUID v4).
36    pub id: String,
37    /// Source node ID (prefixed, e.g. `fact:<uuid>`).
38    pub source: String,
39    /// Target node ID (prefixed, e.g. `fact:<uuid>`).
40    pub target: String,
41    /// Serialized GraphEdgeType JSON.
42    pub edge_type: String,
43    /// Deserialized edge type.
44    #[serde(skip)]
45    pub edge_type_parsed: Option<GraphEdgeType>,
46    /// Edge weight.
47    pub weight: f64,
48    /// Optional metadata JSON.
49    pub metadata: Option<String>,
50    /// Content digest (blake3 of source+target+edge_type+weight+metadata).
51    pub content_digest: String,
52    /// Recorded timestamp (microsecond precision).
53    pub recorded_at: String,
54    /// Whether this edge has been invalidated.
55    pub is_invalidated: bool,
56    /// Invalidation timestamp.
57    pub invalidated_at: Option<String>,
58    /// Invalidation reason.
59    pub invalidation_reason: Option<String>,
60}
61
62/// Parameters for creating a graph edge.
63#[derive(Debug, Clone)]
64pub struct AddGraphEdgeParams {
65    /// Source node ID (prefixed, e.g. `fact:<uuid>`, `namespace:<name>`).
66    pub source: String,
67    /// Target node ID (prefixed).
68    pub target: String,
69    /// Edge type (semantic, temporal, causal, entity).
70    pub edge_type: GraphEdgeType,
71    /// Edge weight (interpretation depends on edge_type).
72    pub weight: f64,
73    /// Optional metadata.
74    pub metadata: Option<serde_json::Value>,
75}
76
77/// Insert a graph edge. Idempotent on content_digest.
78pub(crate) fn insert_graph_edge(
79    conn: &Connection,
80    params: &AddGraphEdgeParams,
81) -> Result<StoredGraphEdge, MemoryError> {
82    let edge_type_json = serde_json::to_string(&params.edge_type)
83        .map_err(|e| MemoryError::Other(format!("failed to serialize edge_type: {e}")))?;
84
85    let metadata_json = match &params.metadata {
86        Some(v) => Some(serde_json::to_string(v)
87            .map_err(|e| MemoryError::Other(format!("failed to serialize metadata: {e}")))?),
88        None => None,
89    };
90
91    let content_digest = compute_edge_digest(
92        &params.source,
93        &params.target,
94        &edge_type_json,
95        params.weight,
96        &metadata_json,
97    );
98
99    // Check for existing edge with same digest (idempotent).
100    let existing: Option<(String, String)> = conn
101        .query_row(
102            "SELECT id, recorded_at FROM graph_edges WHERE content_digest = ?1 AND is_invalidated = 0",
103            params![&content_digest],
104            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
105        )
106        .ok();
107
108    if let Some((id, recorded_at)) = existing {
109        return Ok(StoredGraphEdge {
110            id,
111            source: params.source.clone(),
112            target: params.target.clone(),
113            edge_type: edge_type_json,
114            edge_type_parsed: Some(params.edge_type.clone()),
115            weight: params.weight,
116            metadata: metadata_json,
117            content_digest,
118            recorded_at,
119            is_invalidated: false,
120            invalidated_at: None,
121            invalidation_reason: None,
122        });
123    }
124
125    let id = uuid::Uuid::new_v4().to_string();
126    let recorded_at = Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string();
127
128    conn.execute(
129        "INSERT INTO graph_edges (id, source, target, edge_type, weight, metadata, content_digest, recorded_at)
130         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
131        params![
132            &id,
133            &params.source,
134            &params.target,
135            &edge_type_json,
136            params.weight,
137            metadata_json.as_deref(),
138            &content_digest,
139            &recorded_at,
140        ],
141    )
142    .map_err(|e| MemoryError::Database(e))?;
143
144    Ok(StoredGraphEdge {
145        id,
146        source: params.source.clone(),
147        target: params.target.clone(),
148        edge_type: edge_type_json,
149        edge_type_parsed: Some(params.edge_type.clone()),
150        weight: params.weight,
151        metadata: metadata_json,
152        content_digest,
153        recorded_at,
154        is_invalidated: false,
155        invalidated_at: None,
156        invalidation_reason: None,
157    })
158}
159
160/// List all stored graph edges involving a given node (either as source or
161/// target), excluding invalidated edges.
162pub(crate) fn list_graph_edges_for_node(
163    conn: &Connection,
164    node_id: &str,
165) -> Result<Vec<StoredGraphEdge>, MemoryError> {
166    let mut stmt = conn.prepare(
167        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
168                is_invalidated, invalidated_at, invalidation_reason
169         FROM graph_edges
170         WHERE (source = ?1 OR target = ?1) AND is_invalidated = 0
171         ORDER BY recorded_at ASC",
172    )?;
173    let rows = stmt
174        .query_map(params![node_id], |row| {
175            Ok(StoredGraphEdge {
176                id: row.get(0)?,
177                source: row.get(1)?,
178                target: row.get(2)?,
179                edge_type: row.get(3)?,
180                edge_type_parsed: None,
181                weight: row.get(4)?,
182                metadata: row.get(5)?,
183                content_digest: row.get(6)?,
184                recorded_at: row.get(7)?,
185                is_invalidated: row.get::<_, i64>(8)? != 0,
186                invalidated_at: row.get(9)?,
187                invalidation_reason: row.get(10)?,
188            })
189        })?
190        .collect::<Result<Vec<_>, _>>()?;
191    Ok(rows)
192}
193
194/// List ALL stored graph edges, excluding invalidated ones.
195pub(crate) fn list_all_graph_edges(conn: &Connection) -> Result<Vec<StoredGraphEdge>, MemoryError> {
196    let mut stmt = conn.prepare(
197        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
198                is_invalidated, invalidated_at, invalidation_reason
199         FROM graph_edges
200         WHERE is_invalidated = 0
201         ORDER BY recorded_at ASC",
202    )?;
203    let rows = stmt
204        .query_map([], |row| {
205            Ok(StoredGraphEdge {
206                id: row.get(0)?,
207                source: row.get(1)?,
208                target: row.get(2)?,
209                edge_type: row.get(3)?,
210                edge_type_parsed: None,
211                weight: row.get(4)?,
212                metadata: row.get(5)?,
213                content_digest: row.get(6)?,
214                recorded_at: row.get(7)?,
215                is_invalidated: row.get::<_, i64>(8)? != 0,
216                invalidated_at: row.get(9)?,
217                invalidation_reason: row.get(10)?,
218            })
219        })?
220        .collect::<Result<Vec<_>, _>>()?;
221    Ok(rows)
222}
223
224/// Invalidate a graph edge by ID. Append-only — does not delete the row.
225pub(crate) fn invalidate_graph_edge(
226    conn: &Connection,
227    edge_id: &str,
228    reason: &str,
229) -> Result<(), MemoryError> {
230    let invalidated_at = Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string();
231    let count = conn.execute(
232        "UPDATE graph_edges SET is_invalidated = 1, invalidated_at = ?1, invalidation_reason = ?2
233         WHERE id = ?3 AND is_invalidated = 0",
234        params![&invalidated_at, reason, edge_id],
235    )?;
236    if count == 0 {
237        return Err(MemoryError::Other(format!(
238            "graph edge {} not found or already invalidated",
239            edge_id
240        )));
241    }
242    Ok(())
243}
244
245/// Load stored graph edges for a node and convert them to GraphEdge objects
246/// for the derived graph view. Only non-invalidated edges are included.
247pub(crate) fn stored_edges_for_node(
248    conn: &Connection,
249    node_id: &str,
250) -> Result<Vec<GraphEdge>, MemoryError> {
251    let rows = list_graph_edges_for_node(conn, node_id)?;
252    let mut edges = Vec::new();
253    for row in rows {
254        let edge_type: GraphEdgeType = serde_json::from_str(&row.edge_type)
255            .map_err(|e| MemoryError::Other(format!("failed to deserialize edge_type: {e}")))?;
256        let metadata: Option<serde_json::Value> = match &row.metadata {
257            Some(s) => Some(serde_json::from_str(s).map_err(|e| {
258                MemoryError::Other(format!("failed to deserialize metadata: {e}"))
259            })?),
260            None => None,
261        };
262        edges.push(GraphEdge {
263            source: row.source,
264            target: row.target,
265            edge_type,
266            weight: row.weight,
267            metadata,
268        });
269    }
270    Ok(edges)
271}
272
273/// Load stored outgoing edges for a node (where node is the source).
274/// Only non-invalidated edges are included.
275pub(crate) fn stored_outgoing_edges(
276    conn: &Connection,
277    node_id: &str,
278) -> Result<Vec<GraphEdge>, MemoryError> {
279    let mut stmt = conn.prepare(
280        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
281                is_invalidated, invalidated_at, invalidation_reason
282         FROM graph_edges
283         WHERE source = ?1 AND is_invalidated = 0
284         ORDER BY recorded_at ASC",
285    )?;
286    let rows = stmt
287        .query_map(params![node_id], |row| {
288            Ok(StoredGraphEdge {
289                id: row.get(0)?,
290                source: row.get(1)?,
291                target: row.get(2)?,
292                edge_type: row.get(3)?,
293                edge_type_parsed: None,
294                weight: row.get(4)?,
295                metadata: row.get(5)?,
296                content_digest: row.get(6)?,
297                recorded_at: row.get(7)?,
298                is_invalidated: row.get::<_, i64>(8)? != 0,
299                invalidated_at: row.get(9)?,
300                invalidation_reason: row.get(10)?,
301            })
302        })?
303        .collect::<Result<Vec<_>, _>>()?;
304    rows_to_graph_edges(rows)
305}
306
307/// Load stored incoming edges for a node (where node is the target).
308/// Only non-invalidated edges are included.
309pub(crate) fn stored_incoming_edges(
310    conn: &Connection,
311    node_id: &str,
312) -> Result<Vec<GraphEdge>, MemoryError> {
313    let mut stmt = conn.prepare(
314        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
315                is_invalidated, invalidated_at, invalidation_reason
316         FROM graph_edges
317         WHERE target = ?1 AND is_invalidated = 0
318         ORDER BY recorded_at ASC",
319    )?;
320    let rows = stmt
321        .query_map(params![node_id], |row| {
322            Ok(StoredGraphEdge {
323                id: row.get(0)?,
324                source: row.get(1)?,
325                target: row.get(2)?,
326                edge_type: row.get(3)?,
327                edge_type_parsed: None,
328                weight: row.get(4)?,
329                metadata: row.get(5)?,
330                content_digest: row.get(6)?,
331                recorded_at: row.get(7)?,
332                is_invalidated: row.get::<_, i64>(8)? != 0,
333                invalidated_at: row.get(9)?,
334                invalidation_reason: row.get(10)?,
335            })
336        })?
337        .collect::<Result<Vec<_>, _>>()?;
338    rows_to_graph_edges(rows)
339}
340
341/// Convert StoredGraphEdge rows into GraphEdge objects.
342fn rows_to_graph_edges(rows: Vec<StoredGraphEdge>) -> Result<Vec<GraphEdge>, MemoryError> {
343    let mut edges = Vec::new();
344    for row in rows {
345        let edge_type: GraphEdgeType = serde_json::from_str(&row.edge_type)
346            .map_err(|e| MemoryError::Other(format!("failed to deserialize edge_type: {e}")))?;
347        let metadata: Option<serde_json::Value> = match &row.metadata {
348            Some(s) => Some(serde_json::from_str(s).map_err(|e| {
349                MemoryError::Other(format!("failed to deserialize metadata: {e}"))
350            })?),
351            None => None,
352        };
353        edges.push(GraphEdge {
354            source: row.source,
355            target: row.target,
356            edge_type,
357            weight: row.weight,
358            metadata,
359        });
360    }
361    Ok(edges)
362}
363/// Count total stored edges (non-invalidated).
364pub(crate) fn count_graph_edges(conn: &Connection) -> Result<usize, MemoryError> {
365    let count: i64 = conn
366        .query_row(
367            "SELECT COUNT(*) FROM graph_edges WHERE is_invalidated = 0",
368            [],
369            |row| row.get(0),
370        )
371        .map_err(|e| MemoryError::Database(e))?;
372    Ok(count as usize)
373}
374
375fn compute_edge_digest(
376    source: &str,
377    target: &str,
378    edge_type_json: &str,
379    weight: f64,
380    metadata_json: &Option<String>,
381) -> String {
382    let mut builder = stack_ids::DigestBuilder::new();
383    builder.update(source.as_bytes());
384    builder.update(target.as_bytes());
385    builder.update(edge_type_json.as_bytes());
386    builder.update(&weight.to_le_bytes());
387    if let Some(meta) = metadata_json {
388        builder.update(meta.as_bytes());
389    }
390    builder.finalize().0
391}
semantic_memory/graph_edges.rs

semantic_memory/
graph_edges.rs
