grit_core/

lib.rs

//! # grit-core
//!
//! An embedded, bi-temporal property graph for agent memory: one SQLite file,
//! in-process, deterministic. See `AGENTS.md` at the repo root for the full
//! design contract; the short version:
//!
//! - All writes are [`GraphOp`]s appended to an op-log; graph tables are
//!   derived state applied in the same transaction (sync-ready by design).
//! - Facts are never mutated in place. Edges carry two timelines — event time
//!   (`valid_at`/`invalid_at`) and system time (`created_at`/`expired_at`) —
//!   so "what did I believe in March?" is a query, not archaeology.
//! - No LLM, no network, no embedding computation. Time is injected via
//!   [`Clock`]; embeddings are stored, never created.
//!
//! # Example
//! ```
//! use grit_core::{Budget, GraphOp, Grit, Options, Query, Traversal};
//! # fn main() -> Result<(), grit_core::Error> {
//! # let dir = std::env::temp_dir().join(format!("grit-doc-{}", std::process::id()));
//! # std::fs::create_dir_all(&dir)?;
//! # let path = dir.join("memory.db");
//! # let _ = std::fs::remove_file(&path);
//! let g = Grit::open(&path, Options::new("device-a"))?;
//!
//! let yoneda = g.new_id();
//! let functor = g.new_id();
//! g.apply(GraphOp::AddNode {
//!     id: yoneda,
//!     kind: "concept".into(),
//!     name: "Yoneda lemma".into(),
//!     summary: "Objects are determined by their relationships".into(),
//!     attrs: serde_json::json!({}),
//!     group_id: "category-theory".into(),
//! })?;
//! g.apply(GraphOp::AddNode {
//!     id: functor,
//!     kind: "concept".into(),
//!     name: "Functor".into(),
//!     summary: "Structure-preserving map between categories".into(),
//!     attrs: serde_json::json!({}),
//!     group_id: "category-theory".into(),
//! })?;
//! g.apply(GraphOp::AddEdge {
//!     id: g.new_id(),
//!     src: yoneda,
//!     dst: functor,
//!     rel: "STATED_IN_TERMS_OF".into(),
//!     fact: "The Yoneda lemma is stated in terms of hom-functors".into(),
//!     attrs: serde_json::json!({}),
//!     group_id: "category-theory".into(),
//!     valid_at: None,
//!     invalid_at: None,
//! })?;
//!
//! let hits = g.search(
//!     Query::text("yoneda").group("category-theory").budget(Budget::items(10)),
//! )?;
//! assert!(!hits.is_empty());
//!
//! let ctx = g.traverse(&[yoneda], &Traversal::default())?;
//! assert_eq!(ctx.nodes.len(), 2);
//! # Ok(()) }
//! ```
#![deny(unsafe_code)]
#![warn(missing_docs)]

mod clock;
mod error;
mod export;
mod hlc;
mod migrate;
mod model;
mod ops;
mod query;
mod search;
mod vecext;
mod writer;

use std::path::Path;
use std::sync::mpsc::{Receiver, Sender, channel};
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

use rusqlite::{Connection, OpenFlags};
use uuid::Uuid;

pub use crate::clock::{Clock, ManualClock, SystemClock, TimestampMs};
pub use crate::error::{Error, Result};
pub use crate::export::{ImportStats, import_jsonl};
pub use crate::hlc::{Hlc, HlcGenerator};
pub use crate::migrate::SCHEMA_VERSION;
pub use crate::model::{Edge, Episode, Node, Subgraph};
pub use crate::ops::{GraphOp, OplogEntry};
#[doc(hidden)]
pub use crate::query::{EXPAND_SQL_AS_AT, EXPAND_SQL_CURRENT};
pub use crate::query::{MergeCandidate, NodeHistory, Stats, Traversal};
pub use crate::search::{Budget, Query, SearchHit, SearchKind, SearchTarget};

use crate::writer::{VecTable, WriteMsg};

/// Open-time configuration for [`Grit`].
pub struct Options {
    /// Stable identifier for this device/installation; stamped on every op
    /// and used as the HLC tie-breaker.
    pub device_id: String,
    /// Time source. Defaults to [`SystemClock`]; tests inject [`ManualClock`].
    pub clock: Arc<dyn Clock>,
}

impl Options {
    /// Options with the given device id and the real system clock.
    pub fn new(device_id: impl Into<String>) -> Self {
        Self {
            device_id: device_id.into(),
            clock: Arc::new(SystemClock),
        }
    }

    /// Replace the time source (Design Invariant 3: time is injected).
    pub fn clock(mut self, clock: Arc<dyn Clock>) -> Self {
        self.clock = clock;
        self
    }
}

/// Handle to one grit database: a writer actor plus a read connection.
///
/// All methods take `&self`; `Grit` is `Send + Sync`. Writes are serialized
/// through the single writer thread (SQLite has one writer — embrace it);
/// reads run on a separate WAL connection and never block behind writes.
pub struct Grit {
    write_tx: Sender<WriteMsg>,
    writer: Option<JoinHandle<()>>,
    pub(crate) read_conn: Mutex<Connection>,
    clock: Arc<dyn Clock>,
    hlc: HlcGenerator,
    device_id: String,
}

