SQLite + Bun

src/
  db/
    connection.ts
    migrations.ts
    schema/
    repositories/
    queries/
  search/
    lexical.ts
    vector.ts
    hybrid.ts
  memory/
    extraction.ts
    consolidation.ts
  lineage/
    emitter.ts
    outbox.ts
  workers/
    read-worker.ts
    write-worker.ts
  app.ts
data/
  .gitkeep
test/
  fixtures/
  retrieval-evaluation/

import { Database } from "bun:sqlite";

export interface OpenDatabaseOptions {
  filename: string;
  readonly?: boolean;
}

export function openDatabase(options: OpenDatabaseOptions): Database {
  const db = new Database(options.filename, {
    create: !options.readonly,
    readonly: options.readonly ?? false,
    strict: true,
    safeIntegers: true,
  });

  db.exec(`
    PRAGMA foreign_keys = ON;
    PRAGMA busy_timeout = 5000;
    PRAGMA trusted_schema = OFF;
  `);

  if (options.readonly) {
    db.exec(`PRAGMA query_only = ON;`);
  } else {
    const row = db.query<{ journal_mode: string }, []>(
      `PRAGMA journal_mode = WAL`
    ).get();

    if (row?.journal_mode.toLowerCase() !== "wal") {
      db.close();
      throw new Error(`WAL activation failed: ${row?.journal_mode}`);
    }

    db.exec(`PRAGMA synchronous = NORMAL;`);
  }

  return db;
}

import type { Database } from "bun:sqlite";

interface Migration {
  version: number;
  name: string;
  sql: string;
}

const migrations: readonly Migration[] = [
  {
    version: 1,
    name: "canonical-schema",
    sql: `
      CREATE TABLE document (
        id          TEXT PRIMARY KEY,
        tenant_id   TEXT NOT NULL,
        title       TEXT NOT NULL,
        content     TEXT NOT NULL,
        metadata    TEXT NOT NULL DEFAULT '{}'
                    CHECK (json_valid(metadata)),
        created_at  INTEGER NOT NULL,
        updated_at  INTEGER NOT NULL
      ) STRICT;

      CREATE INDEX document_tenant_updated_idx
      ON document(tenant_id, updated_at DESC);
    `,
  },
];

export function migrate(db: Database): void {
  db.exec(`
    CREATE TABLE IF NOT EXISTS schema_migration (
      version     INTEGER PRIMARY KEY,
      name        TEXT NOT NULL,
      applied_at  INTEGER NOT NULL
    ) STRICT;
  `);

  const applied = new Set(
    db.query<{ version: number }, []>(
      `SELECT version FROM schema_migration`
    ).all().map((row) => row.version),
  );

  const apply = db.transaction((migration: Migration) => {
    db.exec(migration.sql);
    db.query(`
      INSERT INTO schema_migration(version, name, applied_at)
      VALUES ($version, $name, $appliedAt)
    `).run({
      version: migration.version,
      name: migration.name,
      appliedAt: Date.now(),
    });
  });

  for (const migration of migrations) {
    if (!applied.has(migration.version)) {
      apply.immediate(migration);
    }
  }
}

interface ClaimedTask {
  id: string;
  payload: string;
  attempts: bigint;
  leaseUntil: bigint;
}

const claimStatement = db.query<
  ClaimedTask,
  { workerId: string; now: bigint; leaseUntil: bigint }
>(`
  UPDATE task
  SET
    status = 'running',
    lease_owner = $workerId,
    lease_until = $leaseUntil,
    attempts = attempts + 1,
    updated_at = $now
  WHERE id = (
    SELECT id
    FROM task
    WHERE status = 'pending'
       OR (status = 'running' AND lease_until < $now)
    ORDER BY priority DESC, created_at
    LIMIT 1
  )
  RETURNING
    id,
    payload,
    attempts,
    lease_until AS leaseUntil
`);

const claimTask = db.transaction((workerId: string) => {
  const now = BigInt(Date.now());
  return claimStatement.get({
    workerId,
    now,
    leaseUntil: now + 60_000n,
  });
});

