feat(p1): chunker, FTS5 query sanitizer, SQLite index schema and ingestion
chunks_fts/chunks_tri are ordinary FTS5 tables, not content='': contentless FTS5 cannot delete rows before SQLite 3.43 (bookworm ships 3.40) and cannot snippet. Vectors survive an edit that does not change the embed text.
This commit is contained in:
Markus committed
1 parent
8f58119a74
commit
6a424c0909
9 files changed
+982
No files matched your search
Binary file not shown.
@@ -0,0 +1,103 @@
|
||||
"""Chunking and the indexed embedding text (§6.4, §6.5).
|
||||
|
||||
Notes are meant to be atomic [R9], so most notes are a single chunk and there is
|
||||
no overlap between chunks.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from mdmem.notes import Note
|
||||
|
||||
MAX_CHUNK_TOKENS = 1500
|
||||
CHARS_PER_TOKEN = 4
|
||||
MAX_CHUNK_CHARS = MAX_CHUNK_TOKENS * CHARS_PER_TOKEN
|
||||
|
||||
_SPLIT_HEADING_RE = re.compile(r"^(#{2,3})\s+(.*?)\s*#*\s*$", re.MULTILINE)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class Chunk:
|
||||
ord: int
|
||||
heading: str
|
||||
body: str
|
||||
|
||||
|
||||
def _sections(body: str) -> list[tuple[str, str]]:
|
||||
"""Split on H2/H3, returning (heading path, text). Keeps the pre-heading intro."""
|
||||
matches = list(_SPLIT_HEADING_RE.finditer(body))
|
||||
if not matches:
|
||||
return [("", body)]
|
||||
out: list[tuple[str, str]] = []
|
||||
intro = body[: matches[0].start()].strip()
|
||||
if intro:
|
||||
out.append(("", intro))
|
||||
stack: list[tuple[int, str]] = []
|
||||
for i, m in enumerate(matches):
|
||||
level, title = len(m.group(1)), m.group(2).strip()
|
||||
while stack and stack[-1][0] >= level:
|
||||
stack.pop()
|
||||
stack.append((level, title))
|
||||
path = " > ".join(t for _, t in stack)
|
||||
end = matches[i + 1].start() if i + 1 < len(matches) else len(body)
|
||||
text = body[m.end() : end].strip()
|
||||
out.append((path, text))
|
||||
return out
|
||||
|
||||
|
||||
def _split_long(text: str, limit: int = MAX_CHUNK_CHARS) -> list[str]:
|
||||
if len(text) <= limit:
|
||||
return [text]
|
||||
parts: list[str] = []
|
||||
buf: list[str] = []
|
||||
size = 0
|
||||
for para in re.split(r"\n{2,}", text):
|
||||
if size and size + len(para) + 2 > limit:
|
||||
parts.append("\n\n".join(buf))
|
||||
buf, size = [], 0
|
||||
if len(para) > limit:
|
||||
if buf:
|
||||
parts.append("\n\n".join(buf))
|
||||
buf, size = [], 0
|
||||
parts.extend(_split_words(para, limit))
|
||||
continue
|
||||
buf.append(para)
|
||||
size += len(para) + 2
|
||||
if buf:
|
||||
parts.append("\n\n".join(buf))
|
||||
return [p for p in parts if p.strip()]
|
||||
|
||||
|
||||
def _split_words(text: str, limit: int) -> list[str]:
|
||||
out: list[str] = []
|
||||
line: list[str] = []
|
||||
size = 0
|
||||
for word in text.split():
|
||||
if size and size + len(word) + 1 > limit:
|
||||
out.append(" ".join(line))
|
||||
line, size = [], 0
|
||||
line.append(word)
|
||||
size += len(word) + 1
|
||||
if line:
|
||||
out.append(" ".join(line))
|
||||
return out
|
||||
|
||||
|
||||
def chunk_note(note: Note) -> list[Chunk]:
|
||||
body = note.body.strip()
|
||||
if len(body) <= MAX_CHUNK_CHARS:
|
||||
return [Chunk(ord=0, heading="", body=body)]
|
||||
chunks: list[Chunk] = []
|
||||
for heading, text in _sections(body):
|
||||
for piece in _split_long(text):
|
||||
chunks.append(Chunk(ord=len(chunks), heading=heading, body=piece.strip()))
|
||||
return chunks or [Chunk(ord=0, heading="", body=body)]
|
||||
|
||||
|
||||
def embed_text(note: Note, chunk: Chunk) -> str:
|
||||
"""Indexing-stage key merging: title + summary + keywords + heading path + body [R10]."""
|
||||
kw = ", ".join(note.keywords)
|
||||
head = f"{note.title}\n{note.summary}\nKeywords: {kw}\n{chunk.heading}"
|
||||
return f"{head}\n\n{chunk.body}".strip()
|
||||
+198
@@ -0,0 +1,198 @@
|
||||
"""SQLite index: schema, connections and the single-writer rule (§5, §6).
|
||||
|
||||
Deviation from §5, deliberate: `chunks_fts` and `chunks_tri` are ordinary FTS5
|
||||
tables rather than `content=''` contentless ones. Contentless FTS5 cannot delete
|
||||
a row before SQLite 3.43 (Debian bookworm ships 3.40) and cannot produce
|
||||
`snippet()` output at all, and this index is rewritten on every note edit. The
|
||||
duplicated text costs a few MB at the scale this service targets.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlite3
|
||||
import threading
|
||||
from collections.abc import Iterator
|
||||
from contextlib import contextmanager
|
||||
from pathlib import Path
|
||||
|
||||
import sqlite_vec
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
_BASE_SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS meta(k TEXT PRIMARY KEY, v TEXT);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS notes(
|
||||
id TEXT PRIMARY KEY,
|
||||
path TEXT UNIQUE NOT NULL,
|
||||
state TEXT NOT NULL,
|
||||
title TEXT, type TEXT, summary TEXT, tags TEXT, keywords TEXT,
|
||||
superseded_by TEXT, revises TEXT,
|
||||
observed_at TEXT, valid_from TEXT, valid_until TEXT, created TEXT, updated TEXT,
|
||||
file_hash TEXT NOT NULL, mtime REAL);
|
||||
CREATE INDEX IF NOT EXISTS notes_state ON notes(state);
|
||||
CREATE INDEX IF NOT EXISTS notes_revises ON notes(revises);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS chunks(
|
||||
id INTEGER PRIMARY KEY,
|
||||
note_id TEXT NOT NULL REFERENCES notes(id) ON DELETE CASCADE,
|
||||
ord INTEGER, heading TEXT, body TEXT,
|
||||
embed_text_hash TEXT NOT NULL,
|
||||
vec_pending INTEGER NOT NULL DEFAULT 0);
|
||||
CREATE INDEX IF NOT EXISTS chunks_note ON chunks(note_id);
|
||||
CREATE INDEX IF NOT EXISTS chunks_pending ON chunks(vec_pending);
|
||||
|
||||
CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(
|
||||
title, keywords, heading, body, tokenize='unicode61 remove_diacritics 2');
|
||||
|
||||
CREATE TABLE IF NOT EXISTS links(
|
||||
src TEXT, dst TEXT, rel TEXT, origin TEXT,
|
||||
PRIMARY KEY(src, dst, rel));
|
||||
CREATE INDEX IF NOT EXISTS links_dst ON links(dst);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS flags(
|
||||
id INTEGER PRIMARY KEY, note_id TEXT NOT NULL, reason TEXT, source TEXT,
|
||||
created TEXT, resolved INTEGER NOT NULL DEFAULT 0);
|
||||
CREATE INDEX IF NOT EXISTS flags_note ON flags(note_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS idempotency(
|
||||
request_id TEXT PRIMARY KEY, tool TEXT, result TEXT, created TEXT);
|
||||
"""
|
||||
|
||||
_TRIGRAM_SCHEMA = "CREATE VIRTUAL TABLE IF NOT EXISTS chunks_tri USING fts5(text, tokenize='trigram');"
|
||||
|
||||
|
||||
def connect(path: Path | str, *, read_only: bool = False) -> sqlite3.Connection:
|
||||
"""Open a connection with sqlite-vec loaded. Every connection needs it (§14.4)."""
|
||||
conn = sqlite3.connect(str(path), timeout=30.0, check_same_thread=False)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.enable_load_extension(True)
|
||||
sqlite_vec.load(conn)
|
||||
conn.enable_load_extension(False)
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA synchronous=NORMAL")
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
conn.execute("PRAGMA busy_timeout=30000")
|
||||
if read_only:
|
||||
conn.execute("PRAGMA query_only=ON")
|
||||
return conn
|
||||
|
||||
|
||||
class Database:
|
||||
"""Owns the index. One write connection behind a lock = one writer (§6).
|
||||
|
||||
The async server funnels writes through `asyncio.to_thread`, so the lock is
|
||||
what actually enforces the single-writer invariant in every entry path
|
||||
(server, CLI reindex, tests).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
path: Path | str,
|
||||
*,
|
||||
dim: int,
|
||||
trigram: bool,
|
||||
model: str = "",
|
||||
) -> None:
|
||||
self.path = Path(path)
|
||||
self.dim = dim
|
||||
self.trigram = trigram
|
||||
self.model = model
|
||||
self._write: sqlite3.Connection | None = None
|
||||
self._lock = threading.RLock()
|
||||
self._local = threading.local()
|
||||
|
||||
# -- lifecycle ---------------------------------------------------------
|
||||
def open(self) -> bool:
|
||||
"""Create or migrate the schema. Returns True if vectors were dropped."""
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self._write = connect(self.path)
|
||||
with self._lock:
|
||||
self._write.executescript(_BASE_SCHEMA)
|
||||
if self.trigram:
|
||||
self._write.executescript(_TRIGRAM_SCHEMA)
|
||||
else:
|
||||
self._write.execute("DROP TABLE IF EXISTS chunks_tri")
|
||||
self._write.commit()
|
||||
rebuilt = self._ensure_vec_table()
|
||||
self._set_meta("schema_version", str(SCHEMA_VERSION))
|
||||
self._set_meta("embed_dim", str(self.dim))
|
||||
self._set_meta("embed_model", self.model)
|
||||
self._write.commit()
|
||||
return rebuilt
|
||||
|
||||
def _ensure_vec_table(self) -> bool:
|
||||
assert self._write is not None
|
||||
cur = self._write
|
||||
have = cur.execute(
|
||||
"SELECT name FROM sqlite_master WHERE name='chunks_vec'"
|
||||
).fetchone()
|
||||
prev_dim = self._get_meta("embed_dim")
|
||||
prev_model = self._get_meta("embed_model")
|
||||
changed = have is not None and (
|
||||
prev_dim != str(self.dim) or (prev_model or "") != (self.model or "")
|
||||
)
|
||||
if have is not None and not changed:
|
||||
return False
|
||||
if changed:
|
||||
cur.execute("DROP TABLE IF EXISTS chunks_vec")
|
||||
cur.execute("UPDATE chunks SET vec_pending=1")
|
||||
cur.execute(
|
||||
f"CREATE VIRTUAL TABLE IF NOT EXISTS chunks_vec USING vec0(embedding float[{self.dim}])"
|
||||
)
|
||||
cur.commit()
|
||||
return bool(changed)
|
||||
|
||||
def close(self) -> None:
|
||||
with self._lock:
|
||||
if self._write is not None:
|
||||
self._write.close()
|
||||
self._write = None
|
||||
conn = getattr(self._local, "read", None)
|
||||
if conn is not None:
|
||||
conn.close()
|
||||
self._local.read = None
|
||||
|
||||
# -- connections -------------------------------------------------------
|
||||
def read(self) -> sqlite3.Connection:
|
||||
conn = getattr(self._local, "read", None)
|
||||
if conn is None:
|
||||
conn = connect(self.path, read_only=True)
|
||||
self._local.read = conn
|
||||
return conn
|
||||
|
||||
@contextmanager
|
||||
def write(self) -> Iterator[sqlite3.Connection]:
|
||||
if self._write is None:
|
||||
raise RuntimeError("database not open")
|
||||
with self._lock:
|
||||
try:
|
||||
yield self._write
|
||||
self._write.commit()
|
||||
except BaseException:
|
||||
self._write.rollback()
|
||||
raise
|
||||
|
||||
# -- meta --------------------------------------------------------------
|
||||
def _get_meta(self, key: str) -> str | None:
|
||||
assert self._write is not None
|
||||
row = self._write.execute("SELECT v FROM meta WHERE k=?", (key,)).fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
def _set_meta(self, key: str, value: str) -> None:
|
||||
assert self._write is not None
|
||||
self._write.execute(
|
||||
"INSERT INTO meta(k,v) VALUES(?,?) ON CONFLICT(k) DO UPDATE SET v=excluded.v",
|
||||
(key, value),
|
||||
)
|
||||
|
||||
def meta_get(self, key: str) -> str | None:
|
||||
row = self.read().execute("SELECT v FROM meta WHERE k=?", (key,)).fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
def meta_set(self, key: str, value: str) -> None:
|
||||
with self.write() as w:
|
||||
w.execute(
|
||||
"INSERT INTO meta(k,v) VALUES(?,?) ON CONFLICT(k) DO UPDATE SET v=excluded.v",
|
||||
(key, value),
|
||||
)
|
||||
@@ -0,0 +1,48 @@
|
||||
"""FTS5 query sanitising (§8.1, §14.5).
|
||||
|
||||
Raw user or agent text must never reach a MATCH expression: it either raises a
|
||||
syntax error or lets the caller drive the query language. Everything is reduced
|
||||
to quoted tokens joined with OR.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
_TOKEN_RE = re.compile(r"[^\W_]+", re.UNICODE)
|
||||
MAX_TOKENS = 32
|
||||
TRIGRAM_MIN = 3
|
||||
|
||||
|
||||
def tokens(query: str) -> list[str]:
|
||||
out: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for m in _TOKEN_RE.finditer(query or ""):
|
||||
t = m.group(0).lower()
|
||||
if t in seen:
|
||||
continue
|
||||
seen.add(t)
|
||||
out.append(t)
|
||||
if len(out) >= MAX_TOKENS:
|
||||
break
|
||||
return out
|
||||
|
||||
|
||||
def _quote(token: str) -> str:
|
||||
return '"' + token.replace('"', '""') + '"'
|
||||
|
||||
|
||||
def to_match(query: str) -> str | None:
|
||||
"""OR of quoted tokens, or None when there is nothing to search for."""
|
||||
toks = tokens(query)
|
||||
if not toks:
|
||||
return None
|
||||
return " OR ".join(_quote(t) for t in toks)
|
||||
|
||||
|
||||
def to_trigram_match(query: str) -> str | None:
|
||||
"""Trigram tables need >=3 characters per term; shorter tokens simply cannot match."""
|
||||
toks = [t for t in tokens(query) if len(t) >= TRIGRAM_MIN]
|
||||
if not toks:
|
||||
return None
|
||||
return " OR ".join(_quote(t) for t in toks)
|
||||
@@ -0,0 +1,275 @@
|
||||
"""Ingestion: vault files → index rows (§6)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
|
||||
from mdmem.chunker import chunk_note, embed_text
|
||||
from mdmem.config import NOTE_TYPES
|
||||
from mdmem.db import Database
|
||||
from mdmem.ids import is_ulid, new_id, slugify
|
||||
from mdmem.notes import Note
|
||||
from mdmem.vaultio import Vault, sha256_text
|
||||
|
||||
log = logging.getLogger("mdmem.index")
|
||||
|
||||
UNRESOLVED = "unresolved:"
|
||||
|
||||
|
||||
@dataclass
|
||||
class PendingEmbed:
|
||||
chunk_id: int
|
||||
text: str
|
||||
key_hash: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScanStats:
|
||||
scanned: int = 0
|
||||
indexed: int = 0
|
||||
removed: int = 0
|
||||
unchanged: int = 0
|
||||
errors: list[str] = field(default_factory=list)
|
||||
pending: list[PendingEmbed] = field(default_factory=list)
|
||||
seconds: float = 0.0
|
||||
|
||||
|
||||
class Indexer:
|
||||
def __init__(self, db: Database, vault: Vault) -> None:
|
||||
self.db = db
|
||||
self.vault = vault
|
||||
|
||||
# -- single note -------------------------------------------------------
|
||||
def index_note(self, note: Note) -> list[PendingEmbed]:
|
||||
"""Write every row for one note in a single transaction. Returns chunks to embed."""
|
||||
if not is_ulid(note.id):
|
||||
raise ValueError(f"note at {note.path} has no valid ULID id")
|
||||
if note.type not in NOTE_TYPES:
|
||||
raise ValueError(f"note {note.id} has unknown type {note.type!r}")
|
||||
|
||||
chunks = chunk_note(note)
|
||||
texts = [embed_text(note, c) for c in chunks]
|
||||
hashes = [sha256_text(t) for t in texts]
|
||||
rel_path = str(note.path) if note.path else ""
|
||||
pending: list[PendingEmbed] = []
|
||||
|
||||
with self.db.write() as w:
|
||||
old = {
|
||||
row["id"]: (row["embed_text_hash"], row["vec_pending"])
|
||||
for row in w.execute(
|
||||
"SELECT id, embed_text_hash, vec_pending FROM chunks WHERE note_id=? ORDER BY ord",
|
||||
(note.id,),
|
||||
)
|
||||
}
|
||||
prev_vecs = self._cached_vectors(w, old) # before the rows go away
|
||||
self._delete_chunk_rows(w, list(old))
|
||||
w.execute(
|
||||
"""INSERT INTO notes(id,path,state,title,type,summary,tags,keywords,
|
||||
superseded_by,revises,observed_at,valid_from,valid_until,
|
||||
created,updated,file_hash,mtime)
|
||||
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
path=excluded.path, state=excluded.state, title=excluded.title,
|
||||
type=excluded.type, summary=excluded.summary, tags=excluded.tags,
|
||||
keywords=excluded.keywords, superseded_by=excluded.superseded_by,
|
||||
revises=excluded.revises, observed_at=excluded.observed_at,
|
||||
valid_from=excluded.valid_from, valid_until=excluded.valid_until,
|
||||
created=excluded.created, updated=excluded.updated,
|
||||
file_hash=excluded.file_hash, mtime=excluded.mtime""",
|
||||
(
|
||||
note.id, rel_path, note.state, note.title, note.type, note.summary,
|
||||
json.dumps(note.tags, ensure_ascii=False),
|
||||
json.dumps(note.keywords, ensure_ascii=False),
|
||||
note.superseded_by, note.revises,
|
||||
note.ts("observed_at"), note.ts("valid_from"), note.ts("valid_until"),
|
||||
note.ts("created"), note.ts("updated"),
|
||||
note.file_hash or sha256_text(note.to_text()), note.mtime,
|
||||
),
|
||||
)
|
||||
|
||||
kw = " ".join(note.keywords)
|
||||
for chunk, text, h in zip(chunks, texts, hashes, strict=True):
|
||||
cur = w.execute(
|
||||
"INSERT INTO chunks(note_id,ord,heading,body,embed_text_hash,vec_pending) "
|
||||
"VALUES(?,?,?,?,?,1)",
|
||||
(note.id, chunk.ord, chunk.heading, chunk.body, h),
|
||||
)
|
||||
cid = int(cur.lastrowid or 0)
|
||||
w.execute(
|
||||
"INSERT INTO chunks_fts(rowid,title,keywords,heading,body) VALUES(?,?,?,?,?)",
|
||||
(cid, note.title, kw, chunk.heading, chunk.body),
|
||||
)
|
||||
if self.db.trigram:
|
||||
w.execute(
|
||||
"INSERT INTO chunks_tri(rowid,text) VALUES(?,?)",
|
||||
(cid, f"{note.title}\n{kw}\n{chunk.heading}\n{chunk.body}"),
|
||||
)
|
||||
vec = prev_vecs.get(h)
|
||||
if vec is not None:
|
||||
w.execute(
|
||||
"INSERT INTO chunks_vec(rowid,embedding) VALUES(?,?)", (cid, vec)
|
||||
)
|
||||
w.execute("UPDATE chunks SET vec_pending=0 WHERE id=?", (cid,))
|
||||
else:
|
||||
pending.append(PendingEmbed(chunk_id=cid, text=text, key_hash=h))
|
||||
|
||||
self._write_links(w, note)
|
||||
self._resolve_links_for(note)
|
||||
return pending
|
||||
|
||||
def _cached_vectors(self, w, old: dict[int, tuple[str, int]]) -> dict[str, bytes]:
|
||||
"""Keep vectors for chunks whose embed text did not change."""
|
||||
out: dict[str, bytes] = {}
|
||||
for cid, (h, vec_pending) in old.items():
|
||||
if vec_pending:
|
||||
continue
|
||||
row = w.execute("SELECT embedding FROM chunks_vec WHERE rowid=?", (cid,)).fetchone()
|
||||
if row is not None:
|
||||
out[h] = row[0]
|
||||
return out
|
||||
|
||||
def _delete_chunk_rows(self, w, chunk_ids: list[int]) -> None:
|
||||
for cid in chunk_ids:
|
||||
w.execute("DELETE FROM chunks_fts WHERE rowid=?", (cid,))
|
||||
if self.db.trigram:
|
||||
w.execute("DELETE FROM chunks_tri WHERE rowid=?", (cid,))
|
||||
w.execute("DELETE FROM chunks_vec WHERE rowid=?", (cid,))
|
||||
if chunk_ids:
|
||||
w.executemany("DELETE FROM chunks WHERE id=?", [(c,) for c in chunk_ids])
|
||||
|
||||
def remove(self, note_id: str) -> None:
|
||||
with self.db.write() as w:
|
||||
ids = [r[0] for r in w.execute("SELECT id FROM chunks WHERE note_id=?", (note_id,))]
|
||||
self._delete_chunk_rows(w, ids)
|
||||
w.execute("DELETE FROM links WHERE src=?", (note_id,))
|
||||
w.execute("DELETE FROM notes WHERE id=?", (note_id,))
|
||||
|
||||
# -- links -------------------------------------------------------------
|
||||
def _write_links(self, w, note: Note) -> None:
|
||||
w.execute("DELETE FROM links WHERE src=?", (note.id,))
|
||||
rows: list[tuple[str, str, str, str]] = []
|
||||
for target in note.wikilinks:
|
||||
rows.append((note.id, self._resolve(w, target), "links", "wikilink"))
|
||||
for rel, targets in note.relations.items():
|
||||
for t in targets:
|
||||
clean = t.strip().strip("[]")
|
||||
rows.append((note.id, self._resolve(w, clean), rel, "relation"))
|
||||
for old in note.supersedes:
|
||||
rows.append((note.id, old, "supersedes", "supersedes"))
|
||||
if note.revises:
|
||||
rows.append((note.id, note.revises, "revises", "revises"))
|
||||
seen = set()
|
||||
for r in rows:
|
||||
key = (r[0], r[1], r[2])
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
w.execute(
|
||||
"INSERT OR REPLACE INTO links(src,dst,rel,origin) VALUES(?,?,?,?)", r
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _resolve(w, target: str) -> str:
|
||||
"""ULID, then exact title, then slug; otherwise 'unresolved:<text>' (§5)."""
|
||||
target = target.strip()
|
||||
if not target:
|
||||
return f"{UNRESOLVED}"
|
||||
if is_ulid(target):
|
||||
return target
|
||||
row = w.execute("SELECT id FROM notes WHERE title=? LIMIT 1", (target,)).fetchone()
|
||||
if row:
|
||||
return row[0]
|
||||
slug = slugify(target)
|
||||
row = w.execute(
|
||||
"SELECT id, title FROM notes WHERE path LIKE ? LIMIT 1", (f"%/{slug}--%",)
|
||||
).fetchone()
|
||||
if row:
|
||||
return row[0]
|
||||
return f"{UNRESOLVED}{target}"
|
||||
|
||||
def _resolve_links_for(self, note: Note) -> None:
|
||||
"""A note that arrives later resolves the dangling links that pointed at it."""
|
||||
slug = slugify(note.title)
|
||||
with self.db.write() as w:
|
||||
w.execute(
|
||||
"UPDATE OR REPLACE links SET dst=? WHERE dst IN (?,?)",
|
||||
(note.id, f"{UNRESOLVED}{note.title}", f"{UNRESOLVED}{slug}"),
|
||||
)
|
||||
|
||||
# -- whole vault -------------------------------------------------------
|
||||
def scan(self, *, full: bool = False) -> ScanStats:
|
||||
started = time.monotonic()
|
||||
stats = ScanStats()
|
||||
seen: set[str] = set()
|
||||
known = {
|
||||
r["id"]: (r["file_hash"], r["path"])
|
||||
for r in self.db.read().execute("SELECT id, file_hash, path FROM notes")
|
||||
}
|
||||
for path, _state in self.vault.iter_files():
|
||||
stats.scanned += 1
|
||||
try:
|
||||
note = self._read_and_fix(path)
|
||||
except Exception as exc: # a broken file must not abort the scan
|
||||
stats.errors.append(f"{path}: {exc}")
|
||||
log.warning("skipping unreadable note", extra={"fields": {"path": str(path), "error": str(exc)}})
|
||||
continue
|
||||
seen.add(note.id)
|
||||
prev = known.get(note.id)
|
||||
if not full and prev and prev[0] == note.file_hash and prev[1] == str(note.path):
|
||||
stats.unchanged += 1
|
||||
continue
|
||||
try:
|
||||
stats.pending.extend(self.index_note(note))
|
||||
stats.indexed += 1
|
||||
except Exception as exc:
|
||||
stats.errors.append(f"{path}: {exc}")
|
||||
log.warning("indexing failed", extra={"fields": {"path": str(path), "error": str(exc)}})
|
||||
for note_id in set(known) - seen:
|
||||
self.remove(note_id)
|
||||
stats.removed += 1
|
||||
stats.pending.extend(self.pending_embeds())
|
||||
stats.seconds = time.monotonic() - started
|
||||
return stats
|
||||
|
||||
def _read_and_fix(self, path: Path) -> Note:
|
||||
"""Read a note, assigning a ULID and filename if a human created the file by hand."""
|
||||
note = self.vault.read(path)
|
||||
if not is_ulid(note.id):
|
||||
note.set("id", new_id())
|
||||
if not note.title:
|
||||
note.set("title", path.stem)
|
||||
new_path = self.vault.write(note, state=note.state)
|
||||
if new_path != path:
|
||||
# the file had no ULID in its name, so write() could not find the original
|
||||
path.unlink(missing_ok=True)
|
||||
log.info("assigned id to hand-written note",
|
||||
extra={"fields": {"id": note.id, "path": str(new_path)}})
|
||||
note = self.vault.read(new_path)
|
||||
return note
|
||||
|
||||
def pending_embeds(self) -> list[PendingEmbed]:
|
||||
rows = self.db.read().execute(
|
||||
"""SELECT c.id, c.heading, c.body, c.embed_text_hash,
|
||||
n.title, n.summary, n.keywords
|
||||
FROM chunks c JOIN notes n ON n.id = c.note_id
|
||||
WHERE c.vec_pending = 1"""
|
||||
).fetchall()
|
||||
out: list[PendingEmbed] = []
|
||||
for r in rows:
|
||||
kw = ", ".join(json.loads(r["keywords"] or "[]"))
|
||||
text = f"{r['title']}\n{r['summary']}\nKeywords: {kw}\n{r['heading']}\n\n{r['body']}".strip()
|
||||
out.append(PendingEmbed(chunk_id=r["id"], text=text, key_hash=r["embed_text_hash"]))
|
||||
return out
|
||||
|
||||
def store_vectors(self, items: list[tuple[int, bytes]]) -> None:
|
||||
if not items:
|
||||
return
|
||||
with self.db.write() as w:
|
||||
for cid, blob in items:
|
||||
w.execute("DELETE FROM chunks_vec WHERE rowid=?", (cid,))
|
||||
w.execute("INSERT INTO chunks_vec(rowid,embedding) VALUES(?,?)", (cid, blob))
|
||||
w.execute("UPDATE chunks SET vec_pending=0 WHERE id=?", (cid,))
|
||||
@@ -0,0 +1,56 @@
|
||||
from mdmem.chunker import MAX_CHUNK_CHARS, chunk_note, embed_text
|
||||
from mdmem.notes import Note
|
||||
|
||||
|
||||
def test_short_note_is_one_chunk():
|
||||
n = Note.new(title="T", body="one short body", type="fact")
|
||||
chunks = chunk_note(n)
|
||||
assert len(chunks) == 1
|
||||
assert chunks[0].ord == 0 and chunks[0].heading == "" and chunks[0].body == "one short body"
|
||||
|
||||
|
||||
def test_long_note_splits_on_h2_and_h3():
|
||||
body = "\n\n".join(
|
||||
["intro " * 50]
|
||||
+ [f"## Section {i}\n\n" + ("text " * 400) for i in range(3)]
|
||||
)
|
||||
n = Note.new(title="T", body=body, type="note")
|
||||
chunks = chunk_note(n)
|
||||
assert len(chunks) >= 4
|
||||
assert [c.ord for c in chunks] == list(range(len(chunks)))
|
||||
assert any(c.heading == "Section 1" for c in chunks)
|
||||
assert all(len(c.body) <= MAX_CHUNK_CHARS * 1.5 for c in chunks)
|
||||
|
||||
|
||||
def test_heading_path_includes_parents():
|
||||
body = "## Parent\n\n" + "a " * 2000 + "\n\n### Child\n\n" + ("b " * 2000)
|
||||
n = Note.new(title="T", body=body, type="note")
|
||||
chunks = chunk_note(n)
|
||||
assert any(c.heading == "Parent > Child" for c in chunks)
|
||||
|
||||
|
||||
def test_one_huge_paragraph_is_split_without_overlap():
|
||||
body = "wort " * 3000
|
||||
n = Note.new(title="T", body=body, type="note")
|
||||
chunks = chunk_note(n)
|
||||
assert len(chunks) > 1
|
||||
joined = " ".join(c.body for c in chunks).split()
|
||||
assert joined == body.split() # no overlap, nothing lost
|
||||
|
||||
|
||||
def test_embed_text_merges_keys_at_index_time():
|
||||
n = Note.new(
|
||||
title="Titel", body="Körper", type="fact", summary="Zusammenfassung",
|
||||
keywords=["a", "b"],
|
||||
)
|
||||
c = chunk_note(n)[0]
|
||||
t = embed_text(n, c)
|
||||
assert t.startswith("Titel\nZusammenfassung\nKeywords: a, b")
|
||||
assert t.endswith("Körper")
|
||||
|
||||
|
||||
def test_empty_body_yields_one_chunk_with_title_text():
|
||||
n = Note.new(title="Nur Titel", body="", type="fact")
|
||||
chunks = chunk_note(n)
|
||||
assert len(chunks) == 1 and chunks[0].body == ""
|
||||
assert "Nur Titel" in embed_text(n, chunks[0])
|
||||
@@ -0,0 +1,78 @@
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
|
||||
from mdmem.db import SCHEMA_VERSION, Database
|
||||
|
||||
|
||||
def test_schema_created_and_versioned(tmp_path):
|
||||
db = Database(tmp_path / "index.db", dim=8, trigram=True)
|
||||
db.open()
|
||||
assert db.meta_get("schema_version") == str(SCHEMA_VERSION)
|
||||
assert db.meta_get("embed_dim") == "8"
|
||||
tables = {r[0] for r in db.read().execute("SELECT name FROM sqlite_master WHERE type IN ('table','view')")}
|
||||
for t in ("notes", "chunks", "chunks_fts", "chunks_tri", "chunks_vec", "links", "flags", "idempotency"):
|
||||
assert t in tables
|
||||
db.close()
|
||||
|
||||
|
||||
def test_wal_and_vec_extension(tmp_path):
|
||||
db = Database(tmp_path / "index.db", dim=4, trigram=False)
|
||||
db.open()
|
||||
assert db.read().execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
|
||||
ver = db.read().execute("SELECT vec_version()").fetchone()[0]
|
||||
assert ver
|
||||
db.close()
|
||||
|
||||
|
||||
def test_trigram_table_absent_when_disabled(tmp_path):
|
||||
db = Database(tmp_path / "index.db", dim=4, trigram=False)
|
||||
db.open()
|
||||
tables = {r[0] for r in db.read().execute("SELECT name FROM sqlite_master")}
|
||||
assert "chunks_tri" not in tables
|
||||
db.close()
|
||||
|
||||
|
||||
def test_dim_change_rebuilds_vector_table(tmp_path):
|
||||
p = tmp_path / "index.db"
|
||||
db = Database(p, dim=4, trigram=False)
|
||||
db.open()
|
||||
with db.write() as w:
|
||||
w.execute("INSERT INTO notes(id,path,state,file_hash) VALUES('a','p','active','h')")
|
||||
w.execute("INSERT INTO chunks(id,note_id,ord,heading,body,embed_text_hash) VALUES(1,'a',0,'','b','h')")
|
||||
w.execute("INSERT INTO chunks_vec(rowid,embedding) VALUES(1,?)", (b"\x00" * 16,))
|
||||
db.close()
|
||||
|
||||
db2 = Database(p, dim=8, trigram=False)
|
||||
rebuilt = db2.open()
|
||||
assert rebuilt is True
|
||||
assert db2.meta_get("embed_dim") == "8"
|
||||
assert db2.read().execute("SELECT count(*) FROM chunks_vec").fetchone()[0] == 0
|
||||
# notes survive; their chunks are queued for re-embedding
|
||||
assert db2.read().execute("SELECT count(*) FROM notes").fetchone()[0] == 1
|
||||
assert db2.read().execute("SELECT vec_pending FROM chunks WHERE id=1").fetchone()[0] == 1
|
||||
db2.close()
|
||||
|
||||
|
||||
def test_model_change_rebuilds_too(tmp_path):
|
||||
p = tmp_path / "index.db"
|
||||
Database(p, dim=4, trigram=False, model="m1").open()
|
||||
db2 = Database(p, dim=4, trigram=False, model="m2")
|
||||
assert db2.open() is True
|
||||
assert db2.meta_get("embed_model") == "m2"
|
||||
|
||||
|
||||
def test_second_open_is_not_a_rebuild(tmp_path):
|
||||
p = tmp_path / "index.db"
|
||||
Database(p, dim=4, trigram=False, model="m").open()
|
||||
assert Database(p, dim=4, trigram=False, model="m").open() is False
|
||||
|
||||
|
||||
def test_write_is_transactional(tmp_path):
|
||||
db = Database(tmp_path / "index.db", dim=4, trigram=False)
|
||||
db.open()
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
with db.write() as w:
|
||||
w.execute("INSERT INTO notes(id,path,state,file_hash) VALUES('a','p','active','h')")
|
||||
w.execute("INSERT INTO notes(id,path,state,file_hash) VALUES('a','p2','active','h')")
|
||||
assert db.read().execute("SELECT count(*) FROM notes").fetchone()[0] == 0
|
||||
@@ -0,0 +1,66 @@
|
||||
import sqlite3
|
||||
|
||||
import pytest
|
||||
from hypothesis import given, settings
|
||||
from hypothesis import strategies as st
|
||||
|
||||
from mdmem.ftsquery import to_match, to_trigram_match
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def fts():
|
||||
db = sqlite3.connect(":memory:")
|
||||
db.execute("CREATE VIRTUAL TABLE t USING fts5(body, tokenize='unicode61 remove_diacritics 2')")
|
||||
db.execute("INSERT INTO t(body) VALUES ('llama-swap routet Modelle für Bifrost')")
|
||||
db.execute("CREATE VIRTUAL TABLE tri USING fts5(text, tokenize='trigram')")
|
||||
db.execute("INSERT INTO tri(text) VALUES ('bifrost.cloud.traunseenet.com')")
|
||||
return db
|
||||
|
||||
|
||||
def test_plain_terms_are_or_joined():
|
||||
assert to_match("llama swap") == '"llama" OR "swap"'
|
||||
|
||||
|
||||
def test_fts_operators_are_neutralised():
|
||||
q = to_match('NOT bifrost AND "x" OR (y) NEAR z*')
|
||||
assert q is not None
|
||||
for op in ("NOT ", " AND ", "NEAR", "(", ")", "*"):
|
||||
assert op not in q.replace(" OR ", " ")
|
||||
|
||||
|
||||
def test_empty_query_returns_none():
|
||||
assert to_match(" ... ") is None
|
||||
assert to_match("") is None
|
||||
|
||||
|
||||
def test_diacritics_query_matches_folded_index(fts):
|
||||
q = to_match("fur")
|
||||
rows = fts.execute("SELECT rowid FROM t WHERE t MATCH ?", (q,)).fetchall()
|
||||
assert rows # 'für' folded to 'fur' in the index; the query folds the same way
|
||||
|
||||
|
||||
def test_umlaut_query_matches(fts):
|
||||
q = to_match("für Modelle")
|
||||
assert fts.execute("SELECT rowid FROM t WHERE t MATCH ?", (q,)).fetchall()
|
||||
|
||||
|
||||
def test_trigram_matches_substring_of_hostname(fts):
|
||||
q = to_trigram_match("traunseenet")
|
||||
assert q is not None
|
||||
assert fts.execute("SELECT rowid FROM tri WHERE tri MATCH ?", (q,)).fetchall()
|
||||
|
||||
|
||||
def test_trigram_drops_short_tokens():
|
||||
assert to_trigram_match("a b") is None
|
||||
|
||||
|
||||
@settings(max_examples=400, deadline=None)
|
||||
@given(st.text())
|
||||
def test_sanitizer_never_produces_invalid_fts_syntax(s):
|
||||
db = sqlite3.connect(":memory:")
|
||||
db.execute("CREATE VIRTUAL TABLE t USING fts5(body)")
|
||||
db.execute("INSERT INTO t(body) VALUES ('hello world')")
|
||||
for q in (to_match(s), to_trigram_match(s)):
|
||||
if q is None:
|
||||
continue
|
||||
db.execute("SELECT rowid FROM t WHERE t MATCH ?", (q,)).fetchall()
|
||||
@@ -0,0 +1,158 @@
|
||||
import json
|
||||
|
||||
import pytest
|
||||
|
||||
from mdmem.db import Database
|
||||
from mdmem.index import Indexer
|
||||
from mdmem.notes import Note
|
||||
from mdmem.vaultio import Vault
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def idx(tmp_path):
|
||||
vault = Vault(tmp_path / "vault", tmp_path / ".history")
|
||||
vault.ensure_dirs()
|
||||
db = Database(tmp_path / "index.db", dim=8, trigram=True)
|
||||
db.open()
|
||||
yield Indexer(db, vault), vault, db
|
||||
db.close()
|
||||
|
||||
|
||||
def _note(vault, **kw):
|
||||
n = Note.new(**kw)
|
||||
vault.write(n, state=kw.pop("state", "inbox") if "state" in kw else n.state)
|
||||
return n
|
||||
|
||||
|
||||
def test_index_note_writes_all_rows(idx):
|
||||
indexer, vault, db = idx
|
||||
n = Note.new(title="llama-swap", body="routet Modelle", type="fact",
|
||||
summary="s", keywords=["router", "modell"], tags=["homelab"])
|
||||
vault.write(n)
|
||||
n = vault.read(vault.find(n.id))
|
||||
pending = indexer.index_note(n)
|
||||
|
||||
row = db.read().execute("SELECT * FROM notes WHERE id=?", (n.id,)).fetchone()
|
||||
assert row["state"] == "inbox" and row["title"] == "llama-swap"
|
||||
assert json.loads(row["tags"]) == ["homelab"]
|
||||
assert db.read().execute("SELECT count(*) FROM chunks WHERE note_id=?", (n.id,)).fetchone()[0] == 1
|
||||
assert db.read().execute("SELECT count(*) FROM chunks_fts").fetchone()[0] == 1
|
||||
assert db.read().execute("SELECT count(*) FROM chunks_tri").fetchone()[0] == 1
|
||||
assert len(pending) == 1 and pending[0].text.startswith("llama-swap")
|
||||
|
||||
|
||||
def test_fts_finds_keyword_and_body(idx):
|
||||
indexer, vault, db = idx
|
||||
n = Note.new(title="Titel", body="Bifrost fronts llama-swap", type="fact",
|
||||
keywords=["Modellrouter"])
|
||||
vault.write(n)
|
||||
indexer.index_note(vault.read(vault.find(n.id)))
|
||||
hits = db.read().execute(
|
||||
"SELECT rowid FROM chunks_fts WHERE chunks_fts MATCH ?", ('"modellrouter"',)
|
||||
).fetchall()
|
||||
assert hits
|
||||
assert db.read().execute(
|
||||
"SELECT rowid FROM chunks_fts WHERE chunks_fts MATCH ?", ('"fronts"',)
|
||||
).fetchall()
|
||||
|
||||
|
||||
def test_reindex_replaces_chunks_not_duplicates(idx):
|
||||
indexer, vault, db = idx
|
||||
n = Note.new(title="T", body="v1", type="fact")
|
||||
vault.write(n)
|
||||
n = vault.read(vault.find(n.id))
|
||||
indexer.index_note(n)
|
||||
n.body = "v2"
|
||||
vault.write(n)
|
||||
indexer.index_note(vault.read(vault.find(n.id)))
|
||||
assert db.read().execute("SELECT count(*) FROM chunks").fetchone()[0] == 1
|
||||
assert db.read().execute("SELECT count(*) FROM chunks_fts").fetchone()[0] == 1
|
||||
assert db.read().execute("SELECT body FROM chunks").fetchone()[0] == "v2"
|
||||
|
||||
|
||||
def test_links_are_extracted_and_resolved(idx):
|
||||
indexer, vault, db = idx
|
||||
target = Note.new(title="evo-x2", body="the box", type="entity")
|
||||
vault.write(target)
|
||||
indexer.index_note(vault.read(vault.find(target.id)))
|
||||
|
||||
src = Note.new(title="llama-swap", body="runs on [[evo-x2]] and [[nowhere]]",
|
||||
type="fact", relations={"part_of": ["[[evo-x2]]"]})
|
||||
vault.write(src)
|
||||
indexer.index_note(vault.read(vault.find(src.id)))
|
||||
|
||||
links = {(r["dst"], r["rel"]) for r in db.read().execute("SELECT dst, rel FROM links WHERE src=?", (src.id,))}
|
||||
assert (target.id, "links") in links
|
||||
assert (target.id, "part_of") in links
|
||||
assert ("unresolved:nowhere", "links") in links
|
||||
|
||||
|
||||
def test_forward_reference_resolves_when_target_arrives(idx):
|
||||
indexer, vault, db = idx
|
||||
src = Note.new(title="A", body="see [[Later Note]]", type="note")
|
||||
vault.write(src)
|
||||
indexer.index_note(vault.read(vault.find(src.id)))
|
||||
assert db.read().execute("SELECT dst FROM links WHERE src=?", (src.id,)).fetchone()[0].startswith("unresolved:")
|
||||
|
||||
later = Note.new(title="Later Note", body="here", type="note")
|
||||
vault.write(later)
|
||||
indexer.index_note(vault.read(vault.find(later.id)))
|
||||
assert db.read().execute("SELECT dst FROM links WHERE src=?", (src.id,)).fetchone()[0] == later.id
|
||||
|
||||
|
||||
def test_scan_is_incremental_and_removes_vanished(idx):
|
||||
indexer, vault, db = idx
|
||||
a = Note.new(title="A", body="a", type="fact")
|
||||
b = Note.new(title="B", body="b", type="fact")
|
||||
vault.write(a)
|
||||
vault.write(b)
|
||||
s1 = indexer.scan()
|
||||
assert s1.indexed == 2 and s1.removed == 0
|
||||
|
||||
s2 = indexer.scan()
|
||||
assert s2.indexed == 0 and s2.unchanged == 2
|
||||
|
||||
vault.find(a.id).unlink()
|
||||
s3 = indexer.scan()
|
||||
assert s3.removed == 1
|
||||
assert db.read().execute("SELECT count(*) FROM notes").fetchone()[0] == 1
|
||||
assert db.read().execute("SELECT count(*) FROM chunks").fetchone()[0] == 1
|
||||
|
||||
|
||||
def test_handwritten_note_without_id_gets_one(idx):
|
||||
indexer, vault, db = idx
|
||||
p = vault.root / "active" / "handwritten.md"
|
||||
p.write_text("---\ntitle: Von Hand\ntype: fact\n---\nInhalt\n", encoding="utf-8")
|
||||
stats = indexer.scan()
|
||||
assert stats.indexed == 1 and not stats.errors
|
||||
row = db.read().execute("SELECT id, title FROM notes").fetchone()
|
||||
assert len(row["id"]) == 26 and row["title"] == "Von Hand"
|
||||
assert not p.exists() # renamed to slug--ULID.md
|
||||
|
||||
|
||||
def test_broken_yaml_is_skipped_not_fatal(idx):
|
||||
indexer, vault, db = idx
|
||||
ok = Note.new(title="Fine", body="x", type="fact")
|
||||
vault.write(ok)
|
||||
(vault.root / "active" / "broken--01JZZZZZZZZZZZZZZZZZZZZZZZ.md").write_text(
|
||||
"---\nid: 01JZZZZZZZZZZZZZZZZZZZZZZZ\ntype: nonsense-type\n---\nbody\n", encoding="utf-8"
|
||||
)
|
||||
stats = indexer.scan()
|
||||
assert stats.indexed == 1 and len(stats.errors) == 1
|
||||
assert db.read().execute("SELECT count(*) FROM notes").fetchone()[0] == 1
|
||||
|
||||
|
||||
def test_unchanged_embed_text_keeps_its_vector(idx):
|
||||
indexer, vault, db = idx
|
||||
n = Note.new(title="T", body="body", type="fact")
|
||||
vault.write(n)
|
||||
n = vault.read(vault.find(n.id))
|
||||
pending = indexer.index_note(n)
|
||||
indexer.store_vectors([(pending[0].chunk_id, b"\x00" * 32)])
|
||||
|
||||
n.set("tags", ["neu"]) # tags are not part of the embed text
|
||||
vault.write(n)
|
||||
again = indexer.index_note(vault.read(vault.find(n.id)))
|
||||
assert again == []
|
||||
assert db.read().execute("SELECT count(*) FROM chunks_vec").fetchone()[0] == 1
|
||||
assert db.read().execute("SELECT vec_pending FROM chunks").fetchone()[0] == 0
|
||||
Reference in new issue
Block a user