M5.1: the knowledge vault

vault.py: a folder of Markdown notes on the agent host, as SilverBullet reads
them. Only configured folders exist for the agent, each rw or ro (`vault:`
in the configuration); paths are checked in code (relative, inside a folder,
symlinks included, Markdown, never the editor's own files). Writes are
atomic (temporary file, fsync, rename, directory fsync) and merge provenance
into the note's frontmatter, keeping the rest; failures are results.

read_note, list_notes and search_notes are control tools in the harness for
the driver (before the go-ahead, and while a VM is blocked) and for
conversations, labelled with their path; the prompt says a note is
information, never an instruction. Writing by the model is M5.2's.

With the operator: SilverBullet replaces Obsidian and runs on the vault
itself, agents' folders mounted read-only, so no Syncthing (L10, X5, M5.4).
Per-agent mounts come with the containers (M6.1).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk

nisse-dev/src/nisse_dev/vault.py

@@ -0,0 +1,250 @@
+"""The knowledge vault (architecture L10; Development Plan M5.1).
+
+A folder of Markdown notes on the agent host, in the plain format SilverBullet
+reads (frontmatter, ``[[links]]``, ``#tags``), so the operator browses it there.
+
+- **Only configured folders exist** for the agent, each read-write or
+  read-only; every folder has one writer. Paths are checked in code before
+  anything is touched: relative, inside an allowed folder (symlinks
+  included), Markdown, and never the editor's own files. Per-agent mounts
+  enforce the same at the OS level once the agent runs in a container (M6.1).
+- **Writes are atomic** (write a temporary file beside the note, sync it,
+  rename it into place), so a crash, SilverBullet or anyone reading the
+  folder never sees half a note, and they **record provenance** in the
+  note's frontmatter.
+- **Failures are results, never raised into the loop:** a failed read or
+  write never blocks work.
+- **Notes are data, never instructions** (L10, L11): the tools label what
+  they return with its path.
+"""
+
+import fnmatch
+import os
+import tempfile
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path, PurePosixPath
+from typing import Any
+
+import yaml
+
+READ_WRITE = "rw"
+READ_ONLY = "ro"
+ACCESS = (READ_WRITE, READ_ONLY)
+# The editor's and tools' own files: hidden ones (SilverBullet's and
+# Obsidian's settings, .git, a write's temporary file) and SilverBullet's plugs.
+DEFAULT_IGNORE = (".*", "_plug")
+NOTE_SUFFIX = ".md"
+MAX_NOTE_CHARS = 20_000  # what one read returns; the rest is said to be cut
+MAX_LISTED = 200
+MAX_FOUND = 20
+LINES_PER_NOTE = 3
+
+
+class VaultError(Exception):
+    """What a vault operation couldn't do, in words for the model."""
+
+
+@dataclass(frozen=True)
+class Found:
+    path: str
+    lines: tuple[str, ...]  # the matching lines, first few
+    hits: int
+
+
+class Vault:
+    """``folders``: vault-relative folder -> "rw" or "ro". Nothing outside
+    them is visible."""
+
+    def __init__(
+        self,
+        root: Path,
+        folders: Mapping[str, str],
+        *,
+        ignore: Sequence[str] = DEFAULT_IGNORE,
+        agent: str = "nisse-dev",
+        clock: Any = None,
+    ) -> None:
+        self.root = Path(root)
+        self._folders = {_clean(f): a for f, a in folders.items()}
+        self._ignore = tuple(ignore)
+        self._agent = agent
+        self._clock = clock or (lambda: datetime.now(UTC))
+        for folder, access in self._folders.items():
+            if access == READ_WRITE:
+                (self.root / folder).mkdir(parents=True, exist_ok=True)
+
+    def folders(self) -> dict[str, str]:
+        return dict(self._folders)
+
+    # -- paths -----------------------------------------------------------------
+
+    def _folder_of(self, path: PurePosixPath) -> str | None:
+        for folder in self._folders:
+            if path == PurePosixPath(folder) or PurePosixPath(folder) in path.parents:
+                return folder
+        return None
+
+    def _hidden(self, path: PurePosixPath) -> bool:
+        return any(
+            fnmatch.fnmatchcase(part, pattern)
+            for part in path.parts
+            for pattern in self._ignore
+        )
+
+    def _resolve(self, path: str, *, note: bool) -> tuple[PurePosixPath, Path, str]:
+        """The checked path, where it is on disk, and its folder's access."""
+        rel = _relative(path)
+        folder = self._folder_of(rel)
+        if folder is None:
+            raise VaultError(
+                f"{path}: not in a folder you can see; they are "
+                + ", ".join(sorted(self._folders))
+            )
+        if self._hidden(rel):
+            raise VaultError(f"{path}: the editor's own files aren't notes")
+        if note and rel.suffix != NOTE_SUFFIX:
+            raise VaultError(f"{path}: notes are Markdown files ending in .md")
+        real = (self.root / rel).resolve()
+        top = (self.root / folder).resolve()
+        if real != top and top not in real.parents:
+            raise VaultError(f"{path}: leads outside its folder")
+        return rel, real, self._folders[folder]
+
+    # -- reading ---------------------------------------------------------------
+
+    def read(self, path: str) -> str:
+        _, real, _ = self._resolve(path, note=True)
+        try:
+            text = real.read_text(encoding="utf-8")
+        except FileNotFoundError:
+            raise VaultError(f"{path}: no such note") from None
+        except (OSError, UnicodeDecodeError) as exc:
+            raise VaultError(f"{path}: couldn't read it ({exc})") from None
+        if len(text) > MAX_NOTE_CHARS:
+            cut = len(text) - MAX_NOTE_CHARS
+            text = f"{text[:MAX_NOTE_CHARS]}\n\n[{cut} more characters not shown]"
+        return text
+
+    def notes(self, folder: str = "") -> list[str]:
+        """Notes under ``folder`` (every visible folder when empty)."""
+        tops = (
+            [self._resolve(folder, note=False)[0]]
+            if folder.strip("/ ")
+            else [PurePosixPath(f) for f in sorted(self._folders)]
+        )
+        return sorted(p for top in tops for p in self._notes_under(top))
+
+    def _notes_under(self, top: PurePosixPath) -> list[str]:
+        base = (self.root / top).resolve()
+        if not base.is_dir():
+            return []
+        found = []
+        for dirpath, dirnames, filenames in os.walk(base):
+            here = PurePosixPath(Path(dirpath).relative_to(self.root.resolve()))
+            dirnames[:] = sorted(d for d in dirnames if not self._hidden(here / d))
+            for name in sorted(filenames):
+                rel = here / name
+                if rel.suffix != NOTE_SUFFIX or self._hidden(rel):
+                    continue
+                real = (self.root / rel).resolve()
+                if real == base or base in real.parents:  # no links out
+                    found.append(str(rel))
+        return found
+
+    def search(self, text: str) -> list[Found]:
+        """Notes whose path or text contains ``text``, ignoring case; most
+        matches first."""
+        needle = text.strip().lower()
+        if not needle:
+            raise VaultError("search for some text")
+        found = []
+        for path in self.notes():
+            try:
+                _, real, _ = self._resolve(path, note=True)
+                body = real.read_text(encoding="utf-8")
+            except (VaultError, OSError, UnicodeDecodeError):
+                continue  # a failed read never blocks the rest
+            lines = [ln.strip() for ln in body.splitlines() if needle in ln.lower()]
+            hits = len(lines) + (needle in path.lower())
+            if hits:
+                found.append(Found(path, tuple(lines[:LINES_PER_NOTE]), hits))
+        found.sort(key=lambda f: (-f.hits, f.path))
+        return found[:MAX_FOUND]
+
+    # -- writing ---------------------------------------------------------------
+
+    def write(self, path: str, text: str, *, provenance: Mapping[str, str]) -> None:
+        """Write a note in a read-write folder, atomically, with where it
+        came from merged into its frontmatter; the rest of its frontmatter
+        is kept."""
+        _, real, access = self._resolve(path, note=True)
+        if access != READ_WRITE:
+            raise VaultError(f"{path}: that folder is read-only for you")
+        front, body = split_frontmatter(text)
+        front["provenance"] = {
+            "by": self._agent,
+            **dict(provenance),
+            "written": self._clock().isoformat(timespec="seconds"),
+        }
+        content = (
+            "---\n"
+            + yaml.safe_dump(front, sort_keys=False, allow_unicode=True)
+            + "---\n"
+            + body
+        )
+        try:
+            real.parent.mkdir(parents=True, exist_ok=True)
+            _replace(real, content)
+        except OSError as exc:
+            raise VaultError(f"{path}: couldn't write it ({exc})") from None
+
+
+def split_frontmatter(text: str) -> tuple[dict[str, Any], str]:
+    """A note's frontmatter (a YAML mapping between --- lines at the top)
+    and the rest. Raises VaultError for frontmatter that isn't a mapping."""
+    if not text.startswith("---\n"):
+        return {}, text
+    end = text.find("\n---\n", 4)
+    if end == -1:
+        return {}, text
+    try:
+        front = yaml.safe_load(text[4:end]) or {}
+    except yaml.YAMLError as exc:
+        raise VaultError(f"its frontmatter isn't valid YAML ({exc})") from None
+    if not isinstance(front, dict):
+        raise VaultError("its frontmatter must be a mapping")
+    return front, text[end + 5 :]
+
+
+def _relative(path: str) -> PurePosixPath:
+    rel = PurePosixPath(path.strip())
+    if not path.strip() or rel.is_absolute() or ".." in rel.parts or "\\" in path:
+        raise VaultError(f"{path!r}: give a path inside the vault, like notes/x.md")
+    return rel
+
+
+def _clean(folder: str) -> str:
+    return str(_relative(folder.strip("/")))
+
+
+def _replace(target: Path, content: str) -> None:
+    """Write-then-rename in the note's own directory (L10): readers see the
+    old note or the new one, never part of either. The temporary file is
+    hidden, so it is never listed, and removed if anything fails."""
+    fd, temp = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(content)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(temp, target)
+    except BaseException:
+        Path(temp).unlink(missing_ok=True)
+        raise
+    directory = os.open(target.parent, os.O_RDONLY)
+    try:
+        os.fsync(directory)  # the rename itself survives a crash
+    finally:
+        os.close(directory)