const task = claimTask.immediate(crypto.randomUUID());

const persistDocument = db.transaction((input: DocumentInput) => {
  const documentId = insertDocument(input);

  db.exec("SAVEPOINT derived_indexes");
  try {
    insertFtsProjection(documentId, input);
    insertVectorProjection(documentId, input.embedding);
    db.exec("RELEASE derived_indexes");
  } catch (error) {
    db.exec("ROLLBACK TO derived_indexes");
    db.exec("RELEASE derived_indexes");
    markIndexingPending(documentId, error);
  }

  return documentId;
});

WITH RECURSIVE walk(entity_id, depth, path) AS (
  SELECT
    :start_id,
    0,
    ',' || :start_id || ','

  UNION ALL

  SELECT
    relation.target_id,
    walk.depth + 1,
    walk.path || relation.target_id || ','
  FROM walk
  JOIN relation
    ON relation.source_id = walk.entity_id
  WHERE relation.tenant_id = :tenant_id
    AND walk.depth < :max_depth
    AND instr(
      walk.path,
      ',' || relation.target_id || ','
    ) = 0
)
SELECT
  entity_id,
  MIN(depth) AS distance
FROM walk
GROUP BY entity_id
ORDER BY distance
LIMIT :limit;

SELECT
  query_id,
  memory_id,
  fused_score,
  row_number() OVER (
    PARTITION BY query_id
    ORDER BY fused_score DESC
  ) AS rank_within_query,
  fused_score - lead(fused_score) OVER (
    PARTITION BY query_id
    ORDER BY fused_score DESC
  ) AS margin_to_next
FROM retrieval_candidate;

CREATE TABLE document (
  id        TEXT PRIMARY KEY,
  tenant_id TEXT NOT NULL,
  metadata  TEXT NOT NULL
            CHECK (json_valid(metadata)),

  source_uri TEXT GENERATED ALWAYS AS (
    json_extract(metadata, '$.source.uri')
  ) VIRTUAL,

  media_type TEXT GENERATED ALWAYS AS (
    json_extract(metadata, '$.mediaType')
  ) STORED
) STRICT;

CREATE INDEX document_source_uri_idx
ON document(tenant_id, source_uri)
WHERE source_uri IS NOT NULL;

CREATE INDEX memory_active_lookup_idx
ON memory(
  tenant_id,
  kind,
  updated_at DESC,
  confidence,
  importance
)
WHERE status = 'active';

EXPLAIN QUERY PLAN
SELECT
  id,
  content,
  confidence,
  importance,
  updated_at
FROM memory
WHERE tenant_id = :tenant_id
  AND status = 'active'
  AND kind = :kind
ORDER BY updated_at DESC
LIMIT 25;

CREATE TABLE document_chunk (
  id          INTEGER PRIMARY KEY,
  tenant_id   TEXT NOT NULL,
  document_id TEXT NOT NULL,
  heading     TEXT,
  content     TEXT NOT NULL,
  created_at  INTEGER NOT NULL
) STRICT;

CREATE VIRTUAL TABLE document_chunk_fts USING fts5(
  heading,
  content,
  content='document_chunk',
  content_rowid='id',
  tokenize='unicode61 remove_diacritics 2',
  prefix='2 3'
);

CREATE TRIGGER document_chunk_fts_ai
AFTER INSERT ON document_chunk BEGIN
  INSERT INTO document_chunk_fts(rowid, heading, content)
  VALUES (new.id, new.heading, new.content);
END;

CREATE TRIGGER document_chunk_fts_ad
AFTER DELETE ON document_chunk BEGIN
  INSERT INTO document_chunk_fts(
    document_chunk_fts, rowid, heading, content
  ) VALUES (
    'delete', old.id, old.heading, old.content
  );
END;

interface LexicalResult {
  id: number;
  documentId: string;
  heading: string | null;
  snippet: string;
  rank: number;
}

const lexicalSearch = db.query<
  LexicalResult,
  { query: string; tenantId: string; limit: number }
