Update 6 SQLite

2026-08-05 05:23:43 +02:00
parent 7e36c2f3f0
commit b4388d1ebc
40 changed files with 2514 additions and 229 deletions
@@ -0,0 +1,700 @@
+package graph
+
+import (
+	"context"
+	"database/sql"
+	"encoding/binary"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"math"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/local/glpi-neural-brain/internal/model"
+	_ "modernc.org/sqlite"
+)
+
+const schemaVersion = 1
+
+const nodeUpsertSQL = `INSERT INTO nodes(id,kind,label,summary,status,origin,external_id,uri,categories_json,keywords_json,metadata_json,weight,x,y,z,updated_at_ns)
+VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+ON CONFLICT(id) DO UPDATE SET kind=excluded.kind,label=excluded.label,summary=excluded.summary,status=excluded.status,origin=excluded.origin,external_id=excluded.external_id,uri=excluded.uri,categories_json=excluded.categories_json,keywords_json=excluded.keywords_json,metadata_json=excluded.metadata_json,weight=excluded.weight,x=excluded.x,y=excluded.y,z=excluded.z,updated_at_ns=excluded.updated_at_ns`
+
+const edgeUpsertSQL = `INSERT INTO edges(id,source,target,type,origin,status,confidence,weight,explanation,evidence_json,metadata_json,created_at_ns,updated_at_ns)
+VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)
+ON CONFLICT(id) DO UPDATE SET source=excluded.source,target=excluded.target,type=excluded.type,origin=excluded.origin,status=excluded.status,confidence=excluded.confidence,weight=excluded.weight,explanation=excluded.explanation,evidence_json=excluded.evidence_json,metadata_json=excluded.metadata_json,created_at_ns=excluded.created_at_ns,updated_at_ns=excluded.updated_at_ns`
+
+const vectorUpsertSQL = `INSERT INTO vectors(node_id,dimensions,data,updated_at_ns) VALUES(?,?,?,?)
+ON CONFLICT(node_id) DO UPDATE SET dimensions=excluded.dimensions,data=excluded.data,updated_at_ns=excluded.updated_at_ns`
+
+// StorageStatus describes the persistent SQLite graph store. The live graph is
+// still held in memory; only changed rows are written during a flush.
+type StorageStatus struct {
+	Backend           string `json:"backend"`
+	Path              string `json:"path"`
+	SchemaVersion     int    `json:"schema_version"`
+	JournalMode       string `json:"journal_mode"`
+	DatabaseBytes     int64  `json:"database_bytes"`
+	WALBytes          int64  `json:"wal_bytes"`
+	NodeRows          int    `json:"node_rows"`
+	EdgeRows          int    `json:"edge_rows"`
+	VectorRows        int    `json:"vector_rows"`
+	VectorBytes       int64  `json:"vector_bytes"`
+	PersistedVersion  uint64 `json:"persisted_version"`
+	EmbeddingModel    string `json:"embedding_model,omitempty"`
+	EmbeddingDigest   string `json:"embedding_digest,omitempty"`
+	LegacyJSONPresent bool   `json:"legacy_json_present"`
+	PendingNodes      int    `json:"pending_nodes"`
+	PendingEdges      int    `json:"pending_edges"`
+	PendingVectors    int    `json:"pending_vectors"`
+	PendingDeletions  int    `json:"pending_deletions"`
+	PendingMetadata   bool   `json:"pending_metadata"`
+}
+
+func Open(dir string) (*Store, error) {
+	absoluteDir, err := filepath.Abs(dir)
+	if err != nil {
+		return nil, fmt.Errorf("resolve graph data directory: %w", err)
+	}
+	if err := os.MkdirAll(absoluteDir, 0o750); err != nil {
+		return nil, fmt.Errorf("create graph data directory %q: %w", absoluteDir, err)
+	}
+	if err := ensureWritableDirectory(absoluteDir); err != nil {
+		return nil, err
+	}
+
+	path := filepath.Join(absoluteDir, "graph.db")
+	if info, statErr := os.Stat(path); statErr == nil && info.IsDir() {
+		return nil, fmt.Errorf("graph database path %q is a directory; remove or rename it", path)
+	} else if statErr != nil && !errors.Is(statErr, os.ErrNotExist) {
+		return nil, fmt.Errorf("inspect graph database %q: %w", path, statErr)
+	}
+
+	// Use the plain filename here instead of applying PRAGMAs through the DSN.
+	// This makes startup failures attributable to one concrete step and avoids
+	// asking the driver to reserve a large mmap/cache while the connection is
+	// being created. The application keeps the full graph in Go memory already,
+	// so an additional 64 MiB SQLite page cache and 256 MiB mmap are wasteful on
+	// small systems.
+	db, err := sql.Open("sqlite", path)
+	if err != nil {
+		return nil, fmt.Errorf("create sqlite handle for %q: %w", path, err)
+	}
+	// The Store serializes persistence and reads the live graph from memory.
+	// One SQLite connection is sufficient and keeps all connection-local
+	// PRAGMAs deterministic.
+	db.SetMaxOpenConns(1)
+	db.SetMaxIdleConns(1)
+	db.SetConnMaxLifetime(0)
+	db.SetConnMaxIdleTime(0)
+
+	openCtx, openCancel := context.WithTimeout(context.Background(), 15*time.Second)
+	if err := db.PingContext(openCtx); err != nil {
+		openCancel()
+		db.Close()
+		return nil, fmt.Errorf("open sqlite database %q: %w", path, err)
+	}
+	journalMode, err := configureSQLite(openCtx, db)
+	openCancel()
+	if err != nil {
+		db.Close()
+		return nil, fmt.Errorf("configure sqlite database %q: %w", path, err)
+	}
+
+	s := &Store{
+		nodes:          map[string]model.Node{},
+		edges:          map[string]model.Edge{},
+		vectors:        map[string][]float32{},
+		db:             db,
+		dbPath:         path,
+		journalMode:    journalMode,
+		dirtyNodes:     map[string]uint64{},
+		dirtyEdges:     map[string]uint64{},
+		dirtyVectors:   map[string]uint64{},
+		deletedNodes:   map[string]uint64{},
+		deletedEdges:   map[string]uint64{},
+		deletedVectors: map[string]uint64{},
+	}
+	loadCtx, loadCancel := context.WithTimeout(context.Background(), 10*time.Minute)
+	defer loadCancel()
+	if err := s.initSchema(loadCtx); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("initialize sqlite graph schema: %w", err)
+	}
+	if err := s.load(loadCtx); err != nil {
+		db.Close()
+		return nil, fmt.Errorf("load sqlite graph state: %w", err)
+	}
+	return s, nil
+}
+
+func ensureWritableDirectory(dir string) error {
+	info, err := os.Stat(dir)
+	if err != nil {
+		return fmt.Errorf("inspect graph data directory %q: %w", dir, err)
+	}
+	if !info.IsDir() {
+		return fmt.Errorf("graph data path %q is not a directory", dir)
+	}
+	probe, err := os.CreateTemp(dir, ".brain-write-test-*")
+	if err != nil {
+		return fmt.Errorf("graph data directory %q is not writable by the brain process: %w", dir, err)
+	}
+	name := probe.Name()
+	closeErr := probe.Close()
+	removeErr := os.Remove(name)
+	if closeErr != nil {
+		return fmt.Errorf("close graph data write test %q: %w", name, closeErr)
+	}
+	if removeErr != nil {
+		return fmt.Errorf("remove graph data write test %q: %w", name, removeErr)
+	}
+	return nil
+}
+
+func configureSQLite(ctx context.Context, db *sql.DB) (string, error) {
+	// Conservative defaults: the live graph and vectors already reside in Go
+	// memory. SQLite therefore only needs a small page cache and no mmap window.
+	// This keeps startup predictable inside memory-limited containers.
+	statements := []struct {
+		name string
+		sql  string
+	}{
+		{name: "busy_timeout", sql: `PRAGMA busy_timeout=5000`},
+		{name: "foreign_keys", sql: `PRAGMA foreign_keys=ON`},
+		{name: "synchronous", sql: `PRAGMA synchronous=NORMAL`},
+		{name: "temp_store", sql: `PRAGMA temp_store=FILE`},
+		{name: "cache_size", sql: `PRAGMA cache_size=-8192`},
+		{name: "mmap_size", sql: `PRAGMA mmap_size=0`},
+		{name: "wal_autocheckpoint", sql: `PRAGMA wal_autocheckpoint=0`},
+		{name: "journal_size_limit", sql: `PRAGMA journal_size_limit=67108864`},
+	}
+	for _, statement := range statements {
+		if _, err := db.ExecContext(ctx, statement.sql); err != nil {
+			return "", fmt.Errorf("apply PRAGMA %s: %w", statement.name, err)
+		}
+	}
+
+	var journalMode string
+	if err := db.QueryRowContext(ctx, `PRAGMA journal_mode=WAL`).Scan(&journalMode); err != nil {
+		return "", fmt.Errorf("enable WAL journal mode: %w", err)
+	}
+	journalMode = strings.ToLower(strings.TrimSpace(journalMode))
+	if journalMode != "wal" {
+		return "", fmt.Errorf("enable WAL journal mode: SQLite selected %q", journalMode)
+	}
+	return journalMode, nil
+}
+
+func (s *Store) initSchema(ctx context.Context) error {
+	statements := []string{
+		`CREATE TABLE IF NOT EXISTS graph_meta (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+) WITHOUT ROWID`,
+		`CREATE TABLE IF NOT EXISTS nodes (
+    id TEXT PRIMARY KEY,
+    kind TEXT NOT NULL,
+    label TEXT NOT NULL,
+    summary TEXT NOT NULL DEFAULT '',
+    status TEXT NOT NULL DEFAULT '',
+    origin TEXT NOT NULL,
+    external_id TEXT NOT NULL DEFAULT '',
+    uri TEXT NOT NULL DEFAULT '',
+    categories_json TEXT NOT NULL DEFAULT '[]',
+    keywords_json TEXT NOT NULL DEFAULT '[]',
+    metadata_json TEXT NOT NULL DEFAULT '{}',
+    weight REAL NOT NULL DEFAULT 1,
+    x REAL NOT NULL DEFAULT 0,
+    y REAL NOT NULL DEFAULT 0,
+    z REAL NOT NULL DEFAULT 0,
+    updated_at_ns INTEGER NOT NULL
+) WITHOUT ROWID`,
+		`CREATE INDEX IF NOT EXISTS idx_nodes_origin ON nodes(origin)`,
+		`CREATE INDEX IF NOT EXISTS idx_nodes_kind_status ON nodes(kind, status)`,
+		`CREATE INDEX IF NOT EXISTS idx_nodes_external_id ON nodes(external_id)`,
+		`CREATE TABLE IF NOT EXISTS edges (
+    id TEXT PRIMARY KEY,
+    source TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
+    target TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
+    type TEXT NOT NULL,
+    origin TEXT NOT NULL,
+    status TEXT NOT NULL DEFAULT '',
+    confidence REAL NOT NULL DEFAULT 0,
+    weight REAL NOT NULL DEFAULT 1,
+    explanation TEXT NOT NULL DEFAULT '',
+    evidence_json TEXT NOT NULL DEFAULT '[]',
+    metadata_json TEXT NOT NULL DEFAULT '{}',
+    created_at_ns INTEGER NOT NULL,
+    updated_at_ns INTEGER NOT NULL
+) WITHOUT ROWID`,
+		`CREATE INDEX IF NOT EXISTS idx_edges_source ON edges(source)`,
+		`CREATE INDEX IF NOT EXISTS idx_edges_target ON edges(target)`,
+		`CREATE INDEX IF NOT EXISTS idx_edges_origin_status ON edges(origin, status)`,
+		`CREATE TABLE IF NOT EXISTS vectors (
+    node_id TEXT PRIMARY KEY REFERENCES nodes(id) ON DELETE CASCADE,
+    dimensions INTEGER NOT NULL,
+    data BLOB NOT NULL,
+    updated_at_ns INTEGER NOT NULL
+) WITHOUT ROWID`,
+	}
+	for _, statement := range statements {
+		if _, err := s.db.ExecContext(ctx, statement); err != nil {
+			return fmt.Errorf("initialize graph sqlite schema: %w", err)
+		}
+	}
+	var current int
+	err := s.db.QueryRowContext(ctx, `SELECT CAST(value AS INTEGER) FROM graph_meta WHERE key='schema_version'`).Scan(&current)
+	if errors.Is(err, sql.ErrNoRows) {
+		_, err = s.db.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('schema_version', ?)`, schemaVersion)
+		return err
+	}
+	if err != nil {
+		return err
+	}
+	if current != schemaVersion {
+		return fmt.Errorf("unsupported graph database schema %d (expected %d)", current, schemaVersion)
+	}
+	return nil
+}
+
+func (s *Store) load(ctx context.Context) error {
+	if err := s.loadNodes(ctx); err != nil {
+		return err
+	}
+	if err := s.loadEdges(ctx); err != nil {
+		return err
+	}
+	if err := s.loadVectors(ctx); err != nil {
+		return err
+	}
+	var version uint64
+	if err := s.db.QueryRowContext(ctx, `SELECT CAST(value AS INTEGER) FROM graph_meta WHERE key='graph_version'`).Scan(&version); err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return err
+	}
+	var embeddingModel string
+	if err := s.db.QueryRowContext(ctx, `SELECT value FROM graph_meta WHERE key='embedding_model'`).Scan(&embeddingModel); err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return err
+	}
+	var embeddingDigest string
+	if err := s.db.QueryRowContext(ctx, `SELECT value FROM graph_meta WHERE key='embedding_digest'`).Scan(&embeddingDigest); err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return err
+	}
+	s.version = version
+	s.persistedVersion = version
+	s.embeddingModel = embeddingModel
+	s.embeddingDigest = embeddingDigest
+	return nil
+}
+
+func (s *Store) loadNodes(ctx context.Context) error {
+	rows, err := s.db.QueryContext(ctx, `SELECT id,kind,label,summary,status,origin,external_id,uri,categories_json,keywords_json,metadata_json,weight,x,y,z,updated_at_ns FROM nodes`)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var n model.Node
+		var categories, keywords, metadata string
+		var updated int64
+		if err := rows.Scan(&n.ID, &n.Kind, &n.Label, &n.Summary, &n.Status, &n.Origin, &n.ExternalID, &n.URI, &categories, &keywords, &metadata, &n.Weight, &n.X, &n.Y, &n.Z, &updated); err != nil {
+			return err
+		}
+		if err := decodeJSON(categories, &n.Categories); err != nil {
+			return fmt.Errorf("decode node %s categories: %w", n.ID, err)
+		}
+		if err := decodeJSON(keywords, &n.Keywords); err != nil {
+			return fmt.Errorf("decode node %s keywords: %w", n.ID, err)
+		}
+		if err := decodeJSON(metadata, &n.Metadata); err != nil {
+			return fmt.Errorf("decode node %s metadata: %w", n.ID, err)
+		}
+		if n.Metadata == nil {
+			n.Metadata = map[string]any{}
+		}
+		n.UpdatedAt = time.Unix(0, updated).UTC()
+		s.nodes[n.ID] = n
+	}
+	return rows.Err()
+}
+
+func (s *Store) loadEdges(ctx context.Context) error {
+	rows, err := s.db.QueryContext(ctx, `SELECT id,source,target,type,origin,status,confidence,weight,explanation,evidence_json,metadata_json,created_at_ns,updated_at_ns FROM edges`)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var e model.Edge
+		var evidence, metadata string
+		var created, updated int64
+		if err := rows.Scan(&e.ID, &e.Source, &e.Target, &e.Type, &e.Origin, &e.Status, &e.Confidence, &e.Weight, &e.Explanation, &evidence, &metadata, &created, &updated); err != nil {
+			return err
+		}
+		if err := decodeJSON(evidence, &e.Evidence); err != nil {
+			return fmt.Errorf("decode edge %s evidence: %w", e.ID, err)
+		}
+		if err := decodeJSON(metadata, &e.Metadata); err != nil {
+			return fmt.Errorf("decode edge %s metadata: %w", e.ID, err)
+		}
+		if e.Metadata == nil {
+			e.Metadata = map[string]any{}
+		}
+		e.CreatedAt = time.Unix(0, created).UTC()
+		e.UpdatedAt = time.Unix(0, updated).UTC()
+		s.edges[e.ID] = e
+	}
+	return rows.Err()
+}
+
+func (s *Store) loadVectors(ctx context.Context) error {
+	rows, err := s.db.QueryContext(ctx, `SELECT node_id,dimensions,data FROM vectors`)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id string
+		var dimensions int
+		var data []byte
+		if err := rows.Scan(&id, &dimensions, &data); err != nil {
+			return err
+		}
+		v, err := decodeVector(data, dimensions)
+		if err != nil {
+			return fmt.Errorf("decode vector %s: %w", id, err)
+		}
+		s.vectors[id] = v
+	}
+	return rows.Err()
+}
+
+func decodeJSON(raw string, target any) error {
+	if strings.TrimSpace(raw) == "" {
+		return nil
+	}
+	dec := json.NewDecoder(strings.NewReader(raw))
+	dec.UseNumber()
+	return dec.Decode(target)
+}
+
+func encodeJSON(value any, empty string) (string, error) {
+	if value == nil {
+		return empty, nil
+	}
+	b, err := json.Marshal(value)
+	if err != nil {
+		return "", err
+	}
+	return string(b), nil
+}
+
+func encodeVector(v []float32) []byte {
+	out := make([]byte, len(v)*4)
+	for i, value := range v {
+		binary.LittleEndian.PutUint32(out[i*4:], math.Float32bits(value))
+	}
+	return out
+}
+
+func decodeVector(data []byte, dimensions int) ([]float32, error) {
+	if dimensions < 0 || len(data) != dimensions*4 {
+		return nil, fmt.Errorf("invalid float32 vector blob: dimensions=%d bytes=%d", dimensions, len(data))
+	}
+	out := make([]float32, dimensions)
+	for i := range out {
+		out[i] = math.Float32frombits(binary.LittleEndian.Uint32(data[i*4:]))
+	}
+	return out, nil
+}
+
+// PersistVersion writes only records changed since the previous successful
+// flush. Vectors are encoded as little-endian float32 BLOBs.
+func (s *Store) PersistVersion() (uint64, error) {
+	return s.PersistVersionContext(context.Background())
+}
+
+func (s *Store) PersistVersionContext(ctx context.Context) (uint64, error) {
+	type nodeChange struct {
+		gen  uint64
+		node model.Node
+	}
+	type edgeChange struct {
+		gen  uint64
+		edge model.Edge
+	}
+	type vectorChange struct {
+		gen    uint64
+		vector []float32
+	}
+
+	s.mu.RLock()
+	version := s.version
+	persistedVersion := s.persistedVersion
+	embeddingModel := s.embeddingModel
+	embeddingDigest := s.embeddingDigest
+	embeddingMetaGeneration := s.embeddingMetaGeneration
+	nodes := make(map[string]nodeChange, len(s.dirtyNodes))
+	for id, gen := range s.dirtyNodes {
+		if n, ok := s.nodes[id]; ok {
+			nodes[id] = nodeChange{gen: gen, node: n}
+		}
+	}
+	edges := make(map[string]edgeChange, len(s.dirtyEdges))
+	for id, gen := range s.dirtyEdges {
+		if e, ok := s.edges[id]; ok {
+			edges[id] = edgeChange{gen: gen, edge: e}
+		}
+	}
+	vectors := make(map[string]vectorChange, len(s.dirtyVectors))
+	for id, gen := range s.dirtyVectors {
+		if v, ok := s.vectors[id]; ok {
+			vectors[id] = vectorChange{gen: gen, vector: append([]float32(nil), v...)}
+		}
+	}
+	deletedNodes := cloneGenerations(s.deletedNodes)
+	deletedEdges := cloneGenerations(s.deletedEdges)
+	deletedVectors := cloneGenerations(s.deletedVectors)
+	s.mu.RUnlock()
+
+	if len(nodes)+len(edges)+len(vectors)+len(deletedNodes)+len(deletedEdges)+len(deletedVectors) == 0 && embeddingMetaGeneration == 0 && version == persistedVersion {
+		return version, nil
+	}
+
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return 0, err
+	}
+	defer tx.Rollback()
+
+	for id := range deletedEdges {
+		if _, err := tx.ExecContext(ctx, `DELETE FROM edges WHERE id=?`, id); err != nil {
+			return 0, err
+		}
+	}
+	for id := range deletedNodes {
+		if _, err := tx.ExecContext(ctx, `DELETE FROM nodes WHERE id=?`, id); err != nil {
+			return 0, err
+		}
+	}
+	for id := range deletedVectors {
+		if _, err := tx.ExecContext(ctx, `DELETE FROM vectors WHERE node_id=?`, id); err != nil {
+			return 0, err
+		}
+	}
+
+	if len(nodes) > 0 {
+		statement, err := tx.PrepareContext(ctx, nodeUpsertSQL)
+		if err != nil {
+			return 0, err
+		}
+		for _, change := range nodes {
+			if err := upsertNode(ctx, statement, change.node); err != nil {
+				_ = statement.Close()
+				return 0, err
+			}
+		}
+		if err := statement.Close(); err != nil {
+			return 0, err
+		}
+	}
+	if len(edges) > 0 {
+		statement, err := tx.PrepareContext(ctx, edgeUpsertSQL)
+		if err != nil {
+			return 0, err
+		}
+		for _, change := range edges {
+			if err := upsertEdge(ctx, statement, change.edge); err != nil {
+				_ = statement.Close()
+				return 0, err
+			}
+		}
+		if err := statement.Close(); err != nil {
+			return 0, err
+		}
+	}
+	if len(vectors) > 0 {
+		statement, err := tx.PrepareContext(ctx, vectorUpsertSQL)
+		if err != nil {
+			return 0, err
+		}
+		now := time.Now().UTC().UnixNano()
+		for id, change := range vectors {
+			if _, err := statement.ExecContext(ctx, id, len(change.vector), encodeVector(change.vector), now); err != nil {
+				_ = statement.Close()
+				return 0, err
+			}
+		}
+		if err := statement.Close(); err != nil {
+			return 0, err
+		}
+	}
+	if embeddingMetaGeneration > 0 {
+		if _, err := tx.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('embedding_model',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, embeddingModel); err != nil {
+			return 0, err
+		}
+		if _, err := tx.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('embedding_digest',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, embeddingDigest); err != nil {
+			return 0, err
+		}
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO graph_meta(key,value) VALUES('graph_version',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, version); err != nil {
+		return 0, err
+	}
+	if err := tx.Commit(); err != nil {
+		return 0, err
+	}
+
+	s.mu.Lock()
+	for id, change := range nodes {
+		if s.dirtyNodes[id] == change.gen {
+			delete(s.dirtyNodes, id)
+		}
+	}
+	for id, change := range edges {
+		if s.dirtyEdges[id] == change.gen {
+			delete(s.dirtyEdges, id)
+		}
+	}
+	for id, change := range vectors {
+		if s.dirtyVectors[id] == change.gen {
+			delete(s.dirtyVectors, id)
+		}
+	}
+	clearDeletedIfSame(s.deletedNodes, deletedNodes)
+	clearDeletedIfSame(s.deletedEdges, deletedEdges)
+	clearDeletedIfSame(s.deletedVectors, deletedVectors)
+	if s.embeddingMetaGeneration == embeddingMetaGeneration {
+		s.embeddingMetaGeneration = 0
+	}
+	if version > s.persistedVersion {
+		s.persistedVersion = version
+	}
+	s.mu.Unlock()
+
+	// Keep the WAL bounded. PASSIVE never blocks readers and leaves busy pages
+	// for the next scheduled flush.
+	_, _ = s.db.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE)`)
+	return version, nil
+}
+
+func upsertNode(ctx context.Context, statement *sql.Stmt, n model.Node) error {
+	categories, err := encodeJSON(n.Categories, "[]")
+	if err != nil {
+		return err
+	}
+	keywords, err := encodeJSON(n.Keywords, "[]")
+	if err != nil {
+		return err
+	}
+	metadata, err := encodeJSON(n.Metadata, "{}")
+	if err != nil {
+		return err
+	}
+	_, err = statement.ExecContext(ctx, n.ID, n.Kind, n.Label, n.Summary, n.Status, n.Origin, n.ExternalID, n.URI, categories, keywords, metadata, n.Weight, n.X, n.Y, n.Z, n.UpdatedAt.UnixNano())
+	return err
+}
+
+func upsertEdge(ctx context.Context, statement *sql.Stmt, e model.Edge) error {
+	evidence, err := encodeJSON(e.Evidence, "[]")
+	if err != nil {
+		return err
+	}
+	metadata, err := encodeJSON(e.Metadata, "{}")
+	if err != nil {
+		return err
+	}
+	_, err = statement.ExecContext(ctx, e.ID, e.Source, e.Target, e.Type, e.Origin, e.Status, e.Confidence, e.Weight, e.Explanation, evidence, metadata, e.CreatedAt.UnixNano(), e.UpdatedAt.UnixNano())
+	return err
+}
+
+func cloneGenerations(in map[string]uint64) map[string]uint64 {
+	out := make(map[string]uint64, len(in))
+	for id, gen := range in {
+		out[id] = gen
+	}
+	return out
+}
+
+func clearDeletedIfSame(current, snapshot map[string]uint64) {
+	for id, gen := range snapshot {
+		if current[id] == gen {
+			delete(current, id)
+		}
+	}
+}
+
+func (s *Store) Persist() error {
+	_, err := s.PersistVersion()
+	return err
+}
+
+func (s *Store) Checkpoint(ctx context.Context, truncate bool) error {
+	mode := "PASSIVE"
+	if truncate {
+		mode = "TRUNCATE"
+	}
+	_, err := s.db.ExecContext(ctx, `PRAGMA wal_checkpoint(`+mode+`)`)
+	return err
+}
+
+func (s *Store) Close() error {
+	if s.db == nil {
+		return nil
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	_, persistErr := s.PersistVersionContext(ctx)
+	checkpointErr := s.Checkpoint(ctx, true)
+	closeErr := s.db.Close()
+	return errors.Join(persistErr, checkpointErr, closeErr)
+}
+
+func (s *Store) StorageStatus() StorageStatus {
+	s.mu.RLock()
+	var vectorBytes int64
+	for _, vector := range s.vectors {
+		vectorBytes += int64(len(vector) * 4)
+	}
+	status := StorageStatus{
+		Backend: "sqlite", Path: s.dbPath, SchemaVersion: schemaVersion, JournalMode: s.journalMode,
+		NodeRows: len(s.nodes), EdgeRows: len(s.edges), VectorRows: len(s.vectors), VectorBytes: vectorBytes,
+		PersistedVersion: s.persistedVersion,
+		EmbeddingModel:   s.embeddingModel,
+		EmbeddingDigest:  s.embeddingDigest,
+		PendingNodes:     len(s.dirtyNodes), PendingEdges: len(s.dirtyEdges), PendingVectors: len(s.dirtyVectors),
+		PendingDeletions: len(s.deletedNodes) + len(s.deletedEdges) + len(s.deletedVectors),
+		PendingMetadata:  s.embeddingMetaGeneration > 0,
+	}
+	s.mu.RUnlock()
+	if info, err := os.Stat(s.dbPath); err == nil {
+		status.DatabaseBytes = info.Size()
+	}
+	if info, err := os.Stat(s.dbPath + "-wal"); err == nil {
+		status.WALBytes = info.Size()
+	}
+	_, err := os.Stat(filepath.Join(filepath.Dir(s.dbPath), "graph-state.json"))
+	status.LegacyJSONPresent = err == nil
+	return status
+}
+
+// Export creates a compact, transactionally consistent SQLite copy. It can be
+// moved to another installation with the same knowledge-base identifiers.
+func (s *Store) Export(ctx context.Context, destination string) error {
+	if _, err := s.PersistVersionContext(ctx); err != nil {
+		return err
+	}
+	if err := os.MkdirAll(filepath.Dir(destination), 0o750); err != nil {
+		return err
+	}
+	if err := os.Remove(destination); err != nil && !errors.Is(err, os.ErrNotExist) {
+		return err
+	}
+	quoted := strings.ReplaceAll(destination, "'", "''")
+	if _, err := s.db.ExecContext(ctx, `VACUUM INTO '`+quoted+`'`); err != nil {
+		return fmt.Errorf("export graph database: %w", err)
+	}
+	return os.Chmod(destination, 0o640)
+}