Skip to content
This repository was archived by the owner on Aug 20, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 28 additions & 7 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -131,18 +131,39 @@ package.

## Open-core extension contract

The admin package subclasses/imports these -- rename only in coordination with it:
The private package extends this one through the names below. **Rename any of
them only in coordination with it**, because nothing in this repo's tests or CI
will notice if you break one.

Overridden -- the hooks that exist to be replaced:

- `server.create_fastmcp_app(*, auth=None, token_verifier=None, extra_instructions=None)`
-- single source of truth for FastMCP construction.
- `tools.handler.VaultToolHandler._get_provider` -- the hook admin overrides to
resolve a per-workspace vault from the authenticated subject.
- `tools.handler.VaultToolHandler._list_vaults` -- the hook admin overrides to
list a workspace's vaults.
- `tools.handler.VaultToolHandler._get_provider` -- resolve a per-tenant vault
from the authenticated subject.
- `tools.handler.VaultToolHandler._list_vaults` -- list a tenant's vaults.
- `tools.handler.VaultToolHandler._track_usage` -- usage-tracking hook (no-op here).
- `tools._common._current_sub` -- contextvar carrying the authenticated subject.
- `usage.track_event` -- no-op stub here; the real tracker lives in admin.
- `config.KnapConfig` and `server.SERVER_VERSION`.

Imported -- ordinary use of this package's API, listed because it is easy to
mistake some of it for internal:

- `config.KnapConfig`.
- `error_handling`: `KnapError`, `ValidationError`, `sanitize`.
- `providers`: `VaultProvider` and the exception hierarchy.
- `providers.protocol`: the value dataclasses.
- `providers.filesystem.provider.FilesystemVaultProvider`, composed rather than
subclassed.
- **`providers.filesystem.markdown` and `providers.filesystem.paths`.** These
read as internal -- `providers/filesystem/` is described above as the only
place that knows about the filesystem, and `paths.py` is a security boundary
with its own regression file -- and a downstream backend is built on them
anyway. Treat their signatures as public.

Reserved, and not used by anything today: `usage.track_event` is a stub that
nothing in either package calls, and `server.SERVER_VERSION` is used here but
imported by nobody. Keep or drop them on their own merits; "the private package
depends on it" is not currently a reason.

## Conventions

Expand Down
54 changes: 48 additions & 6 deletions knap_mcp/providers/filesystem/frontmatter.py
Original file line number Diff line number Diff line change
Expand Up @@ -141,18 +141,38 @@ def _render(key: str, value: Any) -> Optional[List[str]]:
if isinstance(value, (int, float)):
return [f"{key}: {value}"]
if isinstance(value, str):
return [f"{key}: {_scalar(value)}"]
rendered = _scalar(value)
return None if rendered is None else [f"{key}: {rendered}"]
if isinstance(value, list):
if not value:
return [f"{key}: []"]
if not all(isinstance(item, (str, int, float, bool)) for item in value):
return None
return [f"{key}:", *[f" - {_scalar(str(item))}" for item in value]]
items = [_scalar(str(item)) for item in value]
if any(item is None for item in items):
return None
return [f"{key}:", *[f" - {item}" for item in items]]
return None


def _scalar(value: str) -> str:
"""Quote a string only when YAML would otherwise read it as something else."""
def _scalar(value: str) -> Optional[str]:
"""Write a string as a YAML scalar, or None if we cannot do it safely.

The list below is the cheap answer and ``_reads_back`` is the deciding one.
Keeping both is deliberate: the list settles the overwhelming majority
without a parse and documents the cases worth knowing about, and the round
trip catches whatever the list forgot. It forgot ``#`` for a long time -- a
hash after a space opens a comment, so ``Budget #2026 review`` was written
unquoted and read back as ``Budget``, with the rest gone from the file.

Quoting is not automatically the safe answer either, which is why the quoted
form is checked too. Escaping covers backslash, quote and newline, so a
carriage return survives into the double-quoted scalar and YAML folds it to
a space on the way back. When neither form round-trips we return None and
the caller re-dumps the block with PyYAML: a formatting diff the customer
can see, which is the trade this module already makes for nested values and
is strictly better than writing something that reads back wrong.
"""
if value == "":
return '""'
needs_quotes = (
Expand All @@ -161,13 +181,35 @@ def _scalar(value: str) -> str:
or ": " in value
or value.endswith(":")
or "\n" in value
or " #" in value
or "\t#" in value
or value.lower() in ("true", "false", "null", "yes", "no", "on", "off", "~")
or _looks_numeric(value)
)
if not needs_quotes:
if not needs_quotes and _reads_back(value, value):
return value
escaped = value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
return f'"{escaped}"'
quoted = f'"{escaped}"'
return quoted if _reads_back(quoted, value) else None