>(`
  SELECT
    chunk.id,
    chunk.document_id AS documentId,
    chunk.heading,
    snippet(
      document_chunk_fts,
      1,
      '<mark>',
      '</mark>',
      ' … ',
      24
    ) AS snippet,
    bm25(document_chunk_fts, 2.0, 1.0) AS rank
  FROM document_chunk_fts
  JOIN document_chunk AS chunk
    ON chunk.id = document_chunk_fts.rowid
  WHERE document_chunk_fts MATCH $query
    AND chunk.tenant_id = $tenantId
  ORDER BY rank
  LIMIT $limit
`);

-- Validate consistency with an external-content table.
INSERT INTO document_chunk_fts(document_chunk_fts)
VALUES ('integrity-check');

-- Reconstruct the FTS projection from canonical content.
INSERT INTO document_chunk_fts(document_chunk_fts)
VALUES ('rebuild');

-- Merge index segments.
INSERT INTO document_chunk_fts(document_chunk_fts)
VALUES ('optimize');

bun add sqlite-vec

import { Database } from "bun:sqlite";
import * as sqliteVec from "sqlite-vec";

const DIMENSIONS = 384;

if (process.platform === "darwin") {
  Database.setCustomSQLite(
    process.env.SQLITE_LIBRARY_PATH ??
      "/opt/homebrew/opt/sqlite/lib/libsqlite3.dylib"
  );
}

const db = new Database("./data/knowledge.sqlite", {
  create: true,
  strict: true,
  safeIntegers: true,
});

sqliteVec.load(db);

db.exec(`
  CREATE TABLE chunk (
    id               INTEGER PRIMARY KEY,
    tenant_id        TEXT NOT NULL,
    document_id      TEXT NOT NULL,
    content          TEXT NOT NULL,
    embedding_model  TEXT NOT NULL,
    created_at       INTEGER NOT NULL
  ) STRICT;

  CREATE VIRTUAL TABLE chunk_vec USING vec0(
    chunk_id INTEGER PRIMARY KEY,
    tenant_id TEXT PARTITION KEY,
    embedding FLOAT[${DIMENSIONS}]
      DISTANCE_METRIC=cosine
  );
`);

const insertChunk = db.query(`
  INSERT INTO chunk(
    tenant_id, document_id, content,
    embedding_model, created_at
  ) VALUES (
    $tenantId, $documentId, $content,
    $embeddingModel, $createdAt
  )
  RETURNING id
`);

const insertVector = db.query(`
  INSERT INTO chunk_vec(chunk_id, tenant_id, embedding)
  VALUES ($chunkId, $tenantId, $embedding)
`);

const writeChunk = db.transaction((input: ChunkInput) => {
  if (input.embedding.length !== DIMENSIONS) {
    throw new RangeError("Unexpected embedding dimensions");
  }

  const row = insertChunk.get({
    ...input,
    createdAt: Date.now(),
  }) as { id: number };

  insertVector.run({
    chunkId: row.id,
    tenantId: input.tenantId,
    embedding: input.embedding,
  });

  return row.id;
});

WITH nearest AS (
  SELECT chunk_id, distance
  FROM chunk_vec
  WHERE embedding MATCH :query_embedding
    AND tenant_id = :tenant_id
    AND k = :candidate_count
  ORDER BY distance
)
SELECT
  chunk.id,
  chunk.document_id,
  chunk.content,
  nearest.distance
FROM nearest
JOIN chunk ON chunk.id = nearest.chunk_id
WHERE chunk.tenant_id = :tenant_id
ORDER BY nearest.distance
LIMIT :final_count;

interface RankedCandidate { id: number; rank: number }

export function reciprocalRankFusion(
  lexical: readonly RankedCandidate[],
  semantic: readonly RankedCandidate[],
  constant = 60,
): Array<{ id: number; score: number }> {
  const scores = new Map<number, number>();

  for (const list of [lexical, semantic]) {
    for (const candidate of list) {
      scores.set(
        candidate.id,
        (scores.get(candidate.id) ?? 0) +
          1 / (constant + candidate.rank),
      );
    }
  }

  return [...scores]
    .map(([id, score]) => ({ id, score }))
    .sort((a, b) => b.score - a.score);
}