impl Grit {
    /// Open (creating or migrating as needed) the database at `path`.
    ///
    /// The file plus its WAL sidecars is the entire truth (Design Invariant 2).
    /// In-memory databases are not supported: grit uses two connections
    /// (writer + reader) that must see the same file.
    ///
    /// # Example
    /// ```no_run
    /// use grit_core::{Grit, Options};
    /// let g = Grit::open("memory.db", Options::new("laptop"))?;
    /// # Ok::<(), grit_core::Error>(())
    /// ```
    pub fn open(path: impl AsRef<Path>, opts: Options) -> Result<Self> {
        let path = path.as_ref();
        // Two connections must see the same on-disk file: reject every SQLite
        // spelling of "not a real file" (bare :memory:, memory-mode URIs, and
        // the empty path, which opens a distinct temp db per connection).
        if let Some(s) = path.to_str()
            && (s.is_empty()
                || s == ":memory:"
                || (s.starts_with("file:")
                    && (s.contains(":memory:") || s.contains("mode=memory"))))
        {
            return Err(Error::InvalidOp(
                "in-memory or empty database paths are not supported".into(),
            ));
        }
        vecext::register_sqlite_vec();

        let mut write_conn = Connection::open(path)?;
        configure(&write_conn)?;
        migrate::migrate(&mut write_conn)?;

        let read_conn = Connection::open_with_flags(
            path,
            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
        )?;
        read_conn.busy_timeout(std::time::Duration::from_secs(5))?;
        configure_cache(&read_conn)?;

        let (write_tx, write_rx): (Sender<WriteMsg>, Receiver<WriteMsg>) = channel();
        let writer = writer::spawn(write_conn, Arc::clone(&opts.clock), write_rx);

        Ok(Self {
            write_tx,
            writer: Some(writer),
            read_conn: Mutex::new(read_conn),
            hlc: HlcGenerator::new(opts.device_id.clone()),
            clock: opts.clock,
            device_id: opts.device_id,
        })
    }

    /// Mint a UUIDv7 from the injected clock. Use for the `id` fields of
    /// [`GraphOp`]s.
    pub fn new_id(&self) -> Uuid {
        let ms = self.clock.now_ms().max(0) as u64;
        let ts = uuid::Timestamp::from_unix(
            uuid::NoContext,
            ms / 1000,
            ((ms % 1000) * 1_000_000) as u32,
        );
        Uuid::new_v7(ts)
    }

    /// Apply a local write: mint an [`OplogEntry`] (op id + HLC), append it to
    /// the oplog and update the graph tables in one transaction. Returns the
    /// entry — the unit a sync mechanism would ship to other devices.
    ///
    /// # Example
    /// ```no_run
    /// # use grit_core::{GraphOp, Grit, Options};
    /// # let g = Grit::open("memory.db", Options::new("laptop"))?;
    /// let entry = g.apply(GraphOp::AddNode {
    ///     id: g.new_id(),
    ///     kind: "person".into(),
    ///     name: "Ada Lovelace".into(),
    ///     summary: String::new(),
    ///     attrs: serde_json::json!({}),
    ///     group_id: String::new(),
    /// })?;
    /// assert_eq!(entry.device_id, "laptop");
    /// # Ok::<(), grit_core::Error>(())
    /// ```
    pub fn apply(&self, op: GraphOp) -> Result<OplogEntry> {
        validate(&op)?;
        let entry = OplogEntry {
            id: self.new_id(),
            hlc: self.hlc.next(self.clock.as_ref()),
            device_id: self.device_id.clone(),
            op,
        };
        writer::call(&self.write_tx, |reply| WriteMsg::Apply {
            entry: entry.clone(),
            reply,
        })?;
        Ok(entry)
    }

    /// Apply an op that originated on another device (future sync ingest, and
    /// the test seam for the oplog merge laws). Idempotent: returns `false`
    /// if this op id was already applied. Folds the remote HLC into the local
    /// generator so subsequent local ops sort after everything observed.
    ///
    /// Unlike [`Grit::apply`], structurally invalid ops (e.g. a self-merge)
    /// are recorded and applied as no-ops rather than erroring — one
    /// malformed entry from a buggy device must not wedge a sync loop. The
    /// one exception is a negative HLC wall time, which is rejected: it
    /// cannot be ordered, so nothing downstream of it can converge.
    pub fn apply_remote(&self, entry: OplogEntry) -> Result<bool> {
        if entry.hlc.wall_ms < 0 {
            return Err(Error::InvalidHlc(entry.hlc.encode()));
        }
        self.hlc.observe(&entry.hlc);
        writer::call(&self.write_tx, |reply| WriteMsg::Apply { entry, reply })
    }