def _reads_back(written: str, value: str) -> bool:
"""Does ``written``, as the value of a key, parse back as ``value``?

The one question that matters, asked of the parser instead of answered from
memory. A note is the customer's own writing, so the interesting inputs are
the ones nobody thought of, and the parser is the only thing that knows them
all.

Every exception counts as "no". PyYAML raises more than ``YAMLError`` here:
its timestamp constructor calls ``datetime.date``, so a mistyped date like
``2026-02-30`` comes out as a ``ValueError`` and would otherwise travel all
the way up through ``vault_set_properties`` into the caller's face.
"""
try:
return yaml.safe_load(f"x: {written}") == {"x": value}
except Exception: # noqa: BLE001 - any failure to parse means "do not write this"
return False


def _looks_numeric(value: str) -> bool:
Expand Down
14 changes: 11 additions & 3 deletions knap_mcp/providers/filesystem/index.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
from typing import Any, Dict, Iterable, List, Optional, Set, Tuple

from . import markdown as md
from . import paths as vault_paths
Expand Down Expand Up @@ -102,12 +102,13 @@ def refresh(self, *, force: bool = False) -> None:
# decide what a hidden note may take part in: not link resolution, not
# backlinks, not tag counts, because a link must not resolve into
# `.trash` and a deleted note's tags are not the vault's tags.
root_resolved = self.root.resolve()
for absolute in vault_paths.walk_notes(self.root, include_hidden=True):
try:
stat = absolute.stat()
except OSError:
continue
rel = vault_paths.to_relative(self.root, absolute)
rel = vault_paths.relative_to_walked_root(root_resolved, absolute)
seen[rel] = (stat.st_size, stat.st_mtime_ns)

changed = False
Expand Down Expand Up @@ -184,14 +185,21 @@ def _rebuild_lookups(self) -> None:
for paths in self._by_stem.values():
paths.sort(key=lambda rel: (rel.count("/"), len(rel), rel))

# `seen` shadows `_backlinks` purely so the duplicate check is a hash
# lookup. It was `rel not in holders` on the list, which is a scan, and
# the note that suffers is the one every vault has: the MOC or index
# note the whole vault links to, whose holder list is as long as the
# vault. Same order out, same list type, one dict thrown away at the end.
seen: Dict[str, Set[str]] = {}
for rel, entry in self._notes.items():
if vault_paths.is_hidden(rel):
continue # a trashed note's links are not backlinks
for target in entry.link_targets:
resolved = self.resolve(target, from_path=rel)
if resolved and resolved != rel:
holders = self._backlinks.setdefault(resolved, [])
if rel not in holders:
if rel not in seen.setdefault(resolved, set()):
seen[resolved].add(rel)
holders.append(rel)

# -- reading ------------------------------------------------------------ #
Expand Down
23 changes: 22 additions & 1 deletion knap_mcp/providers/filesystem/markdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,12 +89,30 @@ def blank(match: re.Match) -> str:
return _INLINE_CODE_RE.sub(blank, without_fences)


def parse(raw: str) -> ParsedNote:
def read_body(path: Path) -> str:
"""Read a note for scanning: the text, and none of the bookkeeping.

``read_text`` also hashes the whole file to build a ``rev``, which is the
right thing when the caller is going to hand that rev to a client and
exactly wasted work when it is not. Search opens every candidate note and
uses none of it.
"""
data = path.read_bytes()
return unicodedata.normalize("NFC", data.decode("utf-8", errors="replace"))


def parse(raw: str, *, load_properties: bool = True) -> ParsedNote:
"""Split a note into frontmatter and body.

A frontmatter block only counts at the very start of the file and only when
it closes. An unterminated ``---`` is a horizontal rule in someone's note,
not a broken header, and treating it as one would swallow their document.

``load_properties=False`` skips the YAML parse and leaves ``frontmatter``
empty. The block is still split off the body, so ``body`` and
``body_scannable`` are unchanged; only the parsed mapping is missing. Search
wants exactly that, because the index already holds the properties and it
has compared them before it ever opens the file.
"""
note = ParsedNote(raw=raw)
if not raw.startswith(FRONTMATTER_FENCE):
Expand Down Expand Up @@ -124,6 +142,9 @@ def parse(raw: str) -> ParsedNote:
note.body = note.body[1:]
note.body_scannable = strip_code(note.body)

