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
5 changes: 5 additions & 0 deletions knap_mcp/knowledge.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,11 @@
it. Pass `create=true` to make it from the vault's template when it does not
exist yet.
- vault_get_attachment / vault_put_attachment: non-markdown files by path.
Images, recordings and PDFs are part of the vault, not an afterthought: a note
that says `![[scan.pdf]]` is a note whose content is partly in that file. Read
a note first and the embed comes back in `links` with `embed: true` and a
`resolved_path`, which is the path to pass. Large files come back as a refusal
naming the size rather than an enormous response.

Good habits:
- Resolve folder and note paths with the list and search tools before writing
Expand Down
44 changes: 39 additions & 5 deletions knap_mcp/providers/filesystem/index.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,13 @@ def __init__(self, root: Path):
self._by_path_key: Dict[str, str] = {}
self._by_alias: Dict[str, str] = {}
self._backlinks: Dict[str, List[str]] = {}
# The non-note half of the vault, and it is deliberately kept in its own
# pair of structures rather than mixed into the four above. A note must
# keep winning every lookup it wins today: attachments are consulted
# only where nothing else matched, so nothing that resolves now can
# start resolving somewhere else because somebody dropped in a PNG.
self._assets: set[str] = set()
self._assets_by_key: Dict[str, str] = {}

# -- freshness ---------------------------------------------------------- #

Expand Down Expand Up @@ -124,6 +131,17 @@ def refresh(self, *, force: bool = False) -> None:
del self._notes[rel]
changed = True

# Attachments are names, not content: nothing is parsed and no mtime is
# tracked, because the only question ever asked of one is whether a link
# target means it. Editing an image cannot change that answer.
assets = {
vault_paths.to_relative(self.root, absolute)
for absolute in vault_paths.walk_attachments(self.root, include_hidden=True)
}
if assets != self._assets:
self._assets = assets
changed = True

self._last_walk = now
self._loaded = True
if changed or not self._by_path_key:
Expand Down Expand Up @@ -165,6 +183,16 @@ def _rebuild_lookups(self) -> None:
self._by_path_key = {}
self._by_alias = {}
self._backlinks = {}
self._assets_by_key = {}

# Shallowest first, so two files of the same name in different folders
# resolve to the one nearer the root -- Obsidian's tie-break, and the
# same one _by_stem is sorted by below.
for rel in sorted(self._assets, key=lambda p: (p.count("/"), len(p), p)):
if vault_paths.is_hidden(rel):
continue # an embed must not resolve into .trash or .obsidian
self._assets_by_key.setdefault(rel.lower(), rel)
self._assets_by_key.setdefault(rel.rsplit("/", 1)[-1].lower(), rel)

for rel, entry in self._notes.items():
if vault_paths.is_hidden(rel):
Expand Down Expand Up @@ -286,7 +314,7 @@ def resolve(self, target: str, *, from_path: str = "") -> Optional[str]:
lowered = needle.lower()

if "/" in lowered:
return self._by_path_key.get(lowered)
return self._by_path_key.get(lowered) or self._assets_by_key.get(lowered)

stem = lowered[:-3] if lowered.endswith(".md") else lowered

Expand All @@ -295,6 +323,9 @@ def resolve(self, target: str, *, from_path: str = "") -> Optional[str]:
sibling = self._by_path_key.get(f"{folder.lower()}/{stem}")
if sibling:
return sibling
sibling = self._assets_by_key.get(f"{folder.lower()}/{lowered}")
if sibling:
return sibling

alias = self._by_alias.get(lowered)
if alias:
Expand All @@ -305,10 +336,13 @@ def resolve(self, target: str, *, from_path: str = "") -> Optional[str]:
return direct

candidates = self._by_stem.get(stem)
if not candidates:
return None
# Sorted shallowest-then-shortest in _rebuild_lookups.
return candidates[0]
if candidates:
# Sorted shallowest-then-shortest in _rebuild_lookups.
return candidates[0]

# Last, and only here: an attachment. `![[diagram.png]]` carries its
# extension, so it never reaches this point looking like a note.
return self._assets_by_key.get(lowered)