CREATE TABLE source_event (
  id           TEXT PRIMARY KEY,
  tenant_id    TEXT NOT NULL,
  source_type  TEXT NOT NULL,
  payload      TEXT NOT NULL CHECK (json_valid(payload)),
  content_hash TEXT NOT NULL,
  observed_at  INTEGER NOT NULL,
  ingested_at  INTEGER NOT NULL,
  UNIQUE (tenant_id, content_hash)
) STRICT;

CREATE TABLE memory (
  id             TEXT PRIMARY KEY,
  tenant_id      TEXT NOT NULL,
  kind           TEXT NOT NULL,
  content        TEXT NOT NULL,
  metadata       TEXT NOT NULL DEFAULT '{}'
                 CHECK (json_valid(metadata)),
  confidence     REAL NOT NULL CHECK (confidence BETWEEN 0 AND 1),
  importance     REAL NOT NULL CHECK (importance BETWEEN 0 AND 1),
  status         TEXT NOT NULL CHECK (
    status IN ('candidate','active','superseded','retracted','expired')
  ),
  observed_at    INTEGER NOT NULL,
  valid_from     INTEGER,
  valid_until    INTEGER,
  expires_at     INTEGER,
  supersedes_id  TEXT REFERENCES memory(id),
  source_event_id TEXT REFERENCES source_event(id),
  created_at     INTEGER NOT NULL,
  updated_at     INTEGER NOT NULL
) STRICT;

CREATE TABLE memory_revision (
  memory_id       TEXT NOT NULL REFERENCES memory(id),
  revision        INTEGER NOT NULL,
  content         TEXT NOT NULL,
  metadata        TEXT NOT NULL CHECK (json_valid(metadata)),
  confidence      REAL NOT NULL,
  source_event_id TEXT REFERENCES source_event(id),
  created_at      INTEGER NOT NULL,
  PRIMARY KEY (memory_id, revision)
) WITHOUT ROWID, STRICT;

CREATE TABLE entity (
  id             TEXT PRIMARY KEY,
  tenant_id      TEXT NOT NULL,
  entity_type    TEXT NOT NULL,
  canonical_name TEXT NOT NULL,
  description    TEXT,
  metadata       TEXT NOT NULL DEFAULT '{}'
                 CHECK (json_valid(metadata)),
  UNIQUE (tenant_id, entity_type, canonical_name)
) STRICT;

CREATE TABLE relation (
  id            TEXT PRIMARY KEY,
  tenant_id     TEXT NOT NULL,
  source_id     TEXT NOT NULL REFERENCES entity(id),
  target_id     TEXT NOT NULL REFERENCES entity(id),
  relation_type TEXT NOT NULL,
  weight        REAL NOT NULL DEFAULT 1,
  confidence    REAL NOT NULL DEFAULT 1,
  valid_from    INTEGER,
  valid_until   INTEGER,
  evidence_id   TEXT REFERENCES source_event(id)
) STRICT;

CREATE INDEX relation_outgoing_idx
ON relation(tenant_id, source_id, relation_type, target_id);

CREATE INDEX relation_incoming_idx
ON relation(tenant_id, target_id, relation_type, source_id);

namespace = sqlite://<platform>/<logical-database-identity>
name      = <attached-schema>.<table-or-logical-index>

sqlite://core-platform-ai/knowledge-prod :: main.document
sqlite://core-platform-ai/knowledge-prod :: main.chunk
sqlite://core-platform-ai/knowledge-prod :: main.chunk_vec

interface OpenLineageEvent {
  eventType: "START" | "COMPLETE" | "FAIL";
  eventTime: string;
  run: { runId: string; facets?: Record<string, unknown> };
  job: { namespace: string; name: string };
  inputs?: Array<{ namespace: string; name: string }>;
  outputs?: Array<{ namespace: string; name: string }>;
  producer: string;
  schemaURL: string;
}