if not load_properties:
return note

inner = "\n".join(lines[1:close])
try:
loaded = yaml.safe_load(inner) if inner.strip() else None
Expand Down
28 changes: 28 additions & 0 deletions knap_mcp/providers/filesystem/paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,34 @@ def to_relative(root: Path, absolute: Path) -> str:
return unicodedata.normalize("NFC", resolved.relative_to(root_resolved).as_posix())


def relative_to_walked_root(root_resolved: Path, absolute: Path) -> str:
"""Vault-relative form of a path that ``walk_notes`` itself produced.

``to_relative`` is the one to use for a path that came from a caller: it
resolves both sides and refuses anything that lands outside, and that check
is the vulnerability class the whole module exists for. This is the other
case. ``walk_notes`` starts at ``root.resolve()`` and skips symlinks
outright, so every path it yields is already under the resolved root and got
there without traversing a link. Resolving it again asks the kernel to
confirm something the walk guaranteed, once per note, and on a vault of a
few thousand notes that realpath storm is most of what an index refresh
costs.

Pass ``root_resolved`` already resolved, once, by the caller. Only feed this
paths from the walk. The guarantee is a POSIX one: ``resolve`` does not cross
a bind mount, and ``walk_notes`` refuses symlinks. A Windows junction is
neither, so on Windows this is a check worth keeping rather than skipping.

Misuse still raises ``PathNotAllowedError`` like everything else here, rather
than the ``ValueError`` ``relative_to`` would give: comparing ``parts``
costs nothing, and a module whose whole job is one error type should not have
one entrance that throws a different one.
"""
if not _is_within(absolute, root_resolved):
raise _reject("is not under the vault root")
return unicodedata.normalize("NFC", absolute.relative_to(root_resolved).as_posix())


def is_hidden(rel: str) -> bool:
"""Whether any segment of the path is a dot entry.

Expand Down
7 changes: 5 additions & 2 deletions knap_mcp/providers/filesystem/search.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,10 +130,13 @@ def _match(self, entry: NoteEntry, needle: str) -> Optional[str]:
return f"{key}: {value}"

try:
text, _, _, _ = md.read_text(self.root / entry.path)
text = md.read_body(self.root / entry.path)
except OSError:
return None
note = md.parse(text)
# No rev and no YAML: this is the hot loop of the whole package, it opens
# every candidate note, and it uses neither. The properties were checked
# against the index a few lines up, before the file was opened at all.
note = md.parse(text, load_properties=False)
# Searched against the code-stripped copy so a hit inside a fenced block
# does not surface, but the excerpt is cut from the real body: offsets
# are shared between the two by construction.
Expand Down
15 changes: 13 additions & 2 deletions knap_mcp/providers/filesystem/writes.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ def write(
raise NoteNotFoundError(f"{rel} does not exist")
self._check_rev(rel, absolute, expected_rev, required=True)

if exists and mode in ("append", "prepend"):
if exists and mode in ("append", "prepend", "overwrite"):
current, _, _, _ = md.read_text(absolute)
parsed = md.parse(current)
if mode == "append":
Expand All @@ -109,8 +109,19 @@ def write(
# glues itself onto the last sentence is a corrupted paragraph.
separator = "" if parsed.body.endswith("\n\n") or not parsed.body else "\n"
new_body = f"{parsed.body.rstrip(chr(10))}\n{separator}{body}"
else:
elif mode == "prepend":
new_body = f"{body}\n\n{parsed.body.lstrip(chr(10))}"
else:
# Overwrite replaces the body and keeps the frontmatter, which is
# what `vault_update_note` tells the caller in three places: `body`
# is "the new body, replacing everything after the frontmatter",
# `properties` says omitting it "leaves the existing frontmatter
# alone", and the confirm prompt promises only what is "below the
# frontmatter". It used to drop the block instead, so a model doing
# an ordinary read-modify-write silently deleted `type`, `tags` and
# `aliases` -- the frontmatter discipline the whole retrieval story
# rests on, gone without appearing in the diff it asked about.
new_body = body
text = self._compose(parsed.frontmatter_raw, new_body)
else:
text = self._compose("", body)
Expand Down
Loading
Loading