diff --git a/.gitignore b/.gitignore index 020cc3e..37f73b5 100644 --- a/.gitignore +++ b/.gitignore @@ -34,3 +34,6 @@ logs/ # Coverage /coverage/ tarpaulin-report.html + +# fastembed model cache (downloaded at runtime, not source) +.fastembed_cache/ diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index e681157..0a588ab 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -14,7 +14,7 @@ mod installer; use anyhow::{anyhow, Context, Result}; -use chrono::Utc; +use chrono::{DateTime, Utc}; use clap::{Parser, Subcommand}; use installer::{Action, SkillBundle, Tool}; use memory_core::{MemoryItem, MemoryScope}; @@ -121,6 +121,11 @@ enum Cmd { agent: Option, #[arg(long)] session: Option, + /// RFC3339 timestamp of when this happened in the world (event time), + /// as opposed to now (ingest time). Behavior mining buckets patterns by + /// this, so set it when observing back-dated text. + #[arg(long, value_name = "RFC3339")] + occurred_at: Option>, /// Print structured JSON of what was saved (machine-readable mode /// for hooks). Without it, prints a human-readable summary. #[arg(long)] @@ -493,6 +498,7 @@ async fn main() -> Result<()> { user, agent, session, + occurred_at, json, } => { let text = match content { @@ -518,6 +524,7 @@ async fn main() -> Result<()> { user_id: Some(user.unwrap_or_else(detect_os_user)), agent_id: agent, session_id: session, + occurred_at, }; memory_storage::quota::ensure_under_cap(store.as_ref(), license.cap()) .await diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs index 1a33fe6..cb5af5c 100644 --- a/crates/core/src/config.rs +++ b/crates/core/src/config.rs @@ -16,9 +16,12 @@ use serde::{Deserialize, Serialize}; use std::path::PathBuf; -/// Default entry cap for the free tier. Hard cap on `clawdbot_memory_item` -/// row count; new saves are rejected once reached. Confirmed 2026-05-18. -pub const DEFAULT_FREE_TIER_CAP: i64 = 500; +/// Default entry cap for the local engine. `0` (or any non-positive value) +/// means **unlimited** — the open-source engine imposes no ceiling on +/// `clawdbot_memory_item` row count. A positive value re-enables a hard cap +/// and is intended for commercial embedders (e.g. desktop plan tiers) that +/// set it explicitly via config or `THINKFLEET_FREE_TIER_CAP`. +pub const DEFAULT_FREE_TIER_CAP: i64 = 0; #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct Config { @@ -74,7 +77,8 @@ pub enum BindingPolicy { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct FreeTierConfig { /// Max `clawdbot_memory_item` rows the engine will accept in local-only - /// mode. Confirmed 2026-05-18 to be 500. + /// mode. `0` (the default) = unlimited; a positive value enables a hard + /// cap for commercial embedders. #[serde(default = "default_free_tier_cap")] pub entry_cap: i64, } diff --git a/crates/license/src/lib.rs b/crates/license/src/lib.rs index eca0c50..40852ae 100644 --- a/crates/license/src/lib.rs +++ b/crates/license/src/lib.rs @@ -54,10 +54,12 @@ MCowBQYDK2VwAyEASeFn7v8VS02tyi2XCaeSjzn8WUXki6ksGI8S8lclcgg= /// the customer's, but they can't grow it further. pub const GRACE_PERIOD_SECS: i64 = 7 * 24 * 60 * 60; // 7 days -/// Free-tier cap when no license token is loaded. Matches the -/// settled local-desktop default (2026-05-18) and the existing -/// `quota::effective_cap` free-tier value. -pub const FREE_TIER_MEMORY_CAP: u64 = 500; +/// Free-tier cap when no license token is loaded. `u64::MAX` is the +/// "unlimited" sentinel (see [`License::cap`], which maps it to `None`): +/// the open-source engine imposes no write ceiling. Commercial embedders +/// that want a tiered cap ship a signed license whose `memory_cap` claim +/// carries a finite value. +pub const FREE_TIER_MEMORY_CAP: u64 = u64::MAX; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -329,8 +331,8 @@ impl License { } /// Cap as an `Option` matching the existing `quota::ensure_under_cap` - /// signature. `None` = unlimited. The free-tier u64 cap converts directly - /// to i64 because `FREE_TIER_MEMORY_CAP` is well under `i64::MAX`. + /// signature. `None` = unlimited — returned both for the `u64::MAX` + /// sentinel (the free-tier default) and any claim above `i64::MAX`. pub fn cap(&self) -> Option { match self.claims.memory_cap { u64::MAX => None, @@ -528,7 +530,7 @@ mod tests { let now = Utc::now(); let lic = License::free(now); assert_eq!(lic.tier(), LicenseTier::Free); - assert_eq!(lic.cap(), Some(FREE_TIER_MEMORY_CAP as i64)); + assert_eq!(lic.cap(), None); // free tier is uncapped (unlimited) assert!(lic.allows_writes()); } @@ -550,7 +552,7 @@ mod tests { let now = Utc::now(); let lic = License::from_token("not a real jwt", now); assert_eq!(lic.tier(), LicenseTier::Free); - assert_eq!(lic.cap(), Some(FREE_TIER_MEMORY_CAP as i64)); + assert_eq!(lic.cap(), None); // free tier is uncapped (unlimited) } #[test] diff --git a/crates/mcp/src/lib.rs b/crates/mcp/src/lib.rs index 1258ca1..0069a00 100644 --- a/crates/mcp/src/lib.rs +++ b/crates/mcp/src/lib.rs @@ -207,7 +207,8 @@ fn tool_definitions() -> serde_json::Value { "projectId": { "type": ["string", "null"], "description": "Current project (git repo name is a good default)." }, "userId": { "type": ["string", "null"], "description": "Defaults to the OS username." }, "agentId": { "type": ["string", "null"] }, - "sessionId": { "type": ["string", "null"] } + "sessionId": { "type": ["string", "null"] }, + "occurredAt": { "type": ["string", "null"], "description": "RFC3339 timestamp of when this happened IN THE WORLD, not when you're recording it. Defaults to now. Set it when observing anything back-dated — behavior mining buckets patterns by this timestamp." } } } }, @@ -229,6 +230,7 @@ fn tool_definitions() -> serde_json::Value { "scope": { "type": "string", "enum": ["platform","project","location","agent","user","session"] }, "importance": { "type": "number", "description": "0-10, default 5." }, "confidence": { "type": "number", "description": "0-1, default 1.0." }, + "occurredAt": { "type": ["string", "null"], "description": "RFC3339 timestamp of when this happened IN THE WORLD, as opposed to when you're recording it. Defaults to now. Set this whenever you're recording something back-dated (importing history, logging a past event) — behavior mining buckets patterns by this timestamp, so leaving it unset makes every backfilled event look like it happened at import time." }, "metadata": { "type": ["object", "null"] } } } @@ -270,7 +272,8 @@ fn tool_definitions() -> serde_json::Value { "platformId": { "type": ["string", "null"] }, "projectId": { "type": ["string", "null"] }, "scope": { "type": ["string", "null"] }, - "limit": { "type": "integer", "default": 20 } + "limit": { "type": "integer", "default": 20 }, + "offset": { "type": "integer", "default": 0, "description": "Skip this many rows — page by bumping it. The handler has always honored this; it was just missing from the schema, so agents had no way to know they could page." } } } }, @@ -730,6 +733,21 @@ fn memory_item_from_args(args: &serde_json::Value) -> anyhow::Result item.metadata = v.clone(); } } + // Event time. `MemoryItem::new` stamps `valid_from = now` (ingest time), + // which is right for "I just learned this" and wrong for anything + // back-dated. `valid_from` is the timestamp behavior mining buckets on, so + // without this a backfill produces patterns describing the import run + // rather than the events. `validFrom` is accepted as an alias for callers + // that speak the storage field name. + if let Some(ts) = args + .get("occurredAt") + .or_else(|| args.get("validFrom")) + .and_then(|v| v.as_str()) + { + item.valid_from = ts + .parse::>() + .map_err(|e| anyhow::anyhow!("occurredAt must be an RFC3339 timestamp: {e}"))?; + } Ok(item) } diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index a461094..b8eef6e 100644 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -20,6 +20,7 @@ pub mod graph_extractor; pub mod observe; pub mod query; pub mod quota; +pub mod temporal; pub mod validate; #[cfg(feature = "postgres")] @@ -193,6 +194,16 @@ pub trait Storage: Send + Sync + 'static { /// List edges matching the filter (subject/object/predicate + scope). async fn query_edges(&self, q: &EdgeFilter) -> Result, StorageError>; + /// Close an open edge at `at` (set `validTo = at` where it's currently + /// NULL). Bi-temporal supersession: the edge is retained for history — + /// it just stops being "current", so an `as_of` query before `at` still + /// returns it. No-op if the edge is missing or already closed. + async fn invalidate_edge( + &self, + id: &str, + at: chrono::DateTime, + ) -> Result<(), StorageError>; + // ── project_bindings ──────────────────────────────────── // // Per-machine mapping from a local cwd to a SaaS (platformId, projectId). diff --git a/crates/storage/src/observe.rs b/crates/storage/src/observe.rs index 71f7e39..5727cf6 100644 --- a/crates/storage/src/observe.rs +++ b/crates/storage/src/observe.rs @@ -7,7 +7,7 @@ //! schema fields the extractor doesn't know about, and persist. Shared by //! the MCP tool, the HTTP endpoint, and the CLI subcommand. -use chrono::Utc; +use chrono::{DateTime, Utc}; use memory_core::{ extraction::{extract, ObserveContext, ObserveRole}, MemoryItem, MemoryScope, MemoryStatus, @@ -34,6 +34,13 @@ pub struct ObserveRequest { pub agent_id: Option, #[serde(default)] pub session_id: Option, + /// When this happened *in the world*, as opposed to when it was observed. + /// Defaults to now. Drives `valid_from` on every item extracted from this + /// text — which is the timestamp behavior mining buckets on, so a backfill + /// that leaves this unset produces patterns describing the import run + /// rather than the events. Accepts `validFrom` as an alias. + #[serde(default, alias = "validFrom")] + pub occurred_at: Option>, } #[derive(Debug, Clone, Serialize)] @@ -119,7 +126,10 @@ pub async fn observe( confirmed_by_user_id: None, confirmed_at: None, negative_rating_count: 0, - valid_from: now, + // Event time, defaulting to ingest time. `learned_at` stays `now` + // regardless — that's the bi-temporal split: when it became true vs. + // when we found out. + valid_from: req.occurred_at.unwrap_or(now), valid_to: None, learned_at: now, last_accessed_at: now, diff --git a/crates/storage/src/postgres.rs b/crates/storage/src/postgres.rs index 4c8f2e4..cf68c09 100644 --- a/crates/storage/src/postgres.rs +++ b/crates/storage/src/postgres.rs @@ -1010,7 +1010,20 @@ impl Storage for PostgresStore { sql.push_str(&format!(" AND predicate = ${n}")); binds.push(s.clone()); } - if q.current_only { + // Point-in-time takes precedence over current_only. Cast the text + // binds to timestamptz so the comparison is temporal, not lexical. + if let Some(at) = q.as_of { + let at_s = at.to_rfc3339(); + n += 1; + let a = n; + binds.push(at_s.clone()); + n += 1; + let b = n; + binds.push(at_s); + sql.push_str(&format!( + " AND \"validFrom\" <= ${a}::timestamptz AND (\"validTo\" IS NULL OR \"validTo\" > ${b}::timestamptz)" + )); + } else if q.current_only { sql.push_str(" AND \"validTo\" IS NULL"); } sql.push_str(" ORDER BY updated DESC"); @@ -1028,6 +1041,17 @@ impl Storage for PostgresStore { rows.into_iter().map(row_to_edge).collect() } + async fn invalidate_edge(&self, id: &str, at: DateTime) -> Result<(), StorageError> { + sqlx::query( + r#"UPDATE memory_edge SET "validTo" = $1, updated = now() WHERE id = $2 AND "validTo" IS NULL"#, + ) + .bind(at) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + async fn save_binding(&self, binding: &ProjectBinding) -> Result<(), StorageError> { sqlx::query( r#" diff --git a/crates/storage/src/query.rs b/crates/storage/src/query.rs index 25ce441..18fbee8 100644 --- a/crates/storage/src/query.rs +++ b/crates/storage/src/query.rs @@ -138,7 +138,13 @@ pub struct EdgeFilter { pub subject_id: Option, pub object_id: Option, pub predicate: Option, + /// Only edges still open (`validTo IS NULL`). Superseded by `as_of` + /// when that is set (a point in time is more specific than "now"). pub current_only: bool, + /// Point-in-time query: return edges that were valid AT this instant — + /// `validFrom <= as_of AND (validTo IS NULL OR validTo > as_of)`. This is + /// what makes the graph a *temporal* KG: "what did we believe on date X". + pub as_of: Option>, pub limit: Option, pub offset: Option, } diff --git a/crates/storage/src/quota.rs b/crates/storage/src/quota.rs index 02fae3f..85774de 100644 --- a/crates/storage/src/quota.rs +++ b/crates/storage/src/quota.rs @@ -1,14 +1,17 @@ // Copyright 2026 ThinkFleet, Inc. Licensed under the Apache License, Version 2.0. -//! Free-tier entry cap enforcement. +//! Entry cap enforcement. //! -//! In local-only mode the engine accepts up to `Config::free_tier.entry_cap` -//! memory items (default 500, confirmed 2026-05-18). In SaaS-connected mode -//! the cap is lifted — the SaaS-side plan tier governs quota instead. +//! The open-source engine is **uncapped by default**: `free_tier.entry_cap` +//! defaults to `0` (unlimited), so `effective_cap` returns `None` and every +//! write is accepted. A cap only exists when a commercial embedder opts in — +//! by setting a positive `free_tier.entry_cap` (or `THINKFLEET_FREE_TIER_CAP`) +//! for desktop plan tiers, or in SaaS-connected mode where the plan tier +//! governs quota server-side instead. //! //! Every write path (observe, explicit save in CLI / MCP / HTTP) calls -//! `ensure_under_cap` first. Reaching the cap is a soft failure: the write -//! is rejected with a clear message; existing memories are untouched. +//! `ensure_under_cap` first. When a cap is set, reaching it is a soft failure: +//! the write is rejected with a clear message; existing memories are untouched. use crate::{Storage, StorageError}; @@ -19,9 +22,8 @@ use crate::{Storage, StorageError}; pub enum QuotaError { #[error( "memory cap reached: {count} memories at cap {cap}. \ - Upgrade your plan at https://memmesh.ai to lift the \ - cap, or delete some memories to make room \ - (`memmesh delete `)." + Raise `free_tier.entry_cap` (set it to 0 for unlimited), or \ + delete some memories to make room (`memmesh delete `)." )] CapReached { count: i64, cap: i64 }, @@ -48,15 +50,19 @@ pub async fn ensure_under_cap( Ok(()) } -/// Resolve the effective cap from a config. SaaS-connected mode → `None` -/// (no cap). Local-only mode → `Some(free_tier.entry_cap)`. +/// Resolve the effective cap from a config. `None` = unlimited. /// -/// This is the helper the write paths use so they don't have to repeat the -/// "is the engine in SaaS mode?" check. +/// SaaS-connected mode → `None` (the server-side plan tier governs quota). +/// Local mode → `None` unless `free_tier.entry_cap` is a positive value, in +/// which case that cap applies. The default `entry_cap` is `0` (unlimited), +/// so the open-source engine is uncapped; a positive value is an explicit +/// opt-in for commercial embedders. pub fn effective_cap(config: &memory_core::config::Config) -> Option { if config.is_saas_configured() { - None - } else { - Some(config.free_tier.entry_cap) + return None; + } + match config.free_tier.entry_cap { + cap if cap > 0 => Some(cap), + _ => None, } } diff --git a/crates/storage/src/sqlite.rs b/crates/storage/src/sqlite.rs index bdd8083..d6d9058 100644 --- a/crates/storage/src/sqlite.rs +++ b/crates/storage/src/sqlite.rs @@ -962,7 +962,14 @@ impl Storage for SqliteStore { sql.push_str(" AND predicate = ?"); binds.push(s.clone()); } - if q.current_only { + // Point-in-time takes precedence over current_only: an edge was valid + // AT `as_of` iff it had started and had not yet been closed. + if let Some(at) = q.as_of { + let at = iso(at); + sql.push_str(" AND validFrom <= ? AND (validTo IS NULL OR validTo > ?)"); + binds.push(at.clone()); + binds.push(at); + } else if q.current_only { sql.push_str(" AND validTo IS NULL"); } sql.push_str(" ORDER BY updated DESC"); @@ -980,6 +987,18 @@ impl Storage for SqliteStore { rows.into_iter().map(row_to_edge).collect() } + async fn invalidate_edge(&self, id: &str, at: DateTime) -> Result<(), StorageError> { + sqlx::query( + "UPDATE memory_edge SET validTo = ?, updated = ? WHERE id = ? AND validTo IS NULL", + ) + .bind(iso(at)) + .bind(iso(Utc::now())) + .bind(id) + .execute(&self.pool) + .await?; + Ok(()) + } + async fn save_binding(&self, binding: &ProjectBinding) -> Result<(), StorageError> { sqlx::query( r#" diff --git a/crates/storage/src/temporal.rs b/crates/storage/src/temporal.rs new file mode 100644 index 0000000..937300d --- /dev/null +++ b/crates/storage/src/temporal.rs @@ -0,0 +1,292 @@ +// Copyright 2026 ThinkFleet, Inc. Licensed under the Apache License, Version 2.0. + +//! Temporal knowledge-graph primitives. +//! +//! Edges are bi-temporal: each carries `valid_from` / `valid_to`, so the graph +//! can answer "what did we believe on date X" (via [`EdgeFilter::as_of`]) — not +//! just "what is true now". What was missing was *maintaining* that timeline: +//! when a subject's single-valued fact changes ("lives in Boston" → "lives in +//! Denver"), the old edge must be *closed* at the moment the new one opens, so +//! it stays in history but stops being current. +//! +//! [`supersede_conflicting_edges`] does exactly that: it's the edge-level +//! analogue of belief revision (which supersedes contradicted memory *items*). +//! Given a newly-asserted edge for a functional predicate, it closes any +//! currently-open edge with the same subject + predicate but a different +//! object, at the new edge's `valid_from`. The result is a contiguous, +//! non-overlapping timeline per (subject, predicate): exactly one edge is +//! current, and every past belief remains queryable at its own `as_of`. + +use std::env; + +use memory_core::graph::MemoryEdge; + +use crate::query::EdgeFilter; +use crate::{Storage, StorageError}; + +/// Auto-supersession on edge writes is gated: the timeline maintenance only +/// runs when `MEMORY_TEMPORAL_SUPERSEDE_ENABLED=true`, so enabling it is an +/// explicit, reversible decision (default off = no behavior change). +pub fn supersede_enabled() -> bool { + env::var("MEMORY_TEMPORAL_SUPERSEDE_ENABLED") + .ok() + .as_deref() + == Some("true") +} + +/// Whether `predicate` is FUNCTIONAL (single-valued: a subject has one current +/// value, so a new value supersedes the old). Only these are auto-superseded — +/// multi-valued predicates (`likes`, `invested_in`, `attended`) legitimately +/// have many objects and must never be collapsed. Override the set via +/// `MEMORY_TEMPORAL_FUNCTIONAL_PREDICATES` (comma-separated); the default is +/// deliberately conservative. +pub fn is_functional_predicate(predicate: &str) -> bool { + match env::var("MEMORY_TEMPORAL_FUNCTIONAL_PREDICATES") { + Ok(list) if !list.trim().is_empty() => { + list.split(',').map(str::trim).any(|p| p == predicate) + } + _ => matches!( + predicate, + "works_at" | "lives_in" | "located_in" | "based_in" | "reports_to" | "headquartered_in" + ), + } +} + +/// Same object target? Edges point at either an entity id or a literal +/// (never both), so equality is "same id AND same literal". +fn same_object(a: &MemoryEdge, b: &MemoryEdge) -> bool { + a.object_id == b.object_id && a.object_literal == b.object_literal +} + +/// True if an OPEN edge with the same (platform, project, subject, predicate, +/// object) already exists. Used to dedup re-asserted edges and cross-extractor +/// overlap — the regex and LLM passes independently extract "Alice works at +/// Acme" and would otherwise write two identical edges. Idempotent extraction. +pub async fn open_duplicate_exists( + storage: &S, + edge: &MemoryEdge, +) -> Result { + let open = storage + .query_edges(&EdgeFilter { + platform_id: Some(edge.platform_id.clone()), + project_id: edge.project_id.clone(), + subject_id: Some(edge.subject_id.clone()), + predicate: Some(edge.predicate.clone()), + current_only: true, + ..Default::default() + }) + .await?; + Ok(open.iter().any(|x| x.id != edge.id && same_object(x, edge))) +} + +/// Temporally supersede any open edge that conflicts with `new_edge` — same +/// (platform, project, subject, predicate) but a different object — by closing +/// it at `new_edge.valid_from`. Call this for FUNCTIONAL (single-valued) +/// predicates only (`livesIn`, `worksAt`, `status`), never multi-valued ones +/// (`likes`, `tagged`) where several objects legitimately coexist. +/// +/// Idempotent: re-asserting the same object closes nothing. Returns the ids of +/// the edges it closed (empty when there was no conflict). Does NOT save +/// `new_edge` — the caller owns that, so this composes with any write path. +pub async fn supersede_conflicting_edges( + storage: &S, + new_edge: &MemoryEdge, +) -> Result, StorageError> { + let open = storage + .query_edges(&EdgeFilter { + platform_id: Some(new_edge.platform_id.clone()), + project_id: new_edge.project_id.clone(), + subject_id: Some(new_edge.subject_id.clone()), + predicate: Some(new_edge.predicate.clone()), + current_only: true, + ..Default::default() + }) + .await?; + + let mut closed = Vec::new(); + for e in open { + // Don't close the new edge against itself, and leave a re-assertion + // of the same fact alone (idempotent). + if e.id == new_edge.id || same_object(&e, new_edge) { + continue; + } + storage.invalidate_edge(&e.id, new_edge.valid_from).await?; + closed.push(e.id); + } + Ok(closed) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::sqlite::SqliteStore; + use chrono::{Duration, Utc}; + + #[test] + fn functional_predicate_default_set() { + // Conservative default: single-valued relations supersede; multi-valued + // ones (many objects legitimately coexist) never do. + std::env::remove_var("MEMORY_TEMPORAL_FUNCTIONAL_PREDICATES"); + assert!(is_functional_predicate("works_at")); + assert!(is_functional_predicate("lives_in")); + assert!(!is_functional_predicate("invested_in")); + assert!(!is_functional_predicate("attended")); + assert!(!is_functional_predicate("likes")); + } + use memory_core::graph::MemoryEntity; + use memory_core::MemoryScope; + + fn entity(id: &str) -> MemoryEntity { + let now = Utc::now(); + MemoryEntity { + id: id.to_string(), + created: now, + updated: now, + platform_id: "p".into(), + project_id: Some("proj".into()), + location_id: None, + chatbot_id: None, + chat_identity_id: None, + scope: MemoryScope::Project, + type_: "subject".into(), + canonical_name: id.to_string(), + aliases: vec![], + description: None, + metadata: serde_json::Value::Null, + valid_from: now, + valid_to: None, + superseded_by_id: None, + } + } + + fn literal_edge(id: &str, subject: &str, predicate: &str, object: &str) -> MemoryEdge { + let now = Utc::now(); + MemoryEdge { + id: id.to_string(), + created: now, + updated: now, + platform_id: "p".into(), + project_id: Some("proj".into()), + location_id: None, + chatbot_id: None, + chat_identity_id: None, + scope: MemoryScope::Project, + subject_id: subject.into(), + predicate: predicate.into(), + object_id: None, + object_literal: Some(object.into()), + weight: 1.0, + source_memory_id: None, + metadata: serde_json::Value::Null, + valid_from: now, + valid_to: None, + } + } + + async fn store() -> SqliteStore { + let s = SqliteStore::connect("sqlite::memory:").await.unwrap(); + s.migrate().await.unwrap(); + s + } + + /// Seed the subject entity an edge's FK requires. + async fn seed_subject(s: &SqliteStore, id: &str) { + s.save_entity(&entity(id)).await.unwrap(); + } + + #[tokio::test] + async fn supersedes_on_changed_object() { + let s = store().await; + seed_subject(&s, "alice").await; + let mut old = literal_edge("e-old", "alice", "livesIn", "Boston"); + old.valid_from = Utc::now() - Duration::days(10); + s.save_edge(&old).await.unwrap(); + + let new = literal_edge("e-new", "alice", "livesIn", "Denver"); + let closed = supersede_conflicting_edges(&s, &new).await.unwrap(); + assert_eq!(closed, vec!["e-old".to_string()]); + s.save_edge(&new).await.unwrap(); + + // Exactly one current edge, and it's the new one. + let current = s + .query_edges(&EdgeFilter { + subject_id: Some("alice".into()), + predicate: Some("livesIn".into()), + current_only: true, + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(current.len(), 1); + assert_eq!(current[0].id, "e-new"); + } + + #[tokio::test] + async fn open_duplicate_detected() { + let s = store().await; + seed_subject(&s, "alice").await; + let e = literal_edge("e1", "alice", "livesIn", "Boston"); + s.save_edge(&e).await.unwrap(); + // Same subject+predicate+object, different id → duplicate. + let dup = literal_edge("e2", "alice", "livesIn", "Boston"); + assert!(open_duplicate_exists(&s, &dup).await.unwrap()); + // Different object → not a duplicate. + let diff = literal_edge("e3", "alice", "livesIn", "Denver"); + assert!(!open_duplicate_exists(&s, &diff).await.unwrap()); + } + + #[tokio::test] + async fn point_in_time_returns_past_belief() { + let s = store().await; + seed_subject(&s, "alice").await; + let t0 = Utc::now() - Duration::days(10); + let t_switch = Utc::now() - Duration::days(3); + + let mut old = literal_edge("e-old", "alice", "livesIn", "Boston"); + old.valid_from = t0; + s.save_edge(&old).await.unwrap(); + + let mut new = literal_edge("e-new", "alice", "livesIn", "Denver"); + new.valid_from = t_switch; + supersede_conflicting_edges(&s, &new).await.unwrap(); + s.save_edge(&new).await.unwrap(); + + // As of 5 days ago (before the switch): we believed Boston. + let past = s + .query_edges(&EdgeFilter { + subject_id: Some("alice".into()), + predicate: Some("livesIn".into()), + as_of: Some(Utc::now() - Duration::days(5)), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(past.len(), 1); + assert_eq!(past[0].id, "e-old"); + + // As of now: Denver. + let present = s + .query_edges(&EdgeFilter { + subject_id: Some("alice".into()), + predicate: Some("livesIn".into()), + as_of: Some(Utc::now()), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(present.len(), 1); + assert_eq!(present[0].id, "e-new"); + } + + #[tokio::test] + async fn idempotent_on_same_object() { + let s = store().await; + seed_subject(&s, "acme").await; + let e = literal_edge("e1", "acme", "status", "active"); + s.save_edge(&e).await.unwrap(); + // Re-asserting the same (subject, predicate, object) closes nothing. + let again = literal_edge("e2", "acme", "status", "active"); + let closed = supersede_conflicting_edges(&s, &again).await.unwrap(); + assert!(closed.is_empty()); + } +} diff --git a/docs/architecture.md b/docs/architecture.md index 4377dc5..98b7581 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -40,6 +40,26 @@ Every downstream crate talks to `memory_storage::Storage`. No crate outside - Local + SaaS swap a runtime concern. - Sync engine clean — sync is `Storage` ↔ `Storage`. +## Temporal knowledge graph + +Entities and edges are **bi-temporal**: every edge carries `valid_from` / +`valid_to`, so the graph answers *"what did we believe on date X"*, not just +*"what is true now"*. `EdgeFilter::as_of` runs a point-in-time query +(`validFrom <= as_of AND (validTo IS NULL OR validTo > as_of)`); omitting it +with `current_only` returns only open edges. + +`storage::temporal` maintains that timeline. When a subject's single-valued +fact changes (`lives_in Boston` → `lives_in Denver`), +`supersede_conflicting_edges` *closes* the old edge at the new one's +`valid_from` (via `Storage::invalidate_edge`) instead of deleting it — so the +timeline stays contiguous and non-overlapping: exactly one current edge per +(subject, predicate), every past belief still queryable at its own `as_of`. +Only **functional** predicates supersede (`works_at`, `lives_in`, …); multi- +valued ones (`likes`, `attended`) legitimately keep many objects and are never +collapsed. Auto-supersession on write is gated behind +`MEMORY_TEMPORAL_SUPERSEDE_ENABLED=true` (default off; the primitives are always +available to call explicitly). + ## Mode selection (runtime, never compile-time) ```sh