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::{DateTime, NaiveDateTime, Utc};
29use rusqlite::{params, Connection};
30use std::collections::HashSet;
31
32/// Canonical timestamp format used for all bitemporal graph-edge columns.
33///
34/// Fixed-width, lexicographically sortable, and unambiguous.
35const CANONICAL_TS_FMT: &str = "%Y-%m-%d %H:%M:%S%.6f";
36
37/// Normalize a caller-provided timestamp into the canonical sortable format.
38///
39/// Accepts:
40/// - `%Y-%m-%d %H:%M:%S%.6f` (canonical)
41/// - `%Y-%m-%d %H:%M:%S` (SQLite default)
42/// - RFC3339 / ISO 8601 (`2026-01-01T00:00:00Z`)
43///
44/// Returns `None` if the input cannot be parsed.
45pub(crate) fn canonicalize_timestamp(s: &str) -> Option<String> {
46    // Already canonical?
47    if NaiveDateTime::parse_from_str(s, CANONICAL_TS_FMT).is_ok() {
48        return Some(s.to_string());
49    }
50    // SQLite datetime format with seconds precision.
51    if let Ok(dt) = NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S") {
52        return Some(dt.format(CANONICAL_TS_FMT).to_string());
53    }
54    // RFC3339 / ISO 8601.
55    if let Ok(dt) = DateTime::parse_from_rfc3339(s) {
56        return Some(dt.naive_utc().format(CANONICAL_TS_FMT).to_string());
57    }
58    None
59}
60
61/// Normalize a timestamp for storage. If parsing fails, returns the
62/// original input unchanged so the caller can decide whether to reject it.
63fn normalize_or_pass_through(s: &str) -> String {
64    canonicalize_timestamp(s).unwrap_or_else(|| s.to_string())
65}
66
67/// A stored graph edge row.
68#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
69pub struct StoredGraphEdge {
70    /// Unique edge ID (UUID v4).
71    pub id: String,
72    /// Source node ID (prefixed, e.g. `fact:<uuid>`).
73    pub source: String,
74    /// Target node ID (prefixed, e.g. `fact:<uuid>`).
75    pub target: String,
76    /// Serialized GraphEdgeType JSON.
77    pub edge_type: String,
78    /// Deserialized edge type.
79    #[serde(skip)]
80    pub edge_type_parsed: Option<GraphEdgeType>,
81    /// Edge weight.
82    pub weight: f64,
83    /// Optional metadata JSON.
84    pub metadata: Option<String>,
85    /// Content digest (blake3 of source+target+edge_type+weight+metadata).
86    pub content_digest: String,
87    /// Recorded timestamp (microsecond precision).
88    pub recorded_at: String,
89    /// Whether this edge has been invalidated.
90    pub is_invalidated: bool,
91    /// Invalidation timestamp.
92    pub invalidated_at: Option<String>,
93    /// Invalidation reason.
94    pub invalidation_reason: Option<String>,
95    /// Domain/business time this edge became valid.
96    pub valid_time: Option<String>,
97    /// System time this edge version was recorded.
98    pub recorded_time: Option<String>,
99}
100
101/// Parameters for creating a graph edge.
102#[derive(Debug, Clone)]
103pub struct AddGraphEdgeParams {
104    /// Source node ID (prefixed, e.g. `fact:<uuid>`, `namespace:<name>`).
105    pub source: String,
106    /// Target node ID (prefixed).
107    pub target: String,
108    /// Edge type (semantic, temporal, causal, entity).
109    pub edge_type: GraphEdgeType,
110    /// Edge weight (interpretation depends on edge_type).
111    pub weight: f64,
112    /// Optional metadata.
113    pub metadata: Option<serde_json::Value>,
114    /// Optional domain/business time. Defaults to insertion time.
115    pub valid_time: Option<String>,
116    /// Optional system recorded time. Defaults to insertion time.
117    pub recorded_time: Option<String>,
118}
119
120/// Insert a graph edge. Idempotent on content_digest.
121pub(crate) fn insert_graph_edge(
122    conn: &Connection,
123    params: &AddGraphEdgeParams,
124) -> Result<StoredGraphEdge, MemoryError> {
125    let edge_type_json = serde_json::to_string(&params.edge_type)
126        .map_err(|e| MemoryError::Other(format!("failed to serialize edge_type: {e}")))?;
127
128    let metadata_json = match &params.metadata {
129        Some(v) => Some(
130            serde_json::to_string(v)
131                .map_err(|e| MemoryError::Other(format!("failed to serialize metadata: {e}")))?,
132        ),
133        None => None,
134    };
135
136    let content_digest = compute_edge_digest(
137        &params.source,
138        &params.target,
139        &edge_type_json,
140        params.weight,
141        &metadata_json,
142        params.valid_time.as_deref(),
143        params.recorded_time.as_deref(),
144    );
145
146    // Check for existing edge with same digest (idempotent).
147    let existing: Option<(String, String)> = conn
148        .query_row(
149            "SELECT id, recorded_at FROM graph_edges WHERE content_digest = ?1 AND is_invalidated = 0",
150            params![&content_digest],
151            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
152        )
153        .ok();
154
155    if let Some((id, recorded_at)) = existing {
156        return Ok(StoredGraphEdge {
157            id,
158            source: params.source.clone(),
159            target: params.target.clone(),
160            edge_type: edge_type_json,
161            edge_type_parsed: Some(params.edge_type.clone()),
162            weight: params.weight,
163            metadata: metadata_json,
164            content_digest,
165            recorded_at: recorded_at.clone(),
166            is_invalidated: false,
167            invalidated_at: None,
168            invalidation_reason: None,
169            valid_time: params
170                .valid_time
171                .as_deref()
172                .map(normalize_or_pass_through)
173                .or_else(|| Some(recorded_at.clone())),
174            recorded_time: params
175                .recorded_time
176                .as_deref()
177                .map(normalize_or_pass_through)
178                .or_else(|| Some(recorded_at.clone())),
179        });
180    }
181
182    let id = uuid::Uuid::new_v4().to_string();
183    let recorded_at = Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string();
184    let valid_time = params
185        .valid_time
186        .as_deref()
187        .map(normalize_or_pass_through)
188        .unwrap_or_else(|| recorded_at.clone());
189    let recorded_time = params
190        .recorded_time
191        .as_deref()
192        .map(normalize_or_pass_through)
193        .unwrap_or_else(|| recorded_at.clone());
194
195    conn.execute(
196        "INSERT INTO graph_edges (id, source, target, edge_type, weight, metadata, content_digest, recorded_at, valid_time, recorded_time)
197         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
198        params![
199            &id,
200            &params.source,
201            &params.target,
202            &edge_type_json,
203            params.weight,
204            metadata_json.as_deref(),
205            &content_digest,
206            &recorded_at,
207            &valid_time,
208            &recorded_time,
209        ],
210    )
211    .map_err(|e| MemoryError::Database(e))?;
212
213    Ok(StoredGraphEdge {
214        id,
215        source: params.source.clone(),
216        target: params.target.clone(),
217        edge_type: edge_type_json,
218        edge_type_parsed: Some(params.edge_type.clone()),
219        weight: params.weight,
220        metadata: metadata_json,
221        content_digest,
222        recorded_at,
223        is_invalidated: false,
224        invalidated_at: None,
225        invalidation_reason: None,
226        valid_time: Some(valid_time),
227        recorded_time: Some(recorded_time),
228    })
229}
230
231/// List all stored graph edges involving a given node (either as source or
232/// target), excluding invalidated edges.
233pub(crate) fn list_graph_edges_for_node(
234    conn: &Connection,
235    node_id: &str,
236) -> Result<Vec<StoredGraphEdge>, MemoryError> {
237    let mut stmt = conn.prepare(
238        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
239                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
240         FROM graph_edges
241         WHERE (source = ?1 OR target = ?1) AND is_invalidated = 0
242         ORDER BY recorded_at ASC",
243    )?;
244    let rows = stmt
245        .query_map(params![node_id], |row| {
246            Ok(StoredGraphEdge {
247                id: row.get(0)?,
248                source: row.get(1)?,
249                target: row.get(2)?,
250                edge_type: row.get(3)?,
251                edge_type_parsed: None,
252                weight: row.get(4)?,
253                metadata: row.get(5)?,
254                content_digest: row.get(6)?,
255                recorded_at: row.get(7)?,
256                is_invalidated: row.get::<_, i64>(8)? != 0,
257                invalidated_at: row.get(9)?,
258                invalidation_reason: row.get(10)?,
259                valid_time: row.get(11)?,
260                recorded_time: row.get(12)?,
261            })
262        })?
263        .collect::<Result<Vec<_>, _>>()?;
264    Ok(rows)
265}
266
267/// List ALL stored graph edges, excluding invalidated ones.
268pub(crate) fn list_all_graph_edges(conn: &Connection) -> Result<Vec<StoredGraphEdge>, MemoryError> {
269    let mut stmt = conn.prepare(
270        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
271                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
272         FROM graph_edges
273         WHERE is_invalidated = 0
274         ORDER BY recorded_at ASC",
275    )?;
276    let rows = stmt
277        .query_map([], |row| {
278            Ok(StoredGraphEdge {
279                id: row.get(0)?,
280                source: row.get(1)?,
281                target: row.get(2)?,
282                edge_type: row.get(3)?,
283                edge_type_parsed: None,
284                weight: row.get(4)?,
285                metadata: row.get(5)?,
286                content_digest: row.get(6)?,
287                recorded_at: row.get(7)?,
288                is_invalidated: row.get::<_, i64>(8)? != 0,
289                invalidated_at: row.get(9)?,
290                invalidation_reason: row.get(10)?,
291                valid_time: row.get(11)?,
292                recorded_time: row.get(12)?,
293            })
294        })?
295        .collect::<Result<Vec<_>, _>>()?;
296    Ok(rows)
297}
298
299/// List up to `max_rows` stored graph edges, excluding invalidated ones.
300///
301/// The hard limit is intentionally enforced below the caller, so callers can
302/// bound in-memory growth for operationally sensitive paths while retaining
303/// the same query shape in the unlimited helper.
304pub(crate) fn list_all_graph_edges_with_limit(
305    conn: &Connection,
306    max_rows: usize,
307) -> Result<Vec<StoredGraphEdge>, MemoryError> {
308    let max_rows = max_rows.max(1) as i64;
309    let mut stmt = conn.prepare(
310        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
311                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
312         FROM graph_edges
313         WHERE is_invalidated = 0
314         ORDER BY recorded_at ASC
315         LIMIT ?1",
316    )?;
317    let rows = stmt
318        .query_map(params![max_rows], |row| {
319            Ok(StoredGraphEdge {
320                id: row.get(0)?,
321                source: row.get(1)?,
322                target: row.get(2)?,
323                edge_type: row.get(3)?,
324                edge_type_parsed: None,
325                weight: row.get(4)?,
326                metadata: row.get(5)?,
327                content_digest: row.get(6)?,
328                recorded_at: row.get(7)?,
329                is_invalidated: row.get::<_, i64>(8)? != 0,
330                invalidated_at: row.get(9)?,
331                invalidation_reason: row.get(10)?,
332                valid_time: row.get(11)?,
333                recorded_time: row.get(12)?,
334            })
335        })?
336        .collect::<Result<Vec<_>, _>>()?;
337    Ok(rows)
338}
339
340/// List up to `max_rows` graph edges involving a specific node (as source or
341/// target), excluding invalidated edges.
342pub(crate) fn list_graph_edges_for_node_with_limit(
343    conn: &Connection,
344    node_id: &str,
345    max_rows: usize,
346) -> Result<Vec<StoredGraphEdge>, MemoryError> {
347    let max_rows = max_rows.max(1) as i64;
348    let mut stmt = conn.prepare(
349        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
350                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
351         FROM graph_edges
352         WHERE (source = ?1 OR target = ?1) AND is_invalidated = 0
353         ORDER BY recorded_at ASC
354         LIMIT ?2",
355    )?;
356    let rows = stmt
357        .query_map(params![node_id, max_rows], |row| {
358            Ok(StoredGraphEdge {
359                id: row.get(0)?,
360                source: row.get(1)?,
361                target: row.get(2)?,
362                edge_type: row.get(3)?,
363                edge_type_parsed: None,
364                weight: row.get(4)?,
365                metadata: row.get(5)?,
366                content_digest: row.get(6)?,
367                recorded_at: row.get(7)?,
368                is_invalidated: row.get::<_, i64>(8)? != 0,
369                invalidated_at: row.get(9)?,
370                invalidation_reason: row.get(10)?,
371                valid_time: row.get(11)?,
372                recorded_time: row.get(12)?,
373            })
374        })?
375        .collect::<Result<Vec<_>, _>>()?;
376    Ok(rows)
377}
378
379/// List graph edges involving a node that were valid as of a domain time and
380/// known as of a recorded/system time. Unlike the live list APIs, this can
381/// return edges that are now invalidated if the invalidation happened after
382/// `as_of_recorded_time`.
383pub(crate) fn list_graph_edges_for_node_as_of(
384    conn: &Connection,
385    node_id: &str,
386    as_of_valid_time: &str,
387    as_of_recorded_time: &str,
388) -> Result<Vec<StoredGraphEdge>, MemoryError> {
389    // Normalize as-of inputs so mixed SQL/RFC3339 formats compare correctly.
390    let as_of_valid_time = canonicalize_timestamp(as_of_valid_time).ok_or_else(|| {
391        MemoryError::Other(format!(
392            "invalid as_of_valid_time timestamp: {as_of_valid_time}"
393        ))
394    })?;
395    let as_of_recorded_time = canonicalize_timestamp(as_of_recorded_time).ok_or_else(|| {
396        MemoryError::Other(format!(
397            "invalid as_of_recorded_time timestamp: {as_of_recorded_time}"
398        ))
399    })?;
400
401    let mut stmt = conn.prepare(
402        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
403                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
404         FROM graph_edges
405         WHERE (source = ?1 OR target = ?1)
406           AND COALESCE(valid_time, recorded_at) <= ?2
407           AND COALESCE(recorded_time, recorded_at) <= ?3
408           AND (is_invalidated = 0 OR invalidated_at IS NULL OR invalidated_at > ?3)
409         ORDER BY COALESCE(recorded_time, recorded_at) ASC, id ASC",
410    )?;
411    let rows = stmt
412        .query_map(
413            params![node_id, as_of_valid_time, as_of_recorded_time],
414            |row| {
415                Ok(StoredGraphEdge {
416                    id: row.get(0)?,
417                    source: row.get(1)?,
418                    target: row.get(2)?,
419                    edge_type: row.get(3)?,
420                    edge_type_parsed: None,
421                    weight: row.get(4)?,
422                    metadata: row.get(5)?,
423                    content_digest: row.get(6)?,
424                    recorded_at: row.get(7)?,
425                    is_invalidated: row.get::<_, i64>(8)? != 0,
426                    invalidated_at: row.get(9)?,
427                    invalidation_reason: row.get(10)?,
428                    valid_time: row.get(11)?,
429                    recorded_time: row.get(12)?,
430                })
431            },
432        )?
433        .collect::<Result<Vec<_>, _>>()?;
434    Ok(rows)
435}
436
437/// List graph edges within N hops of the given seed node IDs.
438///
439/// Performs a BFS expansion from the seeds, loading only edges that
440/// connect nodes already in the visited set. This avoids loading the
441/// entire graph when only a local neighborhood is needed (e.g. discord
442/// search, factor graph, graph_path).
443///
444/// `max_hops` controls the BFS depth. `max_nodes` caps the total nodes
445/// visited to prevent runaway expansion on hub nodes.
446pub(crate) fn list_graph_edges_for_neighborhood(
447    conn: &Connection,
448    seed_ids: &[String],
449    max_hops: usize,
450    max_nodes: usize,
451) -> Result<Vec<StoredGraphEdge>, MemoryError> {
452    if seed_ids.is_empty() || max_hops == 0 {
453        return Ok(Vec::new());
454    }
455
456    let mut visited: HashSet<String> = seed_ids.iter().cloned().collect();
457    let mut all_edges: Vec<StoredGraphEdge> = Vec::new();
458    let mut frontier: Vec<String> = seed_ids.iter().cloned().collect();
459
460    for _hop in 0..max_hops {
461        if frontier.is_empty() || visited.len() >= max_nodes {
462            break;
463        }
464
465        let mut next_frontier: Vec<String> = Vec::new();
466
467        for node_id in &frontier {
468            let edges = list_graph_edges_for_node(conn, node_id)?;
469            for edge in edges {
470                // Track both endpoints
471                let other = if edge.source == *node_id {
472                    &edge.target
473                } else {
474                    &edge.source
475                };
476
477                if !visited.contains(other) {
478                    visited.insert(other.clone());
479                    next_frontier.push(other.clone());
480                }
481
482                // Dedup edges by id
483                if !all_edges.iter().any(|e: &StoredGraphEdge| e.id == edge.id) {
484                    all_edges.push(edge);
485                }
486            }
487        }
488
489        frontier = next_frontier;
490        if visited.len() >= max_nodes {
491            break;
492        }
493    }
494
495    // Sort by recorded_at for deterministic ordering
496    all_edges.sort_by(|a, b| a.recorded_at.cmp(&b.recorded_at));
497    Ok(all_edges)
498}
499
500/// Invalidate a graph edge by ID. Append-only — does not delete the row.
501pub(crate) fn invalidate_graph_edge(
502    conn: &Connection,
503    edge_id: &str,
504    reason: &str,
505) -> Result<(), MemoryError> {
506    let invalidated_at =
507        normalize_or_pass_through(&Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string());
508    let count = conn.execute(
509        "UPDATE graph_edges SET is_invalidated = 1, invalidated_at = ?1, invalidation_reason = ?2
510         WHERE id = ?3 AND is_invalidated = 0",
511        params![&invalidated_at, reason, edge_id],
512    )?;
513    if count == 0 {
514        return Err(MemoryError::Other(format!(
515            "graph edge {} not found or already invalidated",
516            edge_id
517        )));
518    }
519    Ok(())
520}
521
522/// Load stored graph edges for a node and convert them to GraphEdge objects
523/// for the derived graph view. Only non-invalidated edges are included.
524#[allow(dead_code)] // public API — used by external consumers, not internally
525pub(crate) fn stored_edges_for_node(
526    conn: &Connection,
527    node_id: &str,
528) -> Result<Vec<GraphEdge>, MemoryError> {
529    let rows = list_graph_edges_for_node(conn, node_id)?;
530    let mut edges = Vec::new();
531    for row in rows {
532        let edge_type: GraphEdgeType = serde_json::from_str(&row.edge_type)
533            .map_err(|e| MemoryError::Other(format!("failed to deserialize edge_type: {e}")))?;
534        let metadata: Option<serde_json::Value> =
535            match &row.metadata {
536                Some(s) => Some(serde_json::from_str(s).map_err(|e| {
537                    MemoryError::Other(format!("failed to deserialize metadata: {e}"))
538                })?),
539                None => None,
540            };
541        edges.push(GraphEdge {
542            source: row.source,
543            target: row.target,
544            edge_type,
545            weight: row.weight,
546            metadata,
547        });
548    }
549    Ok(edges)
550}
551
552/// Load stored outgoing edges for a node (where node is the source).
553/// Only non-invalidated edges are included.
554pub(crate) fn stored_outgoing_edges(
555    conn: &Connection,
556    node_id: &str,
557) -> Result<Vec<GraphEdge>, MemoryError> {
558    let mut stmt = conn.prepare(
559        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
560                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
561         FROM graph_edges
562         WHERE source = ?1 AND is_invalidated = 0
563         ORDER BY recorded_at ASC",
564    )?;
565    let rows = stmt
566        .query_map(params![node_id], |row| {
567            Ok(StoredGraphEdge {
568                id: row.get(0)?,
569                source: row.get(1)?,
570                target: row.get(2)?,
571                edge_type: row.get(3)?,
572                edge_type_parsed: None,
573                weight: row.get(4)?,
574                metadata: row.get(5)?,
575                content_digest: row.get(6)?,
576                recorded_at: row.get(7)?,
577                is_invalidated: row.get::<_, i64>(8)? != 0,
578                invalidated_at: row.get(9)?,
579                invalidation_reason: row.get(10)?,
580                valid_time: row.get(11)?,
581                recorded_time: row.get(12)?,
582            })
583        })?
584        .collect::<Result<Vec<_>, _>>()?;
585    rows_to_graph_edges(rows)
586}
587
588/// Load stored incoming edges for a node (where node is the target).
589/// Only non-invalidated edges are included.
590pub(crate) fn stored_incoming_edges(
591    conn: &Connection,
592    node_id: &str,
593) -> Result<Vec<GraphEdge>, MemoryError> {
594    let mut stmt = conn.prepare(
595        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
596                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
597         FROM graph_edges
598         WHERE target = ?1 AND is_invalidated = 0
599         ORDER BY recorded_at ASC",
600    )?;
601    let rows = stmt
602        .query_map(params![node_id], |row| {
603            Ok(StoredGraphEdge {
604                id: row.get(0)?,
605                source: row.get(1)?,
606                target: row.get(2)?,
607                edge_type: row.get(3)?,
608                edge_type_parsed: None,
609                weight: row.get(4)?,
610                metadata: row.get(5)?,
611                content_digest: row.get(6)?,
612                recorded_at: row.get(7)?,
613                is_invalidated: row.get::<_, i64>(8)? != 0,
614                invalidated_at: row.get(9)?,
615                invalidation_reason: row.get(10)?,
616                valid_time: row.get(11)?,
617                recorded_time: row.get(12)?,
618            })
619        })?
620        .collect::<Result<Vec<_>, _>>()?;
621    rows_to_graph_edges(rows)
622}
623
624/// Convert StoredGraphEdge rows into GraphEdge objects.
625fn rows_to_graph_edges(rows: Vec<StoredGraphEdge>) -> Result<Vec<GraphEdge>, MemoryError> {
626    let mut edges = Vec::new();
627    for row in rows {
628        let edge_type: GraphEdgeType = serde_json::from_str(&row.edge_type)
629            .map_err(|e| MemoryError::Other(format!("failed to deserialize edge_type: {e}")))?;
630        let metadata: Option<serde_json::Value> =
631            match &row.metadata {
632                Some(s) => Some(serde_json::from_str(s).map_err(|e| {
633                    MemoryError::Other(format!("failed to deserialize metadata: {e}"))
634                })?),
635                None => None,
636            };
637        edges.push(GraphEdge {
638            source: row.source,
639            target: row.target,
640            edge_type,
641            weight: row.weight,
642            metadata,
643        });
644    }
645    Ok(edges)
646}
647/// Count total stored edges (non-invalidated).
648pub(crate) fn count_graph_edges(conn: &Connection) -> Result<usize, MemoryError> {
649    let count: i64 = conn
650        .query_row(
651            "SELECT COUNT(*) FROM graph_edges WHERE is_invalidated = 0",
652            [],
653            |row| row.get(0),
654        )
655        .map_err(|e| MemoryError::Database(e))?;
656    Ok(count as usize)
657}
658
659fn compute_edge_digest(
660    source: &str,
661    target: &str,
662    edge_type_json: &str,
663    weight: f64,
664    metadata_json: &Option<String>,
665    explicit_valid_time: Option<&str>,
666    explicit_recorded_time: Option<&str>,
667) -> String {
668    let mut builder = stack_ids::DigestBuilder::new();
669    builder.update(source.as_bytes());
670    builder.update(target.as_bytes());
671    builder.update(edge_type_json.as_bytes());
672    builder.update(&weight.to_le_bytes());
673    if let Some(meta) = metadata_json {
674        builder.update(meta.as_bytes());
675    }
676    if let Some(valid_time) = explicit_valid_time {
677        builder.update(valid_time.as_bytes());
678    }
679    if let Some(recorded_time) = explicit_recorded_time {
680        builder.update(recorded_time.as_bytes());
681    }
682    builder.finalize().to_string()
683}
684
685#[cfg(test)]
686mod tests {
687    use super::*;
688    use crate::db::run_migrations;
689    use rusqlite::Connection;
690    use serde_json::json;
691
692    fn migrated_conn() -> Connection {
693        let conn = Connection::open_in_memory().unwrap();
694        run_migrations(&conn).unwrap();
695        conn
696    }
697
698    #[test]
699    fn graph_edges_have_bitemporal_columns_after_migration() {
700        let conn = migrated_conn();
701        let columns: Vec<String> = conn
702            .prepare("PRAGMA table_info(graph_edges)")
703            .unwrap()
704            .query_map([], |row| row.get::<_, String>(1))
705            .unwrap()
706            .collect::<Result<Vec<_>, _>>()
707            .unwrap();
708
709        assert!(columns.contains(&"valid_time".to_string()));
710        assert!(columns.contains(&"recorded_time".to_string()));
711    }
712
713    #[test]
714    fn graph_edges_as_of_filters_valid_and_recorded_time() {
715        let conn = migrated_conn();
716        let old_edge = AddGraphEdgeParams {
717            source: "fact:a".to_string(),
718            target: "fact:b".to_string(),
719            edge_type: GraphEdgeType::Entity {
720                relation: "supports".to_string(),
721            },
722            weight: 1.0,
723            metadata: Some(json!({"version": "old"})),
724            valid_time: Some("2026-01-01T00:00:00Z".to_string()),
725            recorded_time: Some("2026-01-02T00:00:00Z".to_string()),
726        };
727        let future_edge = AddGraphEdgeParams {
728            source: "fact:a".to_string(),
729            target: "fact:c".to_string(),
730            edge_type: GraphEdgeType::Entity {
731                relation: "supports".to_string(),
732            },
733            weight: 1.0,
734            metadata: Some(json!({"version": "future"})),
735            valid_time: Some("2026-03-01T00:00:00Z".to_string()),
736            recorded_time: Some("2026-03-02T00:00:00Z".to_string()),
737        };
738
739        insert_graph_edge(&conn, &old_edge).unwrap();
740        insert_graph_edge(&conn, &future_edge).unwrap();
741
742        let visible = list_graph_edges_for_node_as_of(
743            &conn,
744            "fact:a",
745            "2026-02-01T00:00:00Z",
746            "2026-02-02T00:00:00Z",
747        )
748        .unwrap();
749
750        assert_eq!(visible.len(), 1);
751        assert_eq!(visible[0].target, "fact:b");
752        // Stored RFC3339 values must be canonicalized to fixed-width SQL format.
753        assert_eq!(
754            visible[0].valid_time.as_deref(),
755            Some("2026-01-01 00:00:00.000000")
756        );
757        assert_eq!(
758            visible[0].recorded_time.as_deref(),
759            Some("2026-01-02 00:00:00.000000")
760        );
761    }
762
763    #[test]
764    fn graph_edges_as_of_handles_mixed_timestamp_formats() {
765        // Edge inserted with canonical SQL timestamp; query with RFC3339.
766        let conn = migrated_conn();
767        let edge = AddGraphEdgeParams {
768            source: "fact:x".to_string(),
769            target: "fact:y".to_string(),
770            edge_type: GraphEdgeType::Semantic {
771                cosine_similarity: 0.9,
772            },
773            weight: 1.0,
774            metadata: None,
775            valid_time: Some("2026-01-01 00:00:00.000000".to_string()),
776            recorded_time: Some("2026-01-01 00:00:00.000000".to_string()),
777        };
778        insert_graph_edge(&conn, &edge).unwrap();
779
780        let visible = list_graph_edges_for_node_as_of(
781            &conn,
782            "fact:x",
783            "2026-01-01T23:59:59Z",
784            "2026-01-01T23:59:59Z",
785        )
786        .unwrap();
787        assert_eq!(visible.len(), 1);
788        assert_eq!(visible[0].target, "fact:y");
789    }
790
791    #[test]
792    fn graph_edges_as_of_rejects_invalid_timestamp() {
793        let conn = migrated_conn();
794        let edge = AddGraphEdgeParams {
795            source: "fact:x".to_string(),
796            target: "fact:y".to_string(),
797            edge_type: GraphEdgeType::Semantic {
798                cosine_similarity: 0.9,
799            },
800            weight: 1.0,
801            metadata: None,
802            valid_time: None,
803            recorded_time: None,
804        };
805        insert_graph_edge(&conn, &edge).unwrap();
806
807        let err = list_graph_edges_for_node_as_of(
808            &conn,
809            "fact:x",
810            "not-a-timestamp",
811            "2026-01-01T00:00:00Z",
812        )
813        .unwrap_err();
814        assert!(err.to_string().contains("invalid as_of_valid_time"));
815    }
816}
semantic_memory/graph_edges.rs

semantic_memory/
graph_edges.rs