def shortest_unique_form(self, rel: str) -> str:
"""How a link to this note should be written after a move.
Expand Down
34 changes: 33 additions & 1 deletion knap_mcp/providers/filesystem/paths.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,38 @@ def walk_notes(
this is the cheap half of indexing, and it is why a refresh over ten
thousand notes costs milliseconds rather than a read of the whole vault.
"""
return _walk(root, subfolder=subfolder, include_hidden=include_hidden, notes=True)


def walk_attachments(
root: Path,
*,
subfolder: str = "",
include_hidden: bool = False,
) -> Iterable[Path]:
"""Yield every file under ``root`` that is *not* a note.

Images, recordings, PDFs: the half of a vault the tools do not parse but
still have to be able to name. A link is only resolvable to something the
index has heard of, so an embed of a file nothing ever walked reads as
broken even with the file sitting right there on disk.
"""
return _walk(root, subfolder=subfolder, include_hidden=include_hidden, notes=False)


def _walk(
root: Path,
*,
subfolder: str,
include_hidden: bool,
notes: bool,
) -> Iterable[Path]:
"""One tree walk, filtered by whether a name ends in ``.md``.

Shared rather than duplicated because the traversal is the expensive part
and the filter is one comparison: an index that wants both halves should
pay for one walk, not two.
"""
start = resolve_in_vault(root, subfolder) if subfolder else root.resolve()
if not start.is_dir():
return
Expand All @@ -219,5 +251,5 @@ def walk_notes(
continue
if entry.is_dir():
stack.append(entry)
elif is_note(name):
elif is_note(name) == notes:
yield entry
55 changes: 55 additions & 0 deletions tests/test_provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,61 @@ def test_an_empty_target_is_none(self, provider) -> None:
assert provider.resolve_link("") is None


class TestAttachmentResolution:
"""An embed of a non-note file has to resolve, or the AI cannot fetch it.

`vault_get_attachment` tells the caller to pass the embed's `resolved_path`,
so a null there is not a cosmetic gap: it is the tool documenting a route
that does not exist.
"""

def test_an_embed_of_an_attachment_resolves_by_path(self, provider) -> None:
assert provider.resolve_link("Attachments/diagram.png") == "Attachments/diagram.png"

def test_an_embed_resolves_by_bare_filename(self, provider) -> None:
assert provider.resolve_link("diagram.png") == "Attachments/diagram.png"

def test_a_read_note_hands_back_the_path_to_fetch(self, provider) -> None:
note = provider.read("index.md")
embed = next(link for link in note.links if link.embed)
assert embed.resolved_path == "Attachments/diagram.png"
assert provider.read_binary(embed.resolved_path).size > 0

def test_a_sibling_attachment_wins_over_one_nearer_the_root(
self, vault_root: Path, provider
) -> None:
(vault_root / "shot.png").write_bytes(b"root")
(vault_root / "Projects" / "shot.png").write_bytes(b"sibling")
provider.index.refresh(force=True)
assert (
provider.resolve_link("shot.png", from_path="Projects/Meeting notes.md")
== "Projects/shot.png"
)
assert provider.resolve_link("shot.png", from_path="index.md") == "shot.png"

def test_a_note_still_wins_a_name_an_attachment_also_claims(
self, vault_root: Path, provider
) -> None:
"""The whole reason attachments are consulted last."""
(vault_root / "Projects" / "index.png").write_bytes(b"decoy")
provider.index.refresh(force=True)
assert provider.resolve_link("index") == "index.md"

def test_an_attachment_in_trash_does_not_resolve(self, vault_root: Path, provider) -> None:
(vault_root / ".trash").mkdir(exist_ok=True)
(vault_root / ".trash" / "deleted.png").write_bytes(b"gone")
provider.index.refresh(force=True)
assert provider.resolve_link("deleted.png") is None

def test_a_new_attachment_is_picked_up_without_a_restart(
self, vault_root: Path, provider
) -> None:
assert provider.resolve_link("late.pdf") is None
(vault_root / "late.pdf").write_bytes(b"%PDF-1.4")
provider.index.refresh(force=True)
assert provider.resolve_link("late.pdf") == "late.pdf"


class TestReadAndSearch:
def test_read_reports_resolved_and_unresolved_links(self, provider) -> None:
note = provider.read("Areas/Work/Acme.md")
Expand Down
Loading