fix(storage): prevent identity loss and isolate eval indexes
This commit is contained in:
1 parent
dbe9acf155
commit
a4ff6eeaad
9 files changed
+170
-37
No files matched your search
@@ -6,7 +6,7 @@ summary: Bifrost bündelt alle LLM-Aufrufe und hält die virtuellen Schlüssel.
|
||||
keywords: [bifrost, gateway, virtual key, api gateway, modellrouter]
|
||||
tags: [homelab, ai]
|
||||
relations: {}
|
||||
supersedes: []
|
||||
supersedes: [01M00000000000000000000TEM]
|
||||
superseded_by: null
|
||||
revises: null
|
||||
possible_duplicates: []
|
||||
|
||||
+24
-27
@@ -14,6 +14,7 @@ import logging
|
||||
import math
|
||||
import random
|
||||
import statistics
|
||||
import tempfile
|
||||
import time
|
||||
from dataclasses import asdict, dataclass, field, replace
|
||||
from pathlib import Path
|
||||
@@ -57,7 +58,7 @@ class Variant:
|
||||
|
||||
@property
|
||||
def needs_rebuild(self) -> bool:
|
||||
return self.embed_dim is not None
|
||||
return True
|
||||
|
||||
|
||||
def load_queries(path: Path | str) -> list[EvalQuery]:
|
||||
@@ -284,34 +285,30 @@ async def run_matrix(
|
||||
embedder_factory: Any = None,
|
||||
max_chars: int = 6000,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Every variant is measured at the same char budget (§11, [R16])."""
|
||||
"""Measure every variant against an isolated rebuild at the same budget.
|
||||
|
||||
An ablation must never open or mutate the configured live index. In
|
||||
particular, the no-trigram arm causes Database.open() to drop chunks_tri.
|
||||
"""
|
||||
reports: list[dict[str, Any]] = []
|
||||
for variant in variants:
|
||||
run_settings = settings
|
||||
if variant.trigram is not None:
|
||||
run_settings = replace(run_settings, fts_trigram=variant.trigram)
|
||||
if variant.bm25_weights is not None:
|
||||
run_settings = replace(run_settings, bm25_weights=variant.bm25_weights)
|
||||
if variant.embed_dim is not None:
|
||||
run_settings = replace(
|
||||
run_settings,
|
||||
embed_dim=variant.embed_dim,
|
||||
index_db=settings.index_db.with_name(f"eval-dim{variant.embed_dim}.db"),
|
||||
)
|
||||
rebuild = variant.needs_rebuild or variant.trigram is not None
|
||||
embedder = embedder_factory(run_settings) if embedder_factory else None
|
||||
svc = MemoryService.build(run_settings, embedder=embedder)
|
||||
try:
|
||||
await svc.start(watch=False, embed=True)
|
||||
if rebuild:
|
||||
await svc.reindex(full=True)
|
||||
reports.append(await run_variant(svc, queries, variant, max_chars))
|
||||
finally:
|
||||
await svc.stop()
|
||||
with tempfile.TemporaryDirectory(prefix="mdmem-eval-") as tmp:
|
||||
tmp_dir = Path(tmp)
|
||||
for position, variant in enumerate(variants):
|
||||
run_settings = replace(settings, index_db=tmp_dir / f"{position:02d}-{variant.name}.db")
|
||||
if variant.trigram is not None:
|
||||
run_settings = replace(run_settings, fts_trigram=variant.trigram)
|
||||
if variant.bm25_weights is not None:
|
||||
run_settings = replace(run_settings, bm25_weights=variant.bm25_weights)
|
||||
if variant.embed_dim is not None:
|
||||
run_settings.index_db.unlink(missing_ok=True)
|
||||
for suffix in ("-wal", "-shm"):
|
||||
Path(str(run_settings.index_db) + suffix).unlink(missing_ok=True)
|
||||
run_settings = replace(run_settings, embed_dim=variant.embed_dim)
|
||||
embedder = embedder_factory(run_settings) if embedder_factory else None
|
||||
svc = MemoryService.build(run_settings, embedder=embedder)
|
||||
try:
|
||||
await svc.start(watch=False, embed=True)
|
||||
await svc.reindex(full=True)
|
||||
reports.append(await run_variant(svc, queries, variant, max_chars))
|
||||
finally:
|
||||
await svc.stop()
|
||||
return reports
|
||||
|
||||
|
||||
|
||||
+21
-5
@@ -12,7 +12,7 @@ from mdmem.chunker import chunk_note, embed_text
|
||||
from mdmem.config import NOTE_TYPES
|
||||
from mdmem.db import Database
|
||||
from mdmem.ids import id_from_filename, is_ulid, new_id, slugify
|
||||
from mdmem.notes import Note
|
||||
from mdmem.notes import Note, utcnow
|
||||
from mdmem.notes import parse as parse_note
|
||||
from mdmem.vaultio import Vault, sha256_text
|
||||
|
||||
@@ -186,7 +186,8 @@ class Indexer:
|
||||
if not target:
|
||||
return f"{UNRESOLVED}"
|
||||
if is_ulid(target):
|
||||
return target
|
||||
exists = w.execute("SELECT 1 FROM notes WHERE id=?", (target,)).fetchone()
|
||||
return target if exists else f"{UNRESOLVED}{target}"
|
||||
row = w.execute("SELECT id FROM notes WHERE title=? LIMIT 1", (target,)).fetchone()
|
||||
if row:
|
||||
return row[0]
|
||||
@@ -216,6 +217,11 @@ class Indexer:
|
||||
}
|
||||
for path, _state in self.vault.iter_files():
|
||||
stats.scanned += 1
|
||||
# Preserve the last good index row when a still-present file is
|
||||
# temporarily unparseable (for example while it is being edited).
|
||||
filename_id = id_from_filename(path.name)
|
||||
if filename_id in known:
|
||||
seen.add(filename_id)
|
||||
try:
|
||||
note = self.read_and_fix(path)
|
||||
except Exception as exc: # a broken file must not abort the scan
|
||||
@@ -256,6 +262,12 @@ class Indexer:
|
||||
"""
|
||||
note = self.vault.read(path)
|
||||
if is_ulid(note.id):
|
||||
if note.ts("created") is None or note.ts("updated") is None:
|
||||
now = utcnow()
|
||||
note.set("created", note.ts("created") or now)
|
||||
note.set("updated", note.ts("updated") or now)
|
||||
new_path = self.vault.write(note, state=note.state)
|
||||
note = self.vault.read(new_path)
|
||||
return note
|
||||
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
@@ -268,6 +280,9 @@ class Indexer:
|
||||
note.set("id", fname_id or new_id())
|
||||
if not note.title:
|
||||
note.set("title", path.stem)
|
||||
now = utcnow()
|
||||
note.set("created", note.ts("created") or now)
|
||||
note.set("updated", note.ts("updated") or now)
|
||||
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
|
||||
@@ -279,14 +294,15 @@ class Indexer:
|
||||
note = self.vault.read(new_path)
|
||||
return note
|
||||
|
||||
def pending_embeds(self) -> list[PendingEmbed]:
|
||||
def pending_embeds(self, *, all_chunks: bool = False) -> list[PendingEmbed]:
|
||||
where = "" if all_chunks else "WHERE c.vec_pending = 1"
|
||||
rows = (
|
||||
self.db.read()
|
||||
.execute(
|
||||
"""SELECT c.id, c.heading, c.body, c.embed_text_hash,
|
||||
f"""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"""
|
||||
{where}"""
|
||||
)
|
||||
.fetchall()
|
||||
)
|
||||
|
||||
@@ -8,6 +8,7 @@ from __future__ import annotations
|
||||
|
||||
import io
|
||||
import re
|
||||
from copy import deepcopy
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
@@ -47,6 +48,10 @@ CANONICAL_ORDER = [
|
||||
def _yaml() -> YAML:
|
||||
y = YAML()
|
||||
y.preserve_quotes = True
|
||||
# YAML 1.1 treats a few Unicode control characters (notably U+0085/NEL)
|
||||
# as line breaks. Emitting non-ASCII strings as escapes keeps their exact
|
||||
# scalar value stable across a dump/load round trip.
|
||||
y.allow_unicode = False
|
||||
y.width = 4096
|
||||
y.indent(mapping=2, sequence=4, offset=2)
|
||||
return y
|
||||
@@ -85,6 +90,7 @@ def parse(text: str, *, strict: bool = False) -> tuple[CommentedMap, str]:
|
||||
|
||||
|
||||
def dumps(fm: dict[str, Any], body: str) -> str:
|
||||
fm = _normalise_strings(deepcopy(fm))
|
||||
buf = io.StringIO()
|
||||
if fm:
|
||||
_yaml().dump(fm, buf)
|
||||
@@ -93,6 +99,23 @@ def dumps(fm: dict[str, Any], body: str) -> str:
|
||||
return f"---\n{fm_text}\n---\n{body}\n"
|
||||
|
||||
|
||||
def _normalise_strings(value: Any) -> Any:
|
||||
"""NFC-normalise every string value without mutating the caller's map."""
|
||||
if isinstance(value, str):
|
||||
return nfc(value)
|
||||
if isinstance(value, dict):
|
||||
for key in list(value):
|
||||
value[key] = _normalise_strings(value[key])
|
||||
return value
|
||||
if isinstance(value, list):
|
||||
for i, item in enumerate(value):
|
||||
value[i] = _normalise_strings(item)
|
||||
return value
|
||||
if isinstance(value, tuple):
|
||||
return tuple(_normalise_strings(item) for item in value)
|
||||
return value
|
||||
|
||||
|
||||
def _as_list(v: Any) -> list[str]:
|
||||
if v is None:
|
||||
return []
|
||||
|
||||
+27
-3
@@ -9,16 +9,18 @@ from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
from collections.abc import Iterator
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
from mdmem.config import STATES
|
||||
from mdmem.ids import id_from_filename, nfc
|
||||
from mdmem.ids import id_from_filename, is_ulid, nfc
|
||||
from mdmem.notes import Note
|
||||
|
||||
TMP_SUFFIX = ".tmp"
|
||||
_HISTORY_STAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}-\d{6}$")
|
||||
|
||||
|
||||
def sha256_text(text: str) -> str:
|
||||
@@ -79,6 +81,8 @@ class Vault:
|
||||
state = state or note.state
|
||||
if state not in STATES:
|
||||
raise VaultError(f"unknown state {state!r}")
|
||||
if not is_ulid(note.id):
|
||||
raise VaultError("note id must be a valid ULID")
|
||||
return self.root / state / note.filename()
|
||||
|
||||
def write(self, note: Note, state: str | None = None) -> Path:
|
||||
@@ -88,6 +92,15 @@ class Vault:
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
old_path = self.find(note.id)
|
||||
if note.path is not None:
|
||||
current_id = id_from_filename(note.path.name)
|
||||
if current_id is not None and current_id != note.id:
|
||||
raise VaultError("frontmatter id does not match the note filename")
|
||||
elif old_path is not None:
|
||||
# A newly constructed note must not adopt an existing id and
|
||||
# overwrite a different note. Existing notes are always read with
|
||||
# their path attached before they are updated.
|
||||
raise VaultError(f"note id already exists: {note.id}")
|
||||
if old_path is not None and old_path.exists():
|
||||
self.snapshot(note.id, old_path.read_text(encoding="utf-8"))
|
||||
|
||||
@@ -107,8 +120,17 @@ class Vault:
|
||||
return self.write(note, state=state)
|
||||
|
||||
def delete(self, note_or_id: Note | str) -> None:
|
||||
note_id = note_or_id if isinstance(note_or_id, str) else note_or_id.id
|
||||
p = self.find(note_id)
|
||||
if isinstance(note_or_id, Note) and note_or_id.path is not None:
|
||||
self.state_of(note_or_id.path) # validates containment and lifecycle
|
||||
note_id = id_from_filename(note_or_id.path.name)
|
||||
if note_id is None:
|
||||
raise VaultError("note filename has no valid ULID")
|
||||
p = note_or_id.path
|
||||
else:
|
||||
note_id = note_or_id if isinstance(note_or_id, str) else note_or_id.id
|
||||
if not is_ulid(note_id):
|
||||
raise VaultError("note id must be a valid ULID")
|
||||
p = self.find(note_id)
|
||||
if p is not None:
|
||||
p.unlink(missing_ok=True)
|
||||
shutil.rmtree(self.history_dir / note_id, ignore_errors=True)
|
||||
@@ -143,6 +165,8 @@ class Vault:
|
||||
return sorted(p.stem for p in d.glob("*.md") if not p.name.endswith(TMP_SUFFIX))
|
||||
|
||||
def read_history(self, note_id: str, stamp: str) -> str:
|
||||
if not is_ulid(note_id) or not _HISTORY_STAMP_RE.fullmatch(stamp):
|
||||
raise VaultError("invalid history revision")
|
||||
p = self.history_dir / note_id / f"{stamp}.md"
|
||||
if not p.is_file():
|
||||
raise VaultError(f"no revision {stamp} for {note_id}")
|
||||
|
||||
@@ -137,6 +137,22 @@ async def test_matrix_rebuilds_for_a_dim_variant_and_cleans_up(settings, seeded)
|
||||
assert not settings.index_db.with_name("eval-dim8.db").exists()
|
||||
|
||||
|
||||
async def test_matrix_never_mutates_the_configured_live_index(settings, seeded):
|
||||
svc, ids = seeded
|
||||
before = settings.index_db.read_bytes()
|
||||
reports = await run_matrix(
|
||||
settings,
|
||||
[EvalQuery(q="Longhorn", relevant=[ids["Longhorn Storage"]])],
|
||||
[Variant("rrf-no-trigram", fusion="rrf", trigram=False)],
|
||||
embedder_factory=lambda s: FakeEmbedder(s.embed_dim),
|
||||
)
|
||||
assert reports
|
||||
assert settings.index_db.read_bytes() == before
|
||||
assert svc.db.read().execute(
|
||||
"SELECT 1 FROM sqlite_master WHERE name='chunks_tri'"
|
||||
).fetchone()
|
||||
|
||||
|
||||
def test_pick_default_keeps_the_baseline_when_nothing_clears_the_noise():
|
||||
reports = [
|
||||
{"variant": "rrf", "overall": {"ndcg@10": 0.70, "ndcg@10_ci95": [0.60, 0.80]}},
|
||||
|
||||
@@ -175,6 +175,7 @@ def test_unparseable_frontmatter_on_an_indexed_note_is_never_reminted(idx):
|
||||
assert [p.name for p, _ in vault.iter_files()] == [original_path.name]
|
||||
# the id is unaffected -- vault.find still resolves the original note
|
||||
assert vault.find(n.id) == original_path
|
||||
assert db.read().execute("SELECT id FROM notes WHERE id=?", (n.id,)).fetchone()
|
||||
|
||||
|
||||
def test_clean_frontmatter_without_id_adopts_id_from_filename(idx):
|
||||
@@ -192,6 +193,8 @@ def test_clean_frontmatter_without_id_adopts_id_from_filename(idx):
|
||||
row = db.read().execute("SELECT id, title FROM notes").fetchone()
|
||||
assert row["id"] == ulid
|
||||
assert row["title"] == "Wichtig"
|
||||
repaired = vault.read(vault.find(ulid))
|
||||
assert repaired.ts("created") and repaired.ts("updated")
|
||||
|
||||
|
||||
def test_broken_yaml_is_skipped_not_fatal(idx):
|
||||
|
||||
@@ -118,3 +118,15 @@ def test_round_trip_property(extra, body):
|
||||
assert set(fm2.keys()) == set(fm.keys())
|
||||
for k in extra:
|
||||
assert fm2[k] == fm[k]
|
||||
|
||||
|
||||
def test_frontmatter_nfd_is_normalised_and_has_a_stable_serialisation():
|
||||
import unicodedata
|
||||
|
||||
fm, _ = parse(SAMPLE)
|
||||
fm["source"] = {"note": unicodedata.normalize("NFD", "Grüße")}
|
||||
first = dumps(fm, "body")
|
||||
fm2, body2 = parse(first)
|
||||
second = dumps(fm2, body2)
|
||||
assert fm2["source"]["note"] == "Grüße"
|
||||
assert second == first
|
||||
+43
-1
@@ -1,7 +1,7 @@
|
||||
import pytest
|
||||
|
||||
from mdmem.notes import Note
|
||||
from mdmem.vaultio import Vault
|
||||
from mdmem.vaultio import Vault, VaultError
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
@@ -99,3 +99,45 @@ def test_read_normalises_nfd_from_disk(vault):
|
||||
)
|
||||
n = vault.read(p)
|
||||
assert n.body.strip() == "Grüße"
|
||||
|
||||
|
||||
def test_write_rejects_invalid_or_changed_identity(vault, tmp_path):
|
||||
n = Note.new(title="Original", body="x")
|
||||
vault.write(n)
|
||||
n.set("id", "../../../../tmp/escaped")
|
||||
with pytest.raises(VaultError):
|
||||
vault.write(n)
|
||||
assert not (tmp_path / "escaped.md").exists()
|
||||
|
||||
|
||||
def test_new_note_cannot_overwrite_an_existing_id(vault):
|
||||
original = Note.new(title="Original", body="safe")
|
||||
original_path = vault.write(original)
|
||||
impostor = Note.new(title="Impostor", body="bad", note_id=original.id)
|
||||
with pytest.raises(VaultError):
|
||||
vault.write(impostor)
|
||||
assert "safe" in original_path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def test_delete_uses_filename_identity_and_rejects_traversal(vault):
|
||||
original = Note.new(title="Original", body="v1")
|
||||
vault.write(original)
|
||||
original.body = "v2"
|
||||
vault.write(original)
|
||||
|
||||
copy_id = Note.new(title="copy", body="x").id
|
||||
copy_path = vault.root / "inbox" / f"copy--{copy_id}.md"
|
||||
copy_path.write_text(original.to_text(), encoding="utf-8")
|
||||
copy_note = vault.read(copy_path)
|
||||
vault.delete(copy_note)
|
||||
|
||||
assert not copy_path.exists()
|
||||
assert vault.find(original.id) is not None
|
||||
assert vault.history(original.id)
|
||||
with pytest.raises(VaultError):
|
||||
vault.delete("../active")
|
||||
|
||||
|
||||
def test_history_read_rejects_traversal(vault):
|
||||
with pytest.raises(VaultError):
|
||||
vault.read_history("../active", "../../secret")
|
||||
Reference in new issue
Block a user