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 graph edges involving a node that were valid as of a domain time and
300/// known as of a recorded/system time. Unlike the live list APIs, this can
301/// return edges that are now invalidated if the invalidation happened after
302/// `as_of_recorded_time`.
303pub(crate) fn list_graph_edges_for_node_as_of(
304    conn: &Connection,
305    node_id: &str,
306    as_of_valid_time: &str,
307    as_of_recorded_time: &str,
308) -> Result<Vec<StoredGraphEdge>, MemoryError> {
309    // Normalize as-of inputs so mixed SQL/RFC3339 formats compare correctly.
310    let as_of_valid_time = canonicalize_timestamp(as_of_valid_time).ok_or_else(|| {
311        MemoryError::Other(format!(
312            "invalid as_of_valid_time timestamp: {as_of_valid_time}"
313        ))
314    })?;
315    let as_of_recorded_time = canonicalize_timestamp(as_of_recorded_time).ok_or_else(|| {
316        MemoryError::Other(format!(
317            "invalid as_of_recorded_time timestamp: {as_of_recorded_time}"
318        ))
319    })?;
320
321    let mut stmt = conn.prepare(
322        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
323                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
324         FROM graph_edges
325         WHERE (source = ?1 OR target = ?1)
326           AND COALESCE(valid_time, recorded_at) <= ?2
327           AND COALESCE(recorded_time, recorded_at) <= ?3
328           AND (is_invalidated = 0 OR invalidated_at IS NULL OR invalidated_at > ?3)
329         ORDER BY COALESCE(recorded_time, recorded_at) ASC, id ASC",
330    )?;
331    let rows = stmt
332        .query_map(
333            params![node_id, as_of_valid_time, as_of_recorded_time],
334            |row| {
335                Ok(StoredGraphEdge {
336                    id: row.get(0)?,
337                    source: row.get(1)?,
338                    target: row.get(2)?,
339                    edge_type: row.get(3)?,
340                    edge_type_parsed: None,
341                    weight: row.get(4)?,
342                    metadata: row.get(5)?,
343                    content_digest: row.get(6)?,
344                    recorded_at: row.get(7)?,
345                    is_invalidated: row.get::<_, i64>(8)? != 0,
346                    invalidated_at: row.get(9)?,
347                    invalidation_reason: row.get(10)?,
348                    valid_time: row.get(11)?,
349                    recorded_time: row.get(12)?,
350                })
351            },
352        )?
353        .collect::<Result<Vec<_>, _>>()?;
354    Ok(rows)
355}
356
357/// List graph edges within N hops of the given seed node IDs.
358///
359/// Performs a BFS expansion from the seeds, loading only edges that
360/// connect nodes already in the visited set. This avoids loading the
361/// entire graph when only a local neighborhood is needed (e.g. discord
362/// search, factor graph, graph_path).
363///
364/// `max_hops` controls the BFS depth. `max_nodes` caps the total nodes
365/// visited to prevent runaway expansion on hub nodes.
366pub(crate) fn list_graph_edges_for_neighborhood(
367    conn: &Connection,
368    seed_ids: &[String],
369    max_hops: usize,
370    max_nodes: usize,
371) -> Result<Vec<StoredGraphEdge>, MemoryError> {
372    if seed_ids.is_empty() || max_hops == 0 {
373        return Ok(Vec::new());
374    }
375
376    let mut visited: HashSet<String> = seed_ids.iter().cloned().collect();
377    let mut all_edges: Vec<StoredGraphEdge> = Vec::new();
378    let mut frontier: Vec<String> = seed_ids.iter().cloned().collect();
379
380    for _hop in 0..max_hops {
381        if frontier.is_empty() || visited.len() >= max_nodes {
382            break;
383        }
384
385        let mut next_frontier: Vec<String> = Vec::new();
386
387        for node_id in &frontier {
388            let edges = list_graph_edges_for_node(conn, node_id)?;
389            for edge in edges {
390                // Track both endpoints
391                let other = if edge.source == *node_id {
392                    &edge.target
393                } else {
394                    &edge.source
395                };
396
397                if !visited.contains(other) {
398                    visited.insert(other.clone());
399                    next_frontier.push(other.clone());
400                }
401
402                // Dedup edges by id
403                if !all_edges.iter().any(|e: &StoredGraphEdge| e.id == edge.id) {
404                    all_edges.push(edge);
405                }
406            }
407        }
408
409        frontier = next_frontier;
410        if visited.len() >= max_nodes {
411            break;
412        }
413    }
414
415    // Sort by recorded_at for deterministic ordering
416    all_edges.sort_by(|a, b| a.recorded_at.cmp(&b.recorded_at));
417    Ok(all_edges)
418}
419
420/// Invalidate a graph edge by ID. Append-only — does not delete the row.
421pub(crate) fn invalidate_graph_edge(
422    conn: &Connection,
423    edge_id: &str,
424    reason: &str,
425) -> Result<(), MemoryError> {
426    let invalidated_at =
427        normalize_or_pass_through(&Utc::now().format("%Y-%m-%d %H:%M:%S%.6f").to_string());
428    let count = conn.execute(
429        "UPDATE graph_edges SET is_invalidated = 1, invalidated_at = ?1, invalidation_reason = ?2
430         WHERE id = ?3 AND is_invalidated = 0",
431        params![&invalidated_at, reason, edge_id],
432    )?;
433    if count == 0 {
434        return Err(MemoryError::Other(format!(
435            "graph edge {} not found or already invalidated",
436            edge_id
437        )));
438    }
439    Ok(())
440}
441
442/// Load stored graph edges for a node and convert them to GraphEdge objects
443/// for the derived graph view. Only non-invalidated edges are included.
444#[allow(dead_code)] // public API — used by external consumers, not internally
445pub(crate) fn stored_edges_for_node(
446    conn: &Connection,
447    node_id: &str,
448) -> Result<Vec<GraphEdge>, MemoryError> {
449    let rows = list_graph_edges_for_node(conn, node_id)?;
450    let mut edges = Vec::new();
451    for row in rows {
452        let edge_type: GraphEdgeType = serde_json::from_str(&row.edge_type)
453            .map_err(|e| MemoryError::Other(format!("failed to deserialize edge_type: {e}")))?;
454        let metadata: Option<serde_json::Value> =
455            match &row.metadata {
456                Some(s) => Some(serde_json::from_str(s).map_err(|e| {
457                    MemoryError::Other(format!("failed to deserialize metadata: {e}"))
458                })?),
459                None => None,
460            };
461        edges.push(GraphEdge {
462            source: row.source,
463            target: row.target,
464            edge_type,
465            weight: row.weight,
466            metadata,
467        });
468    }
469    Ok(edges)
470}
471
472/// Load stored outgoing edges for a node (where node is the source).
473/// Only non-invalidated edges are included.
474pub(crate) fn stored_outgoing_edges(
475    conn: &Connection,
476    node_id: &str,
477) -> Result<Vec<GraphEdge>, MemoryError> {
478    let mut stmt = conn.prepare(
479        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
480                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
481         FROM graph_edges
482         WHERE source = ?1 AND is_invalidated = 0
483         ORDER BY recorded_at ASC",
484    )?;
485    let rows = stmt
486        .query_map(params![node_id], |row| {
487            Ok(StoredGraphEdge {
488                id: row.get(0)?,
489                source: row.get(1)?,
490                target: row.get(2)?,
491                edge_type: row.get(3)?,
492                edge_type_parsed: None,
493                weight: row.get(4)?,
494                metadata: row.get(5)?,
495                content_digest: row.get(6)?,
496                recorded_at: row.get(7)?,
497                is_invalidated: row.get::<_, i64>(8)? != 0,
498                invalidated_at: row.get(9)?,
499                invalidation_reason: row.get(10)?,
500                valid_time: row.get(11)?,
501                recorded_time: row.get(12)?,
502            })
503        })?
504        .collect::<Result<Vec<_>, _>>()?;
505    rows_to_graph_edges(rows)
506}
507
508/// Load stored incoming edges for a node (where node is the target).
509/// Only non-invalidated edges are included.
510pub(crate) fn stored_incoming_edges(
511    conn: &Connection,
512    node_id: &str,
513) -> Result<Vec<GraphEdge>, MemoryError> {
514    let mut stmt = conn.prepare(
515        "SELECT id, source, target, edge_type, weight, metadata, content_digest, recorded_at,
516                is_invalidated, invalidated_at, invalidation_reason, valid_time, recorded_time
517         FROM graph_edges
518         WHERE target = ?1 AND is_invalidated = 0
519         ORDER BY recorded_at ASC",
520    )?;
521    let rows = stmt
522        .query_map(params![node_id], |row| {
523            Ok(StoredGraphEdge {
524                id: row.get(0)?,
525                source: row.get(1)?,
526                target: row.get(2)?,
527                edge_type: row.get(3)?,
528                edge_type_parsed: None,
529                weight: row.get(4)?,
530                metadata: row.get(5)?,
531                content_digest: row.get(6)?,
532                recorded_at: row.get(7)?,
533                is_invalidated: row.get::<_, i64>(8)? != 0,
534                invalidated_at: row.get(9)?,
535                invalidation_reason: row.get(10)?,
536                valid_time: row.get(11)?,
537                recorded_time: row.get(12)?,
538            })
539        })?
540        .collect::<Result<Vec<_>, _>>()?;
541    rows_to_graph_edges(rows)
542}
543
544/// Convert StoredGraphEdge rows into GraphEdge objects.
545fn rows_to_graph_edges(rows: Vec<StoredGraphEdge>) -> Result<Vec<GraphEdge>, MemoryError> {
546    let mut edges = Vec::new();
547    for row in rows {
548        let edge_type: GraphEdgeType = serde_json::from_str(&row.edge_type)
549            .map_err(|e| MemoryError::Other(format!("failed to deserialize edge_type: {e}")))?;
550        let metadata: Option<serde_json::Value> =
551            match &row.metadata {
552                Some(s) => Some(serde_json::from_str(s).map_err(|e| {
553                    MemoryError::Other(format!("failed to deserialize metadata: {e}"))
554                })?),
555                None => None,
556            };
557        edges.push(GraphEdge {
558            source: row.source,
559            target: row.target,
560            edge_type,
561            weight: row.weight,
562            metadata,
563        });
564    }
565    Ok(edges)
566}
567/// Count total stored edges (non-invalidated).
568pub(crate) fn count_graph_edges(conn: &Connection) -> Result<usize, MemoryError> {
569    let count: i64 = conn
570        .query_row(
571            "SELECT COUNT(*) FROM graph_edges WHERE is_invalidated = 0",
572            [],
573            |row| row.get(0),
574        )
575        .map_err(|e| MemoryError::Database(e))?;
576    Ok(count as usize)
577}
578
579fn compute_edge_digest(
580    source: &str,
581    target: &str,
582    edge_type_json: &str,
583    weight: f64,
584    metadata_json: &Option<String>,
585    explicit_valid_time: Option<&str>,
586    explicit_recorded_time: Option<&str>,
587) -> String {
588    let mut builder = stack_ids::DigestBuilder::new();
589    builder.update(source.as_bytes());
590    builder.update(target.as_bytes());
591    builder.update(edge_type_json.as_bytes());
592    builder.update(&weight.to_le_bytes());
593    if let Some(meta) = metadata_json {
594        builder.update(meta.as_bytes());
595    }
596    if let Some(valid_time) = explicit_valid_time {
597        builder.update(valid_time.as_bytes());
598    }
599    if let Some(recorded_time) = explicit_recorded_time {
600        builder.update(recorded_time.as_bytes());
601    }
602    builder.finalize().0
603}
604
605#[cfg(test)]
606mod tests {
607    use super::*;
608    use crate::db::run_migrations;
609    use rusqlite::Connection;
610    use serde_json::json;
611
612    fn migrated_conn() -> Connection {
613        let conn = Connection::open_in_memory().unwrap();
614        run_migrations(&conn).unwrap();
615        conn
616    }
617
618    #[test]
619    fn graph_edges_have_bitemporal_columns_after_migration() {
620        let conn = migrated_conn();
621        let columns: Vec<String> = conn
622            .prepare("PRAGMA table_info(graph_edges)")
623            .unwrap()
624            .query_map([], |row| row.get::<_, String>(1))
625            .unwrap()
626            .collect::<Result<Vec<_>, _>>()
627            .unwrap();
628
629        assert!(columns.contains(&"valid_time".to_string()));
630        assert!(columns.contains(&"recorded_time".to_string()));
631    }
632
633    #[test]
634    fn graph_edges_as_of_filters_valid_and_recorded_time() {
635        let conn = migrated_conn();
636        let old_edge = AddGraphEdgeParams {
637            source: "fact:a".to_string(),
638            target: "fact:b".to_string(),
639            edge_type: GraphEdgeType::Entity {
640                relation: "supports".to_string(),
641            },
642            weight: 1.0,
643            metadata: Some(json!({"version": "old"})),
644            valid_time: Some("2026-01-01T00:00:00Z".to_string()),
645            recorded_time: Some("2026-01-02T00:00:00Z".to_string()),
646        };
647        let future_edge = AddGraphEdgeParams {
648            source: "fact:a".to_string(),
649            target: "fact:c".to_string(),
650            edge_type: GraphEdgeType::Entity {
651                relation: "supports".to_string(),
652            },
653            weight: 1.0,
654            metadata: Some(json!({"version": "future"})),
655            valid_time: Some("2026-03-01T00:00:00Z".to_string()),
656            recorded_time: Some("2026-03-02T00:00:00Z".to_string()),
657        };
658
659        insert_graph_edge(&conn, &old_edge).unwrap();
660        insert_graph_edge(&conn, &future_edge).unwrap();
661
662        let visible = list_graph_edges_for_node_as_of(
663            &conn,
664            "fact:a",
665            "2026-02-01T00:00:00Z",
666            "2026-02-02T00:00:00Z",
667        )
668        .unwrap();
669
670        assert_eq!(visible.len(), 1);
671        assert_eq!(visible[0].target, "fact:b");
672        // Stored RFC3339 values must be canonicalized to fixed-width SQL format.
673        assert_eq!(
674            visible[0].valid_time.as_deref(),
675            Some("2026-01-01 00:00:00.000000")
676        );
677        assert_eq!(
678            visible[0].recorded_time.as_deref(),
679            Some("2026-01-02 00:00:00.000000")
680        );
681    }
682
683    #[test]
684    fn graph_edges_as_of_handles_mixed_timestamp_formats() {
685        // Edge inserted with canonical SQL timestamp; query with RFC3339.
686        let conn = migrated_conn();
687        let edge = AddGraphEdgeParams {
688            source: "fact:x".to_string(),
689            target: "fact:y".to_string(),
690            edge_type: GraphEdgeType::Semantic {
691                cosine_similarity: 0.9,
692            },
693            weight: 1.0,
694            metadata: None,
695            valid_time: Some("2026-01-01 00:00:00.000000".to_string()),
696            recorded_time: Some("2026-01-01 00:00:00.000000".to_string()),
697        };
698        insert_graph_edge(&conn, &edge).unwrap();
699
700        let visible = list_graph_edges_for_node_as_of(
701            &conn,
702            "fact:x",
703            "2026-01-01T23:59:59Z",
704            "2026-01-01T23:59:59Z",
705        )
706        .unwrap();
707        assert_eq!(visible.len(), 1);
708        assert_eq!(visible[0].target, "fact:y");
709    }
710
711    #[test]
712    fn graph_edges_as_of_rejects_invalid_timestamp() {
713        let conn = migrated_conn();
714        let edge = AddGraphEdgeParams {
715            source: "fact:x".to_string(),
716            target: "fact:y".to_string(),
717            edge_type: GraphEdgeType::Semantic {
718                cosine_similarity: 0.9,
719            },
720            weight: 1.0,
721            metadata: None,
722            valid_time: None,
723            recorded_time: None,
724        };
725        insert_graph_edge(&conn, &edge).unwrap();
726
727        let err = list_graph_edges_for_node_as_of(
728            &conn,
729            "fact:x",
730            "not-a-timestamp",
731            "2026-01-01T00:00:00Z",
732        )
733        .unwrap_err();
734        assert!(err.to_string().contains("invalid as_of_valid_time"));
735    }
736}
semantic_memory/graph_edges.rs

semantic_memory/
graph_edges.rs