export async function emitLineage(
  baseUrl: string,
  event: OpenLineageEvent,
  signal?: AbortSignal,
): Promise<void> {
  const response = await fetch(
    new URL("/api/v1/lineage", baseUrl),
    {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify(event),
      signal,
    },
  );

  if (!response.ok) {
    throw new Error(
      `Lineage emission failed: ${response.status}`
    );
  }
}

CREATE TABLE lineage_outbox (
  id              TEXT PRIMARY KEY,
  event_json      TEXT NOT NULL CHECK (json_valid(event_json)),
  status          TEXT NOT NULL DEFAULT 'pending'
                  CHECK (status IN ('pending','sending','sent','failed')),
  attempts        INTEGER NOT NULL DEFAULT 0,
  next_attempt_at INTEGER NOT NULL,
  created_at      INTEGER NOT NULL,
  sent_at         INTEGER
) STRICT;

CREATE INDEX lineage_outbox_pending_idx
ON lineage_outbox(status, next_attempt_at)
WHERE status IN ('pending', 'failed');

{
  "_producer": "https://example.com/core-platform-ai/sqlite-lineage",
  "_schemaURL": "https://schemas.example.com/openlineage/vector-index/1-0-0.json",
  "embeddingModel": "embedding-v3",
  "embeddingModelRevision": "2026-07-01",
  "dimensions": 768,
  "distanceMetric": "cosine",
  "normalization": "l2",
  "vectorEngine": "sqlite-vec",
  "indexType": "flat-exact",
  "chunkingStrategy": "semantic-section",
  "generationId": "embedding-generation-2026-07"
}

Provisioned capacity must include:

main database
+ WAL or rollback journal
+ temporary sort and index files
+ table-rebuild migration copy
+ backup or snapshot staging
+ vector/FTS rebuild space
+ filesystem safety reserve

-- Inspect checkpoint progress without aggressively blocking.
PRAGMA wal_checkpoint(PASSIVE);

-- Controlled maintenance only.
PRAGMA wal_checkpoint(RESTART);
PRAGMA wal_checkpoint(TRUNCATE);

-- Automatic threshold in WAL pages.
PRAGMA wal_autocheckpoint = 1000;

application.sqlite
application.sqlite-wal
application.sqlite-shm
application.sqlite-journal   # rollback mode / recovery scenarios

// workers/read-worker.ts
import { Database } from "bun:sqlite";

const db = new Database("./data/application.sqlite", {
  readonly: true,
  strict: true,
  safeIntegers: true,
});

db.exec(`
  PRAGMA query_only = ON;
  PRAGMA busy_timeout = 5000;
`);

self.onmessage = (event: MessageEvent<SearchRequest>) => {
  try {
    const rows = executeBoundedSearch(db, event.data);
    self.postMessage({
      requestId: event.data.requestId,
      ok: true,
      rows,
    });
  } catch (error) {
    self.postMessage({
      requestId: event.data.requestId,
      ok: false,
      error: error instanceof Error ? error.message : String(error),
    });
  }
};

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: knowledge-api
spec:
  serviceName: knowledge-api
  replicas: 1
  selector:
    matchLabels:
      app: knowledge-api
  template:
    metadata:
      labels:
        app: knowledge-api
    spec:
      containers:
        - name: app
          image: example/knowledge-api:1.0.0
          volumeMounts:
            - name: database
              mountPath: /var/lib/knowledge
        - name: litestream
          image: litestream/litestream:latest
          args: ["replicate"]
          volumeMounts:
            - name: database
              mountPath: /var/lib/knowledge
  volumeClaimTemplates:
    - metadata:
        name: database
      spec:
        accessModes: ["ReadWriteOncePod"]
        resources:
          requests:
            storage: 20Gi

import { DatabaseSync, constants } from "node:sqlite";

const db = new DatabaseSync("./data/import.sqlite", {
  readOnly: true,
});

db.enableDefensive(true);
db.exec(`
  PRAGMA trusted_schema = OFF;
  PRAGMA query_only = ON;
  PRAGMA cell_size_check = ON;
  PRAGMA mmap_size = 0;
`);