    /// Read the oplog from local sequence `after_seq` (exclusive; 0 = start).
    /// Returns `(seq, entry)` pairs in local apply order.
    pub fn oplog_since(&self, after_seq: i64) -> Result<Vec<(i64, OplogEntry)>> {
        let conn = self.read();
        let mut stmt = conn.prepare_cached(
            "SELECT seq, id, hlc, device_id, op FROM oplog WHERE seq > ?1 ORDER BY seq",
        )?;
        let rows = stmt.query_map([after_seq], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, String>(3)?,
                row.get::<_, String>(4)?,
            ))
        })?;
        let mut out = Vec::new();
        for row in rows {
            let (seq, id, hlc, device_id, op) = row?;
            out.push((
                seq,
                OplogEntry {
                    id: Uuid::parse_str(&id)
                        .map_err(|_| Error::Corrupt(format!("bad uuid in oplog: {id}")))?,
                    hlc: hlc.parse()?,
                    device_id,
                    op: serde_json::from_str(&op)?,
                },
            ));
        }
        Ok(out)
    }

    /// Register the embedding model whose vectors this file stores. Creates
    /// the `vec_nodes`/`vec_edges` tables sized to `dim` if they are missing
    /// (also after an import — vectors are never exported, the tables are
    /// rebuilt empty and rows re-embed). Vectors are partitioned by their
    /// row's `group_id`, so vector search filters by group inside the KNN
    /// scan. Embeddings are recomputable local state — not part of the oplog
    /// (Design Invariant 5).
    pub fn register_embedding_model(
        &self,
        model_id: impl Into<String>,
        dim: usize,
        model_version: impl Into<String>,
    ) -> Result<()> {
        writer::call(&self.write_tx, |reply| WriteMsg::RegisterModel {
            model_id: model_id.into(),
            dim,
            version: model_version.into(),
            reply,
        })
    }

    /// Store (or replace) the embedding for a node. The caller computed it;
    /// grit only stores it, in the node's group partition. The node row must
    /// already exist ([`Error::NotFound`] otherwise) — apply the
    /// [`GraphOp::AddNode`] first, then embed. Embedding a purged id is a
    /// silent no-op (right to forget).
    pub fn set_node_embedding(&self, node_id: Uuid, vector: Vec<f32>) -> Result<()> {
        writer::call(&self.write_tx, |reply| WriteMsg::SetEmbedding {
            table: VecTable::Nodes,
            id: node_id,
            vector,
            reply,
        })
    }

    /// Store (or replace) the embedding for an edge (its fact sentence), in
    /// the edge's group partition. The edge row must already exist
    /// ([`Error::NotFound`] otherwise); embedding a purged id is a no-op.
    pub fn set_edge_embedding(&self, edge_id: Uuid, vector: Vec<f32>) -> Result<()> {
        writer::call(&self.write_tx, |reply| WriteMsg::SetEmbedding {
            table: VecTable::Edges,
            id: edge_id,
            vector,
            reply,
        })
    }

    pub(crate) fn now_ms(&self) -> TimestampMs {
        self.clock.now_ms()
    }

    /// Lock the read connection, recovering from poisoning: a panic on
    /// another thread mid-read leaves the connection itself sound (rusqlite
    /// resets statements), and a storage library must not convert one panic
    /// into a permanent panic on every subsequent call.
    pub(crate) fn read(&self) -> std::sync::MutexGuard<'_, Connection> {
        self.read_conn
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }
}

// Compile-time proof of the documented `Send + Sync` claim on `Grit`.
const _: () = {
    const fn assert_send_sync<T: Send + Sync>() {}
    assert_send_sync::<Grit>();
};

impl Drop for Grit {
    fn drop(&mut self) {
        let _ = self.write_tx.send(WriteMsg::Shutdown);
        if let Some(handle) = self.writer.take() {
            let _ = handle.join();
        }
    }
}

fn validate(op: &GraphOp) -> Result<()> {
    match op {
        GraphOp::MergeNodes { from, into } if from == into => {
            Err(Error::InvalidOp("cannot merge a node into itself".into()))
        }
        GraphOp::Purge { ids } if ids.is_empty() => {
            Err(Error::InvalidOp("purge with no ids".into()))
        }
        GraphOp::UpdateNode {
            name: None,
            summary: None,
            kind: None,
            attrs: None,
            ..
        } => Err(Error::InvalidOp("update with no fields".into())),
        _ => Ok(()),
    }
}

fn configure(conn: &Connection) -> Result<()> {
    conn.busy_timeout(std::time::Duration::from_secs(5))?;
    conn.pragma_update(None, "journal_mode", "WAL")?;
    conn.pragma_update(None, "synchronous", "NORMAL")?;
    configure_cache(conn)?;
    Ok(())
}

/// Read-performance pragmas, applied to writer and reader connections alike.
/// SQLite's defaults (2 MB page cache, no mmap) assume a shared server box;
/// grit is a personal memory store where the traversal latency budget
/// (AGENTS.md invariant 6) is worth 32 MB of cache and a read-only mmap.
fn configure_cache(conn: &Connection) -> Result<()> {
    conn.pragma_update(None, "cache_size", -32_768)?; // KiB units when negative: 32 MB
    conn.pragma_update(None, "mmap_size", 268_435_456)?; // 256 MB
    Ok(())
}