diff --git a/knap_mcp/knowledge.py b/knap_mcp/knowledge.py index 172a3a7..4cd681a 100644 --- a/knap_mcp/knowledge.py +++ b/knap_mcp/knowledge.py @@ -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 diff --git a/knap_mcp/providers/filesystem/index.py b/knap_mcp/providers/filesystem/index.py index a55b445..b350c0a 100644 --- a/knap_mcp/providers/filesystem/index.py +++ b/knap_mcp/providers/filesystem/index.py @@ -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 ---------------------------------------------------------- # @@ -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: @@ -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): @@ -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 @@ -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: @@ -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. diff --git a/knap_mcp/providers/filesystem/paths.py b/knap_mcp/providers/filesystem/paths.py index ce5c3e0..e5c21e1 100644 --- a/knap_mcp/providers/filesystem/paths.py +++ b/knap_mcp/providers/filesystem/paths.py @@ -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 @@ -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 diff --git a/tests/test_provider.py b/tests/test_provider.py index 22490c6..1e0c484 100644 --- a/tests/test_provider.py +++ b/tests/test_provider.py @@ -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")