const allowedTables = new Set([
  "document",
  "document_chunk",
  "document_chunk_fts",
]);

db.setAuthorizer((action, arg1) => {
  if (
    action === constants.SQLITE_ATTACH ||
    action === constants.SQLITE_DETACH ||
    action === constants.SQLITE_PRAGMA ||
    action === constants.SQLITE_INSERT ||
    action === constants.SQLITE_UPDATE ||
    action === constants.SQLITE_DELETE
  ) {
    return constants.SQLITE_DENY;
  }

  if (
    action === constants.SQLITE_READ &&
    arg1 !== null &&
    !allowedTables.has(arg1)
  ) {
    return constants.SQLITE_DENY;
  }

  return constants.SQLITE_OK;
});

export interface KnowledgeRepository {
  transact<T>(operation: () => T): T;

  writeMemory(input: NewMemory): void;

  lexicalSearch(
    input: LexicalSearchRequest,
  ): readonly RetrievalCandidate[];

  vectorSearch(
    input: VectorSearchRequest,
  ): readonly RetrievalCandidate[];

  traverseEntities(
    input: GraphSearchRequest,
  ): readonly GraphCandidate[];

  health(): RepositoryHealth;
  close(): void;
}

BunSqliteKnowledgeRepository
NodeSqliteKnowledgeRepository
TursoKnowledgeRepository
PostgresKnowledgeRepository
InMemoryKnowledgeRepository  # tests only

import { Database } from "bun:sqlite";

if (process.platform === "darwin") {
  Database.setCustomSQLite(
    process.env.SQLITE_LIBRARY_PATH ??
      "/opt/homebrew/opt/sqlite/lib/libsqlite3.dylib"
  );
}

const db = new Database("./data/application.sqlite");
db.loadExtension("./native/approved-extension");

import { DatabaseSync } from "node:sqlite";

const db = new DatabaseSync("./data/application.sqlite");

db.function(
  "normalize_identifier",
  { deterministic: true, directOnly: true },
  (value: unknown): string | null => {
    if (typeof value !== "string") return null;
    return value
      .normalize("NFKC")
      .trim()
      .toLowerCase()
      .replace(/[^a-z0-9]+/g, "-")
      .replace(/^-|-$/g, "");
  },
);

-- Fast routine validation.
PRAGMA quick_check;
PRAGMA foreign_key_check;

-- More expensive full validation.
PRAGMA integrity_check;

-- Planner statistics maintenance.
PRAGMA optimize;

-- Controlled compact snapshot.
VACUUM INTO 'knowledge-snapshot.sqlite';

SELECT sqlite_version();
PRAGMA compile_options;
PRAGMA journal_mode;
PRAGMA synchronous;
PRAGMA foreign_keys;
PRAGMA trusted_schema;
PRAGMA busy_timeout;
PRAGMA page_size;
PRAGMA page_count;
PRAGMA freelist_count;
PRAGMA database_list;

PRAGMA foreign_keys = ON;
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;
PRAGMA busy_timeout = 5000;
PRAGMA trusted_schema = OFF;

EXPLAIN QUERY PLAN
SELECT ...;

PRAGMA table_xinfo('memory');
PRAGMA index_list('memory');
PRAGMA index_xinfo('memory_active_lookup_idx');
PRAGMA foreign_key_check;
PRAGMA quick_check;
PRAGMA optimize;

.timer on
.stats on
.eqp full
.scanstats on
.expert --sample 100

sqlite3 knowledge.sqlite ".backup main knowledge-backup.sqlite"

sqlite3 corrupt.sqlite ".recover" > recovered.sql
sqlite3 recovered.sqlite < recovered.sql

Authoritative shared records  → PostgreSQL / distributed SQL
Vector retrieval              → specialized vector service if required
Graph analytics               → graph engine or offline job
Agent-local memory            → SQLite
Edge/read cache               → SQLite snapshot or embedded replica
Portable export               → SQLite