diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..2096886 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +.git +.idea +.venv +__pycache__ +tests +.claude +index +skills +agents +0*_*/ diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..30cf57e --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,10 @@ +# Default ignored files +/shelf/ +/workspace.xml +# Editor-based HTTP Client requests +/httpRequests/ +# Ignored default folder with query files +/queries/ +# Datasource local storage ignored files +/dataSources/ +/dataSources.local.xml diff --git a/.idea/knowledge-base-framework.iml b/.idea/knowledge-base-framework.iml new file mode 100644 index 0000000..d6ebd48 --- /dev/null +++ b/.idea/knowledge-base-framework.iml @@ -0,0 +1,9 @@ + + + + + + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..b1ae761 --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,7 @@ + + + + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..687d7a3 --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..35eb1dd --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/AGENT_QUERY_GUIDE.md b/AGENT_QUERY_GUIDE.md index 2e09bda..9cd75a1 100644 --- a/AGENT_QUERY_GUIDE.md +++ b/AGENT_QUERY_GUIDE.md @@ -4,14 +4,17 @@ title: Agent Query Guide type: meta status: active tags: [agents, jq, graph] -last_reviewed: 2026-06-16 +last_reviewed: 2026-10-06 --- # Agent Query Guide -`index/graph.json` is the selected vault's navigation index. MCP is the canonical -cross-repository interface: `graph_summary`, `graph_search_nodes`, -`graph_neighbors`, and `read_note`; use `refresh_graph(validate=true)` to update it. +The graph is the selected vault's navigation index, built in memory from the +Markdown by `kb_framework.graph.build_graph` — no file is read or committed. +MCP is the canonical cross-repository interface: `graph_summary`, +`graph_search_nodes`, `graph_neighbors`, and `read_note`; the server rebuilds +the graph after every write, and `refresh_graph(validate=true)` rebuilds it on +demand and reports validation problems. MCP-only agents use the tools above; shell examples below are for operators, not agent fallback. For those queries, explicitly set `KNOWLEDGE_BASE_ROOT` and change into @@ -72,7 +75,9 @@ optional JSON brace suffix carrying the relationship and annotation: fails on any that remain. Locate and convert them with `python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/list-wiki-links.py" --vault-root "$KNOWLEDGE_BASE_ROOT"`. -Examples with `jq`: +Examples with `jq` (operators only; first dump the graph with +`python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" --write`, +which writes `index/graph.json` — keep `index/` out of git): ```bash # list repos @@ -118,8 +123,8 @@ Validation: # outside their vocabularies, OR any remaining legacy [[wiki links]] python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" --validate -# regenerate graph and validate -python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" && python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" --validate +# write a graph dump for jq AND validate in one run +python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" --write --validate # list any remaining legacy [[wiki links]] (rejected by --validate) to convert python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/list-wiki-links.py" --vault-root "$KNOWLEDGE_BASE_ROOT" diff --git a/Makefile b/Makefile index 843d197..7c08c7d 100644 --- a/Makefile +++ b/Makefile @@ -15,7 +15,8 @@ generate-graph: validate-graph: $(PYTHON) scripts/generate-graph.py --vault-root "$(VAULT_ROOT)" --validate -graph: generate-graph validate-graph +# `graph` is the everyday gate: validate only, nothing written. +graph: validate-graph install-hooks: $(PYTHON) scripts/install-hooks.py --vault-root "$(VAULT_ROOT)" diff --git a/README.md b/README.md index 3ceabce..0edc7cb 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ writing notes. 02_areas/ ongoing responsibilities & standards (policies, value streams, decisions) 03_resources/ reference material (repo info, ingested externals) 04_archives/ inactive items moved out of the three live categories -index/ generated graph (index/graph.json) +kb_framework/ importable core: in-memory graph build, queries, note access policy scripts/ graph generator, link inventory, MCP server, git hook agents/ subagent definitions, symlinked into .claude/agents/ skills/ skill definitions, symlinked into .claude/skills/ @@ -37,10 +37,10 @@ folder with an `index.md`. ## Bootstrap ```sh -make install-hooks # pre-commit hook: regenerate + validate the graph +make install-hooks # pre-commit hook: validate the graph (nothing written) make install-global-claude # link skills & agents into ~/.claude so any repo can use them make install-mcp-claude # register scripts/mcp-server.py as the knowledge-database MCP server -make graph # generate index/graph.json and validate it +make graph # validate the graph (built in memory); `make generate-graph` writes index/graph.json make test # run the tooling tests ``` @@ -66,13 +66,15 @@ make -C /path/to/framework install-mcp-claude VAULT_ROOT=/path/to/vault The hook installer honors Git's hooks directory (including worktrees and `core.hooksPath`), preserves unmanaged hooks, and writes a wrapper pointing at framework tooling. Reinstall after moving the framework or vault. The hook -regenerates and validates the selected vault, then rejects unstaged graph changes. +validates the selected vault's graph in memory; it writes nothing and never +requires a staged `index/graph.json`. Configure each stdio MCP client with command `python3` and arguments `["/path/to/framework/scripts/mcp-server.py", "--vault-root", "/path/to/vault"]`. -MCP reads, writes and Git operations use that vault; `refresh_graph` invokes -the framework generator by absolute path and writes the selected graph even -when validation is requested. A README resource is advertised only if present. +MCP reads, writes and Git operations use that vault. The graph is built in +memory on first use and rebuilt after every write; `refresh_graph` rebuilds it +on demand and, with `validate=true`, applies the commit gate. No graph file is +read or written. A README resource is advertised only if present. Global skill and agent symlinks must point into this framework. MCP is their only interface for vault access; unavailable operations are reported without @@ -87,18 +89,18 @@ tools/list, graph_summary, search, read and refresh. Only remove old tool copies after this passes. On failure restore the configs and symlinks and retain the old copies. Restart existing clients to load the new server command. -A minimal data repository contains `01_projects/` through `04_archives/`, -`index/graph.json` and repository hygiene. Before removing a root hub, replace +A minimal data repository contains `01_projects/` through `04_archives/` +and repository hygiene; `index/` is not committed. Before removing a root hub, replace its inbound links with links between PARA hubs and validate the graph. Preserve untracked files separately from the tracked-file cleanup. ## Generation -- Run `make graph` -- Output is written to `index/graph.json` +- The graph is derived in memory by `kb_framework.graph.build_graph`; run `make graph` to validate it +- `make generate-graph` writes `index/graph.json` for operators who want a file; keep `index/` out of git ## Rules - Markdown is the source of truth -- `index/graph.json` is derived and generated +- The graph is derived; never hand-edit or commit a graph dump - Keep notes linked using typed markdown links: `[text](note-id){"type":"…"}` (legacy `[[wiki links]]` are rejected by `make validate-graph`) - Organize by PARA actionability; move notes between buckets as that changes - Fix broken links — find them with `make graph` or `make validate-graph` diff --git a/agents/the-knowing-one.md b/agents/the-knowing-one.md index 02e9e38..c1232d3 100644 --- a/agents/the-knowing-one.md +++ b/agents/the-knowing-one.md @@ -21,7 +21,7 @@ external URLs that notes cite — never to reach the vault itself. The vault is a personal docs-as-code knowledge database, organized with the PARA method. Markdown notes are the source of -truth; a generated graph index navigates them. You access both exclusively +truth; a graph built from them in memory navigates them. You access both exclusively through MCP. ## MCP tools @@ -35,7 +35,7 @@ through MCP. - `graph_neighbors` — outbound + inbound links for a node id, each with its edge `type` + `annotation` (traverse the graph). - `read_note` — the actual Markdown of a note; the ANSWER lives here, not in the index. -- `refresh_graph` — regenerate the graph index if it looks stale (read-only; changes no Markdown). +- `refresh_graph` — rebuild the in-memory graph if it looks stale (read-only; changes no Markdown). ## Mandate diff --git a/agents/the-scribe.md b/agents/the-scribe.md index b583524..c6a3840 100644 --- a/agents/the-scribe.md +++ b/agents/the-scribe.md @@ -1,6 +1,6 @@ --- name: the-scribe -description: MCP-only resource ingestor for the knowledge-base vault (PARA-organized). Hand it a URL, a local file path, or a PDF and it runs the `knowledge-base-ingest-resource` pipeline entirely through the `knowledge-database` MCP server — captures the original bytes (text or binary) into `raw/`, writes the resource note and one note per topic, wires typed markdown links, regenerates + validates the graph, then commits the named paths and pushes. Has no shell. Use when the user says "ingest", "add this resource", or "save this link/paper" and wants the result written and published. +description: MCP-only resource ingestor for the knowledge-base vault (PARA-organized). Hand it a URL, a local file path, or a PDF and it runs the `knowledge-base-ingest-resource` pipeline entirely through the `knowledge-database` MCP server — captures the original bytes (text or binary) into `raw/`, writes the resource note and one note per topic, wires typed markdown links, rebuilds + validates the graph, then commits the named paths and pushes. Has no shell. Use when the user says "ingest", "add this resource", or "save this link/paper" and wants the result written and published. tools: Read, WebFetch, Skill, mcp__knowledge-database__* mcpServers: - knowledge-database @@ -44,7 +44,7 @@ Write: - `edit_note(path, old_string, new_string, replace_all?)` — exact-string edit, for backlinks into notes that already exist. -- `refresh_graph(validate=true)` — regenerate and validate `index/graph.json`. +- `refresh_graph(validate=true)` — rebuild the in-memory graph and validate it; `problems` lists what to fix. Publish only when the caller explicitly authorizes it: - `git_commit(message, paths[])` — stage the named paths and commit. @@ -90,12 +90,12 @@ do not translate missing operations into shell commands. `type` is from the controlled vocabulary — `related, supersedes, superseded-by, depends-on, part-of, parent, derived-from, see-also, source`. Legacy `[[wiki links]]` fail validation. -- **Never write the graph.** `index/graph.json` is derived; `write_note` refuses - it. Use `refresh_graph`. Fix every unresolved link until validation is clean — - the pre-commit hook rejects a stale or invalid graph. +- **Never write a graph file.** The graph is derived in memory; `write_note` + refuses `index/graph.json`. Use `refresh_graph`. Fix every unresolved link until + validation is clean — the pre-commit hook rejects an invalid graph. - **Name every path you commit.** `git_commit` takes an explicit list; wildcard - staging is rejected by the server. Include the resource folder, every - backlinked note, and `index/graph.json`. + staging is rejected by the server. Include the resource folder and every + backlinked note; there is no graph file to commit. - **Follow the caller's publishing instructions.** Do not commit or push unless explicitly authorized. Use the selected vault's branch and review policy; do not assume a branch name or a direct-push workflow. Without publishing diff --git a/kb_framework/__init__.py b/kb_framework/__init__.py new file mode 100644 index 0000000..6cdafad --- /dev/null +++ b/kb_framework/__init__.py @@ -0,0 +1,10 @@ +"""kb_framework: the importable core of the knowledge-base framework. + +Exists so the stdio MCP server (`scripts/mcp-server.py`) and the hosted +read-only connector share one implementation of graph building, graph +queries, note access policy and framework instructions. The graph is built +in memory from the vault's Markdown; nothing here writes to disk. +""" + +__all__ = ["graph", "query", "query_tools", "vault", "instructions"] +__version__ = "0.2.0" diff --git a/kb_framework/graph.py b/kb_framework/graph.py new file mode 100644 index 0000000..7190a0b --- /dev/null +++ b/kb_framework/graph.py @@ -0,0 +1,481 @@ +"""Build and validate the vault graph in memory. + +One node per Markdown note (frontmatter plus link counts), one edge per typed +markdown link. `build_graph` walks a vault root and returns a `GraphBuild`; it +never touches `index/graph.json`. `validation_problems` is the gate the +pre-commit hook and `--validate` apply: unresolved targets, edge relationship +types or note `type` values outside their controlled vocabularies, and legacy +`[[wiki links]]` are all rejected. +""" + +from __future__ import annotations + +import json +import re +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, List, Tuple + + +# Obsidian-style wiki link: [[target#anchor|alias]] +WIKI_LINK_RE = re.compile(r"\[\[([^\]|#]+)(?:#[^\]|]+)?(?:\|([^\]]+))?\]\]") +# Standard markdown link: [display](target); (? edge anchor_text + kind: str # "wiki_link" | "markdown_link" + rel_type: str # semantic relationship type (default "related") + annotation: str # per-edge "what to expect there" note + lineno: int # 1-based line number in the note body + raw: str # the raw link text as written (for the inventory script) + is_external: bool # markdown link whose target is a URL/anchor -> no edge + + +@dataclass +class Note: + path: Path + rel_path: str + stem: str + meta: Dict[str, object] + title: str + note_id: str + links: List[LinkRef] + + +@dataclass(frozen=True) +class UnresolvedLink: + from_id: str + from_path: str + target: str + + +@dataclass(frozen=True) +class WikiLinkUse: + from_id: str + from_path: str + target: str + lineno: int + raw: str + + +def parse_frontmatter(text: str) -> Tuple[Dict[str, object], str]: + if not text.startswith("---\n"): + return {}, text + + end = text.find("\n---\n", 4) + if end == -1: + return {}, text + + fm_block = text[4:end] + body = text[end + 5 :] + meta: Dict[str, object] = {} + + for raw in fm_block.splitlines(): + line = raw.strip() + if not line or line.startswith("#") or ":" not in line: + continue + key, value = line.split(":", 1) + key = key.strip() + value = value.strip() + if value.startswith("[") and value.endswith("]"): + inner = value[1:-1].strip() + if inner: + meta[key] = [v.strip().strip('"').strip("'") for v in inner.split(",")] + else: + meta[key] = [] + else: + meta[key] = value.strip('"').strip("'") + + return meta, body + + +def derive_title(meta: Dict[str, object], body: str, stem: str) -> str: + if isinstance(meta.get("title"), str) and meta["title"].strip(): + return str(meta["title"]).strip() + for line in body.splitlines(): + if line.startswith("# "): + return line[2:].strip() + return stem.replace("-", " ").strip().title() + + +def mask_code(body: str) -> str: + """Blank out code so links inside code samples are ignored, preserving line + boundaries so line-based parsing (gloss fallback, line numbers) stays aligned.""" + body = FENCED_CODE_RE.sub(lambda m: "\n" * m.group(0).count("\n"), body) + body = INLINE_CODE_RE.sub(lambda m: " " * len(m.group(0)), body) + return body + + +def scan_brace(text: str, start: int) -> Tuple[object, int]: + """Given text[start] == '{', scan to the matching '}' respecting JSON string + quoting, and json.loads the block. Returns (obj_or_None, end_index_after_brace). + On an unterminated or malformed brace, returns (None, start) so the caller + treats it as no brace.""" + depth = 0 + in_str = False + esc = False + i = start + while i < len(text): + c = text[i] + if in_str: + if esc: + esc = False + elif c == "\\": + esc = True + elif c == '"': + in_str = False + elif c == '"': + in_str = True + elif c == "{": + depth += 1 + elif c == "}": + depth -= 1 + if depth == 0: + raw = text[start : i + 1] + try: + return json.loads(raw), i + 1 + except (json.JSONDecodeError, ValueError): + return None, start + i += 1 + return None, start + + +def clean_md_target(raw: str) -> str: + """Normalize a markdown link target: drop a trailing ``"title"`` and <> wrappers.""" + raw = raw.strip() + if raw.startswith("<") and raw.endswith(">"): + raw = raw[1:-1].strip() + # `[text](target "title")` -> target + for sep in (' "', " '"): + if sep in raw: + raw = raw.split(sep, 1)[0].strip() + return raw + + +def parse_links(body: str) -> List[LinkRef]: + masked = mask_code(body) + refs: List[LinkRef] = [] + + for lineno, line in enumerate(masked.splitlines(), start=1): + # Collect wiki + markdown link matches on this line, in order. + raw_matches: List[Tuple[int, int, str, str, str, str]] = [] + for m in WIKI_LINK_RE.finditer(line): + target = m.group(1).strip() + alias = (m.group(2) or target).strip() + if target: + raw_matches.append( + (m.start(), m.end(), "wiki_link", target, alias, m.group(0)) + ) + for m in MD_LINK_RE.finditer(line): + target = clean_md_target(m.group(2)) + display = m.group(1).strip() + if target: + raw_matches.append( + ( + m.start(), + m.end(), + "markdown_link", + target, + display or target, + m.group(0), + ) + ) + raw_matches.sort(key=lambda x: x[0]) + + line_refs: List[Tuple[LinkRef, int]] = [] # (ref, end-index-after-brace) + for start, end, kind, target, anchor, raw in raw_matches: + rel_type, annotation = "related", "" + brace_end = end + if end < len(line) and line[end] == "{": + obj, be = scan_brace(line, end) + if isinstance(obj, dict): + brace_end = be + t = obj.get("type") + if isinstance(t, str) and t.strip(): + rel_type = t.strip() + info = obj.get("link-info") + if isinstance(info, str) and info.strip(): + annotation = info.strip() + is_external = kind == "markdown_link" and bool( + EXTERNAL_TARGET_RE.match(target) + ) + ref = LinkRef( + target=target, + anchor=anchor, + kind=kind, + rel_type=rel_type, + annotation=annotation, + lineno=lineno, + raw=raw, + is_external=is_external, + ) + line_refs.append((ref, brace_end)) + + # Annotation fallback: a trailing "— gloss" applies only when the link is + # the sole link on its line (so we don't mis-assign a shared gloss). + if len(line_refs) == 1: + ref, brace_end = line_refs[0] + if not ref.annotation: + gloss = GLOSS_RE.match(line[brace_end:]) + if gloss: + ref.annotation = gloss.group(1).strip() + + refs.extend(ref for ref, _ in line_refs) + + return refs + + +def norm(s: str) -> str: + return s.strip().lower() + + +def build_notes(root: Path) -> List[Note]: + notes: List[Note] = [] + for p in sorted(root.rglob("*.md")): + rel = p.relative_to(root) + if rel.parts and rel.parts[0] in {"index", "scripts", ".obsidian"}: + continue + # `raw/` holds untouched ingested originals (external READMEs, specs, …), + # not vault notes — skip them so their file-relative markdown links are not + # treated as (broken) graph edges. + if "raw" in rel.parts: + continue + + text = p.read_text(encoding="utf-8") + meta, body = parse_frontmatter(text) + title = derive_title(meta, body, p.stem) + note_id = str(meta.get("id") or rel.with_suffix("")).strip() + links = parse_links(body) + # parse_links numbers lines within `body`; shift to absolute file line + # numbers by the number of lines the frontmatter block occupied. + line_offset = text.count("\n", 0, len(text) - len(body)) + if line_offset: + for ref in links: + ref.lineno += line_offset + + notes.append( + Note( + path=p, + rel_path=rel.as_posix(), + stem=p.stem, + meta=meta, + title=title, + note_id=note_id, + links=links, + ) + ) + return notes + + +def build_alias_map(notes: List[Note]) -> Dict[str, str]: + alias_map: Dict[str, str] = {} + stem_buckets: Dict[str, List[str]] = defaultdict(list) + + for n in notes: + stem_buckets[norm(n.stem)].append(n.note_id) + + for n in notes: + alias_map[norm(n.note_id)] = n.note_id + alias_map[norm(n.rel_path)] = n.note_id + alias_map[norm(Path(n.rel_path).with_suffix("").as_posix())] = n.note_id + alias_map[norm(n.title)] = n.note_id + + for stem, ids in stem_buckets.items(): + if len(ids) == 1: + alias_map[stem] = ids[0] + + return alias_map + + +def generate_graph( + root: Path, existing_generated_at: str = "" +) -> Tuple[Dict[str, object], List[UnresolvedLink], List[WikiLinkUse]]: + notes = build_notes(root) + alias_map = build_alias_map(notes) + + edges_set = set() + outbound: Dict[str, List[str]] = defaultdict(list) + inbound_count: Dict[str, int] = defaultdict(int) + unresolved_set = set() + wiki_uses_set = set() + + for n in notes: + for ref in n.links: + # Legacy [[wiki links]] are still parsed (so we can flag them) but are + # no longer an accepted link form — --validate rejects any that remain. + if ref.kind == "wiki_link": + wiki_uses_set.add( + WikiLinkUse( + from_id=n.note_id, + from_path=n.rel_path, + target=ref.target, + lineno=ref.lineno, + raw=ref.raw, + ) + ) + if ref.is_external: + continue + target = alias_map.get(norm(ref.target)) + if not target: + unresolved_set.add( + UnresolvedLink( + from_id=n.note_id, + from_path=n.rel_path, + target=ref.target, + ) + ) + continue + edge = (n.note_id, target, ref.kind, ref.rel_type, ref.annotation, ref.anchor) + if edge in edges_set: + continue + edges_set.add(edge) + outbound[n.note_id].append(target) + inbound_count[target] += 1 + + nodes = [] + for n in sorted(notes, key=lambda x: x.note_id): + tags = n.meta.get("tags", []) + if not isinstance(tags, list): + tags = [] + nodes.append( + { + "id": n.note_id, + "title": n.title, + "path": n.rel_path, + "repo": n.meta.get("repo", "global"), + "type": n.meta.get("type", "note"), + "status": n.meta.get("status", "active"), + "tags": tags, + "last_reviewed": n.meta.get("last_reviewed", ""), + "outbound_links": sorted(set(outbound.get(n.note_id, []))), + "inbound_count": inbound_count.get(n.note_id, 0), + } + ) + + edges = [ + { + "from": src, + "to": dst, + "kind": kind, + "type": rel_type, + "anchor_text": anchor, + "annotation": annotation, + "weight": 1.0, + } + for src, dst, kind, rel_type, annotation, anchor in sorted(edges_set) + ] + + graph = { + "version": 1, + "generated_at": existing_generated_at + if existing_generated_at + else datetime.now(timezone.utc).isoformat(), + "nodes": nodes, + "edges": edges, + } + unresolved = sorted(unresolved_set, key=lambda x: (x.from_path, x.target)) + wiki_uses = sorted(wiki_uses_set, key=lambda x: (x.from_path, x.lineno, x.raw)) + return graph, unresolved, wiki_uses + + +def find_invalid_types(graph: Dict[str, object]) -> List[Dict[str, object]]: + return [ + e + for e in graph.get("edges", []) # type: ignore[union-attr] + if e.get("type") not in RELATION_TYPES + ] + + +def find_invalid_node_types(graph: Dict[str, object]) -> List[Dict[str, object]]: + return [ + n + for n in graph.get("nodes", []) # type: ignore[union-attr] + if n.get("type") not in NOTE_TYPES + ] + + +@dataclass +class GraphBuild: + """Result of one in-memory build: the graph plus everything --validate reports.""" + + graph: Dict[str, object] + unresolved: List[UnresolvedLink] = field(default_factory=list) + wiki_uses: List[WikiLinkUse] = field(default_factory=list) + + @property + def invalid_edge_types(self) -> List[Dict[str, object]]: + return find_invalid_types(self.graph) + + @property + def invalid_node_types(self) -> List[Dict[str, object]]: + return find_invalid_node_types(self.graph) + + @property + def ok(self) -> bool: + return not validation_problems(self) + + +def build_graph(root: Path, existing_generated_at: str = "") -> GraphBuild: + """Build the graph for `root` in memory.""" + graph, unresolved, wiki_uses = generate_graph(Path(root), existing_generated_at) + return GraphBuild(graph=graph, unresolved=unresolved, wiki_uses=wiki_uses) + + +def validation_problems(build: GraphBuild) -> List[str]: + """Human-readable validation failures, empty when the build is clean.""" + problems: List[str] = [] + for link in build.unresolved: + problems.append(f"unresolved link: {link.from_path}: [[{link.target}]]") + for e in build.invalid_edge_types: + problems.append( + f"unknown relationship type: {e.get('from')} -> {e.get('to')}: type={e.get('type')!r}" + ) + for n in build.invalid_node_types: + problems.append(f"unknown note type: {n.get('id')} ({n.get('path')}): type={n.get('type')!r}") + for w in build.wiki_uses: + problems.append(f"legacy wiki link: {w.from_path}:{w.lineno}: {w.raw}") + return problems diff --git a/kb_framework/instructions.py b/kb_framework/instructions.py new file mode 100644 index 0000000..919a2e7 --- /dev/null +++ b/kb_framework/instructions.py @@ -0,0 +1,35 @@ +"""Locate and read the framework's agent instructions. + +`read_instructions` returns AGENTS.md and AGENT_QUERY_GUIDE.md so an agent +whose only access is MCP still gets the operating rules. In a source checkout +the documents live next to this package (the framework root); in an installed +wheel they are bundled under `kb_framework/_docs/`. Both lookups are explicit — +there is never a fallback to files inside the data vault. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Dict + +DOCUMENT_NAMES = ("AGENTS.md", "AGENT_QUERY_GUIDE.md") +_PACKAGE_DIR = Path(__file__).resolve().parent +# Source checkout: the framework repository root (parent of this package). +FRAMEWORK_ROOT = _PACKAGE_DIR.parent +_BUNDLED_DOCS = _PACKAGE_DIR / "_docs" + + +def instructions_root() -> Path: + """Directory holding the instruction documents for this installation.""" + if (FRAMEWORK_ROOT / DOCUMENT_NAMES[0]).is_file(): + return FRAMEWORK_ROOT + return _BUNDLED_DOCS + + +def read_instructions() -> Dict[str, Any]: + root = instructions_root() + documents = [ + {"path": name, "content": (root / name).read_text(encoding="utf-8")} + for name in DOCUMENT_NAMES + ] + return {"framework_root": str(root), "documents": documents} diff --git a/kb_framework/query.py b/kb_framework/query.py new file mode 100644 index 0000000..b27a431 --- /dev/null +++ b/kb_framework/query.py @@ -0,0 +1,159 @@ +"""Pure query functions over an in-memory graph dict. + +These back the read-only MCP tools (`graph_summary`, `graph_search_nodes`, +`graph_neighbors`) and the id-to-path lookup `read_note` needs. They take the +graph as data and never read the filesystem, so the stdio server and the +connector return identical payloads for identical graphs. +""" + +from __future__ import annotations + +from collections import Counter +from typing import Any, Dict, List + +# graph_summary defaults: enough most-linked notes to spot the hubs, and enough +# tags to cover the vocabulary an agent actually filters on. +DEFAULT_TOP_LINKED = 10 +DEFAULT_TOP_TAGS = 50 +DEFAULT_SEARCH_LIMIT = 30 +DEFAULT_NEIGHBOR_LIMIT = 50 + + +def nodes_by_id(graph: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: + nodes = graph.get("nodes", []) + return {str(node.get("id")): node for node in nodes if isinstance(node, dict)} + + +def match_filters(node: Dict[str, Any], args: Dict[str, Any]) -> bool: + repo = args.get("repo") + note_type = args.get("type") + status = args.get("status") + tag = args.get("tag") + query = str(args.get("query", "")).strip().lower() + + if repo and node.get("repo") != repo: + return False + if note_type and node.get("type") != note_type: + return False + if status and node.get("status") != status: + return False + if tag and tag not in (node.get("tags") or []): + return False + if query: + haystacks = [ + str(node.get("id", "")).lower(), + str(node.get("title", "")).lower(), + str(node.get("path", "")).lower(), + " ".join(str(t).lower() for t in (node.get("tags") or [])), + ] + if not any(query in h for h in haystacks): + return False + return True + + +def summary( + graph: Dict[str, Any], + topLinked: int = DEFAULT_TOP_LINKED, + topTags: int = DEFAULT_TOP_TAGS, +) -> Dict[str, Any]: + nodes = graph.get("nodes", []) + edges = graph.get("edges", []) + top_n = int(topLinked) + top_tags = int(topTags) + + by_repo: Counter[str] = Counter(str(node.get("repo", "unknown")) for node in nodes) + tag_counts: Counter[str] = Counter( + str(tag) for node in nodes for tag in (node.get("tags") or []) + ) + # Frequent tags first, alphabetical within a tie, so a truncated list is + # both the most useful slice and stable across calls. + ranked_tags = sorted(tag_counts.items(), key=lambda item: (-item[1], item[0])) + + sorted_nodes = sorted( + nodes, + key=lambda n: int(n.get("inbound_count", 0)), + reverse=True, + )[:top_n] + return { + "version": graph.get("version"), + "generated_at": graph.get("generated_at"), + "node_count": len(nodes), + "edge_count": len(edges), + "by_repo": dict(by_repo), + "tag_count": len(tag_counts), + "by_tag": [ + {"tag": tag, "count": count} for tag, count in ranked_tags[:top_tags] + ], + "top_linked": [ + { + "id": node.get("id"), + "title": node.get("title"), + "path": node.get("path"), + "inbound_count": node.get("inbound_count", 0), + } + for node in sorted_nodes + ], + } + + +def search_nodes(graph: Dict[str, Any], args: Dict[str, Any]) -> Dict[str, Any]: + nodes = graph.get("nodes", []) + limit = int(args.get("limit", DEFAULT_SEARCH_LIMIT)) + filtered = [node for node in nodes if match_filters(node, args)] + filtered.sort(key=lambda n: str(n.get("id", ""))) + sliced = filtered[:limit] + return { + "total_matches": len(filtered), + "returned": len(sliced), + "items": sliced, + } + + +def neighbors( + graph: Dict[str, Any], id: str, limit: int = DEFAULT_NEIGHBOR_LIMIT +) -> Dict[str, Any]: + node_id = str(id) + limit = int(limit) + nodes = nodes_by_id(graph) + if node_id not in nodes: + raise ValueError(f"node '{node_id}' not found") + + edges = graph.get("edges", []) + outbound_edges = [e for e in edges if e.get("from") == node_id] + inbound_edges = [e for e in edges if e.get("to") == node_id] + + def describe(edge_list: List[Dict[str, Any]], direction: str) -> List[Dict[str, Any]]: + # neighbor id is the "other end" relative to node_id + other_key = "to" if direction == "outbound" else "from" + result: List[Dict[str, Any]] = [] + for edge in edge_list: + item_id = str(edge.get(other_key)) + node = nodes.get(item_id) + result.append( + { + "id": node.get("id") if node else item_id, + "title": node.get("title") if node else "", + "path": node.get("path") if node else "", + "direction": direction, + "type": edge.get("type", "related"), + "anchor_text": edge.get("anchor_text", ""), + "annotation": edge.get("annotation", ""), + } + ) + return result[:limit] + + return { + "node": nodes[node_id], + "inbound_count": len(inbound_edges), + "outbound_count": len(outbound_edges), + "inbound": describe(inbound_edges, "inbound"), + "outbound": describe(outbound_edges, "outbound"), + } + + +def node_path(graph: Dict[str, Any], id: str) -> str: + """Vault-relative path of the note behind a node id.""" + node = nodes_by_id(graph).get(str(id)) + if not node: + raise ValueError(f"node '{id}' not found") + return str(node.get("path")) diff --git a/kb_framework/query_tools.py b/kb_framework/query_tools.py new file mode 100644 index 0000000..cb6dfb3 --- /dev/null +++ b/kb_framework/query_tools.py @@ -0,0 +1,164 @@ +"""The read-only MCP tool contract: definitions and dispatch. + +Both front ends (the stdio server and the hosted connector) advertise exactly +these five tools with exactly these schemas, and `call` produces the result +dict (`content` + `structuredContent`) each returns. Keeping the contract here +means an agent skill written against one front end stays valid on the other. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict, List + +from kb_framework import instructions, query, vault + +READ_ONLY_TOOL_NAMES = ( + "read_instructions", + "graph_summary", + "graph_search_nodes", + "graph_neighbors", + "read_note", +) + + +def definitions() -> List[Dict[str, Any]]: + return [ + { + "name": "read_instructions", + "description": ( + "Read framework AGENTS.md and AGENT_QUERY_GUIDE.md before working " + "with the vault. These operating documents live outside an " + "external vault and do not require a generated graph." + ), + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + }, + { + "name": "graph_summary", + "description": ( + "Orientation for a cold start: node and edge counts, the repo " + "scopes, the most-linked notes, and the tag vocabulary as " + "`by_tag` (most frequent first) plus `tag_count`, the number of " + "distinct tags. Read `by_tag` before filtering " + "`graph_search_nodes` by `tag` — that filter matches a tag " + "verbatim, so a guessed spelling returns nothing." + ), + "inputSchema": { + "type": "object", + "properties": { + "topLinked": { + "type": "integer", + "minimum": 1, + "maximum": 50, + "description": "Most-linked notes to return (default 10).", + }, + "topTags": { + "type": "integer", + "minimum": 1, + "maximum": 500, + "description": "Tags to return, most frequent first (default 50).", + }, + }, + "additionalProperties": False, + }, + }, + { + "name": "graph_search_nodes", + "description": "Search graph nodes by metadata and free text.", + "inputSchema": { + "type": "object", + "properties": { + "repo": {"type": "string"}, + "type": {"type": "string"}, + "status": {"type": "string"}, + "tag": {"type": "string"}, + "query": {"type": "string"}, + "limit": {"type": "integer", "minimum": 1, "maximum": 200}, + }, + }, + }, + { + "name": "graph_neighbors", + "description": ( + "Show inbound and outbound links for a node ID. Each neighbor " + "includes the edge `type` (relationship, e.g. supersedes, " + "depends-on), `anchor_text` (the link's display text), and " + "`annotation` (what to expect in that note)." + ), + "inputSchema": { + "type": "object", + "required": ["id"], + "properties": { + "id": {"type": "string"}, + "limit": {"type": "integer", "minimum": 1, "maximum": 200}, + }, + }, + }, + { + "name": "read_note", + "description": ( + "Read note content by graph node ID or vault-relative path. " + "Provide exactly one of `id` or `path`. " + "If both or neither are provided, the call returns an error." + ), + "inputSchema": { + "type": "object", + "properties": { + "id": {"type": "string", "description": "Graph node ID (e.g. 'AGENTS')."}, + "path": {"type": "string", "description": "Vault-relative file path (e.g. 'README.md')."}, + }, + "additionalProperties": False, + }, + }, + ] + + +def text_result(payload: Dict[str, Any]) -> Dict[str, Any]: + """Wrap a payload as an MCP tool result: pretty JSON text plus structuredContent.""" + return { + "content": [{"type": "text", "text": json.dumps(payload, indent=2)}], + "structuredContent": payload, + } + + +def read_note_path(graph: Dict[str, Any], args: Dict[str, Any]) -> str: + """Resolve `read_note` arguments to the vault-relative path to read.""" + has_id = bool(args.get("id")) + has_path = bool(args.get("path")) + if has_id and has_path: + raise ValueError("provide exactly one of 'id' or 'path', not both") + if not has_id and not has_path: + raise ValueError("exactly one of 'id' or 'path' is required") + if has_id: + return query.node_path(graph, str(args["id"])) + return str(args["path"]) + + +def call( + name: str, args: Dict[str, Any], *, graph: Dict[str, Any], root: Path +) -> Dict[str, Any]: + """Run one read-only tool. Raises ValueError for unknown tools or bad input.""" + if name == "read_instructions": + return text_result(instructions.read_instructions()) + if name == "graph_summary": + return text_result( + query.summary( + graph, + topLinked=args.get("topLinked", query.DEFAULT_TOP_LINKED), + topTags=args.get("topTags", query.DEFAULT_TOP_TAGS), + ) + ) + if name == "graph_search_nodes": + return text_result(query.search_nodes(graph, args)) + if name == "graph_neighbors": + return text_result( + query.neighbors(graph, str(args["id"]), args.get("limit", query.DEFAULT_NEIGHBOR_LIMIT)) + ) + if name == "read_note": + return text_result(vault.read_note_payload(root, read_note_path(graph, args))) + raise ValueError(f"unknown tool '{name}'") diff --git a/kb_framework/vault.py b/kb_framework/vault.py new file mode 100644 index 0000000..cb30409 --- /dev/null +++ b/kb_framework/vault.py @@ -0,0 +1,57 @@ +"""Filesystem access to a vault, confined to its root. + +`resolve_in_root` is the single place that turns a vault-relative path into an +absolute one and refuses empty, absolute or escaping values, so no tool can +reach outside the vault. `read_note_payload` returns the `read_note` payload +shape shared by the stdio server and the connector: UTF-8 text inline, or a +binary descriptor (size, sha256, hint) for captured originals under `raw/`. +""" + +from __future__ import annotations + +import hashlib +from pathlib import Path +from typing import Any, Dict + + +def resolve_in_root(root: Path, raw_path: Any, *, label: str = "path") -> Path: + """Turn a vault-relative path into an absolute path inside `root`.""" + value = str(raw_path or "").strip() + if not value: + raise ValueError(f"'{label}' is required") + candidate = Path(value) + if candidate.is_absolute(): + raise ValueError(f"'{label}' must be vault-relative, got absolute '{value}'") + root = Path(root).resolve() + target = (root / candidate).resolve() + if target != root and not target.is_relative_to(root): + raise ValueError(f"'{label}' resolves outside repository root: '{value}'") + return target + + +def relative_to_root(root: Path, target: Path) -> str: + return target.relative_to(Path(root).resolve()).as_posix() + + +def read_note_payload(root: Path, raw_path: Any) -> Dict[str, Any]: + """The `read_note` payload for a vault-relative path.""" + target = resolve_in_root(root, raw_path) + if not target.exists(): + raise FileNotFoundError(f"note not found at {target}") + relative = relative_to_root(root, target) + data = target.read_bytes() + try: + text = data.decode("utf-8") + except UnicodeDecodeError: + return { + "path": relative, + "binary": True, + "bytes": len(data), + "sha256": hashlib.sha256(data).hexdigest(), + "hint": ( + f"{target} is not UTF-8 text (e.g. a PDF or image captured under " + "raw/). Open it with a file-reading tool that handles binaries; " + "this server returns text only." + ), + } + return {"path": relative, "binary": False, "content": text} diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..4ed6e1b --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "kb-framework" +version = "0.2.0" +description = "Knowledge-base framework core: in-memory graph, query tools, vault access policy" +readme = "README.md" +requires-python = ">=3.12" +license = { text = "MIT" } +dependencies = [] + +[tool.hatch.build.targets.wheel] +packages = ["kb_framework"] + +# Bundle the agent instructions so `read_instructions` works from an installed +# wheel (e.g. inside the hosted connector image), not only from a checkout. +[tool.hatch.build.targets.wheel.force-include] +"AGENTS.md" = "kb_framework/_docs/AGENTS.md" +"AGENT_QUERY_GUIDE.md" = "kb_framework/_docs/AGENT_QUERY_GUIDE.md" + +[tool.hatch.build.targets.sdist] +include = ["kb_framework", "AGENTS.md", "AGENT_QUERY_GUIDE.md", "README.md", "pyproject.toml"] diff --git a/scripts/generate-graph.py b/scripts/generate-graph.py index e278373..7406ffe 100644 --- a/scripts/generate-graph.py +++ b/scripts/generate-graph.py @@ -1,439 +1,51 @@ #!/usr/bin/env python3 -"""Build `index/graph.json` from the vault's Markdown, and validate it. +"""Validate the vault graph, and optionally write it to `index/graph.json`. -The graph is the derived navigation layer agents query instead of walking the -tree: one node per note (frontmatter plus link counts) and one edge per typed -markdown link. `--validate` is the gate that keeps it trustworthy — it fails on -links that resolve to nothing, on edge relationship types or note `type` values -outside their controlled vocabularies, and on legacy `[[wiki links]]`. +Thin CLI over `kb_framework.graph`: the graph is built in memory from the +vault's Markdown. `--validate` is the gate the pre-commit hook applies — it +fails on links that resolve to nothing, on edge relationship types or note +`type` values outside their controlled vocabularies, and on legacy +`[[wiki links]]`. `--write` is for operators who want the file; nothing else +writes it, and the MCP servers never read it. """ import argparse import json -import re import sys -from collections import defaultdict -from dataclasses import dataclass -from datetime import datetime, timezone from pathlib import Path -from typing import Dict, List, Tuple +from typing import Dict, List from vault_root import parse_vault_args - -# Obsidian-style wiki link: [[target#anchor|alias]] -WIKI_LINK_RE = re.compile(r"\[\[([^\]|#]+)(?:#[^\]|]+)?(?:\|([^\]]+))?\]\]") -# Standard markdown link: [display](target); (? edge anchor_text - kind: str # "wiki_link" | "markdown_link" - rel_type: str # semantic relationship type (default "related") - annotation: str # per-edge "what to expect there" note - lineno: int # 1-based line number in the note body - raw: str # the raw link text as written (for the inventory script) - is_external: bool # markdown link whose target is a URL/anchor -> no edge - - -@dataclass -class Note: - path: Path - rel_path: str - stem: str - meta: Dict[str, object] - title: str - note_id: str - links: List[LinkRef] - - -@dataclass(frozen=True) -class UnresolvedLink: - from_id: str - from_path: str - target: str - - -@dataclass(frozen=True) -class WikiLinkUse: - from_id: str - from_path: str - target: str - lineno: int - raw: str - - -def parse_frontmatter(text: str) -> Tuple[Dict[str, object], str]: - if not text.startswith("---\n"): - return {}, text - - end = text.find("\n---\n", 4) - if end == -1: - return {}, text - - fm_block = text[4:end] - body = text[end + 5 :] - meta: Dict[str, object] = {} - - for raw in fm_block.splitlines(): - line = raw.strip() - if not line or line.startswith("#") or ":" not in line: - continue - key, value = line.split(":", 1) - key = key.strip() - value = value.strip() - if value.startswith("[") and value.endswith("]"): - inner = value[1:-1].strip() - if inner: - meta[key] = [v.strip().strip('"').strip("'") for v in inner.split(",")] - else: - meta[key] = [] - else: - meta[key] = value.strip('"').strip("'") - - return meta, body - - -def derive_title(meta: Dict[str, object], body: str, stem: str) -> str: - if isinstance(meta.get("title"), str) and meta["title"].strip(): - return str(meta["title"]).strip() - for line in body.splitlines(): - if line.startswith("# "): - return line[2:].strip() - return stem.replace("-", " ").strip().title() - - -def mask_code(body: str) -> str: - """Blank out code so links inside code samples are ignored, preserving line - boundaries so line-based parsing (gloss fallback, line numbers) stays aligned.""" - body = FENCED_CODE_RE.sub(lambda m: "\n" * m.group(0).count("\n"), body) - body = INLINE_CODE_RE.sub(lambda m: " " * len(m.group(0)), body) - return body - - -def scan_brace(text: str, start: int) -> Tuple[object, int]: - """Given text[start] == '{', scan to the matching '}' respecting JSON string - quoting, and json.loads the block. Returns (obj_or_None, end_index_after_brace). - On an unterminated or malformed brace, returns (None, start) so the caller - treats it as no brace.""" - depth = 0 - in_str = False - esc = False - i = start - while i < len(text): - c = text[i] - if in_str: - if esc: - esc = False - elif c == "\\": - esc = True - elif c == '"': - in_str = False - elif c == '"': - in_str = True - elif c == "{": - depth += 1 - elif c == "}": - depth -= 1 - if depth == 0: - raw = text[start : i + 1] - try: - return json.loads(raw), i + 1 - except (json.JSONDecodeError, ValueError): - return None, start - i += 1 - return None, start - - -def clean_md_target(raw: str) -> str: - """Normalize a markdown link target: drop a trailing ``"title"`` and <> wrappers.""" - raw = raw.strip() - if raw.startswith("<") and raw.endswith(">"): - raw = raw[1:-1].strip() - # `[text](target "title")` -> target - for sep in (' "', " '"): - if sep in raw: - raw = raw.split(sep, 1)[0].strip() - return raw - - -def parse_links(body: str) -> List[LinkRef]: - masked = mask_code(body) - refs: List[LinkRef] = [] - - for lineno, line in enumerate(masked.splitlines(), start=1): - # Collect wiki + markdown link matches on this line, in order. - raw_matches: List[Tuple[int, int, str, str, str, str]] = [] - for m in WIKI_LINK_RE.finditer(line): - target = m.group(1).strip() - alias = (m.group(2) or target).strip() - if target: - raw_matches.append( - (m.start(), m.end(), "wiki_link", target, alias, m.group(0)) - ) - for m in MD_LINK_RE.finditer(line): - target = clean_md_target(m.group(2)) - display = m.group(1).strip() - if target: - raw_matches.append( - ( - m.start(), - m.end(), - "markdown_link", - target, - display or target, - m.group(0), - ) - ) - raw_matches.sort(key=lambda x: x[0]) - - line_refs: List[Tuple[LinkRef, int]] = [] # (ref, end-index-after-brace) - for start, end, kind, target, anchor, raw in raw_matches: - rel_type, annotation = "related", "" - brace_end = end - if end < len(line) and line[end] == "{": - obj, be = scan_brace(line, end) - if isinstance(obj, dict): - brace_end = be - t = obj.get("type") - if isinstance(t, str) and t.strip(): - rel_type = t.strip() - info = obj.get("link-info") - if isinstance(info, str) and info.strip(): - annotation = info.strip() - is_external = kind == "markdown_link" and bool( - EXTERNAL_TARGET_RE.match(target) - ) - ref = LinkRef( - target=target, - anchor=anchor, - kind=kind, - rel_type=rel_type, - annotation=annotation, - lineno=lineno, - raw=raw, - is_external=is_external, - ) - line_refs.append((ref, brace_end)) - - # Annotation fallback: a trailing "— gloss" applies only when the link is - # the sole link on its line (so we don't mis-assign a shared gloss). - if len(line_refs) == 1: - ref, brace_end = line_refs[0] - if not ref.annotation: - gloss = GLOSS_RE.match(line[brace_end:]) - if gloss: - ref.annotation = gloss.group(1).strip() - - refs.extend(ref for ref, _ in line_refs) - - return refs - - -def norm(s: str) -> str: - return s.strip().lower() - - -def build_notes(root: Path) -> List[Note]: - notes: List[Note] = [] - for p in sorted(root.rglob("*.md")): - rel = p.relative_to(root) - if rel.parts and rel.parts[0] in {"index", "scripts", ".obsidian"}: - continue - # `raw/` holds untouched ingested originals (external READMEs, specs, …), - # not vault notes — skip them so their file-relative markdown links are not - # treated as (broken) graph edges. - if "raw" in rel.parts: - continue - - text = p.read_text(encoding="utf-8") - meta, body = parse_frontmatter(text) - title = derive_title(meta, body, p.stem) - note_id = str(meta.get("id") or rel.with_suffix("")).strip() - links = parse_links(body) - # parse_links numbers lines within `body`; shift to absolute file line - # numbers by the number of lines the frontmatter block occupied. - line_offset = text.count("\n", 0, len(text) - len(body)) - if line_offset: - for ref in links: - ref.lineno += line_offset - - notes.append( - Note( - path=p, - rel_path=rel.as_posix(), - stem=p.stem, - meta=meta, - title=title, - note_id=note_id, - links=links, - ) - ) - return notes - - -def build_alias_map(notes: List[Note]) -> Dict[str, str]: - alias_map: Dict[str, str] = {} - stem_buckets: Dict[str, List[str]] = defaultdict(list) - - for n in notes: - stem_buckets[norm(n.stem)].append(n.note_id) - - for n in notes: - alias_map[norm(n.note_id)] = n.note_id - alias_map[norm(n.rel_path)] = n.note_id - alias_map[norm(Path(n.rel_path).with_suffix("").as_posix())] = n.note_id - alias_map[norm(n.title)] = n.note_id - - for stem, ids in stem_buckets.items(): - if len(ids) == 1: - alias_map[stem] = ids[0] - - return alias_map - - -def generate_graph( - root: Path, existing_generated_at: str = "" -) -> Tuple[Dict[str, object], List[UnresolvedLink], List[WikiLinkUse]]: - notes = build_notes(root) - alias_map = build_alias_map(notes) - - edges_set = set() - outbound: Dict[str, List[str]] = defaultdict(list) - inbound_count: Dict[str, int] = defaultdict(int) - unresolved_set = set() - wiki_uses_set = set() - - for n in notes: - for ref in n.links: - # Legacy [[wiki links]] are still parsed (so we can flag them) but are - # no longer an accepted link form — --validate rejects any that remain. - if ref.kind == "wiki_link": - wiki_uses_set.add( - WikiLinkUse( - from_id=n.note_id, - from_path=n.rel_path, - target=ref.target, - lineno=ref.lineno, - raw=ref.raw, - ) - ) - if ref.is_external: - continue - target = alias_map.get(norm(ref.target)) - if not target: - unresolved_set.add( - UnresolvedLink( - from_id=n.note_id, - from_path=n.rel_path, - target=ref.target, - ) - ) - continue - edge = (n.note_id, target, ref.kind, ref.rel_type, ref.annotation, ref.anchor) - if edge in edges_set: - continue - edges_set.add(edge) - outbound[n.note_id].append(target) - inbound_count[target] += 1 - - nodes = [] - for n in sorted(notes, key=lambda x: x.note_id): - tags = n.meta.get("tags", []) - if not isinstance(tags, list): - tags = [] - nodes.append( - { - "id": n.note_id, - "title": n.title, - "path": n.rel_path, - "repo": n.meta.get("repo", "global"), - "type": n.meta.get("type", "note"), - "status": n.meta.get("status", "active"), - "tags": tags, - "last_reviewed": n.meta.get("last_reviewed", ""), - "outbound_links": sorted(set(outbound.get(n.note_id, []))), - "inbound_count": inbound_count.get(n.note_id, 0), - } - ) - - edges = [ - { - "from": src, - "to": dst, - "kind": kind, - "type": rel_type, - "anchor_text": anchor, - "annotation": annotation, - "weight": 1.0, - } - for src, dst, kind, rel_type, annotation, anchor in sorted(edges_set) - ] - - graph = { - "version": 1, - "generated_at": existing_generated_at - if existing_generated_at - else datetime.now(timezone.utc).isoformat(), - "nodes": nodes, - "edges": edges, - } - unresolved = sorted(unresolved_set, key=lambda x: (x.from_path, x.target)) - wiki_uses = sorted(wiki_uses_set, key=lambda x: (x.from_path, x.lineno, x.raw)) - return graph, unresolved, wiki_uses +from kb_framework.graph import ( # noqa: F401 (re-exported for list-wiki-links.py and tests) + NOTE_TYPES, + RELATION_TYPES, + GraphBuild, + LinkRef, + Note, + UnresolvedLink, + WikiLinkUse, + build_alias_map, + build_graph, + build_notes, + find_invalid_node_types, + find_invalid_types, + generate_graph, + norm, + parse_frontmatter, + parse_links, +) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( - description="Generate and validate Obsidian graph JSON" + description="Validate the vault graph; optionally write index/graph.json" ) parser.add_argument( "--validate", action="store_true", help="validate the graph: unresolved targets, edge relationship types, " - "note types, " - "and legacy [[wiki links]] (all rejected)", + "note types, and legacy [[wiki links]] (all rejected)", ) parser.add_argument( "--write", @@ -447,7 +59,6 @@ def print_unresolved(unresolved: List[UnresolvedLink]) -> None: if not unresolved: print("Validation passed: no unresolved links") return - print(f"Validation failed: {len(unresolved)} unresolved link(s)") for link in unresolved: print(f"- {link.from_path}: [[{link.target}]]") @@ -457,92 +68,69 @@ def print_wiki_links(wiki_uses: List[WikiLinkUse]) -> None: if not wiki_uses: print("Validation passed: no legacy [[wiki links]]") return - print(f"Validation failed: {len(wiki_uses)} legacy [[wiki link(s)]]") print(" use typed markdown links; find/convert them with scripts/list-wiki-links.py") for w in wiki_uses: print(f"- {w.from_path}:{w.lineno}: {w.raw}") -def find_invalid_types(graph: Dict[str, object]) -> List[Dict[str, object]]: - return [ - e - for e in graph.get("edges", []) # type: ignore[union-attr] - if e.get("type") not in RELATION_TYPES - ] - - def print_invalid_types(invalid: List[Dict[str, object]]) -> None: if not invalid: print("Validation passed: all edge types are in the controlled vocabulary") return - print(f"Validation failed: {len(invalid)} edge(s) with unknown relationship type") print(f" allowed: {', '.join(sorted(RELATION_TYPES))}") for e in invalid: print(f"- {e.get('from')} -> {e.get('to')}: type={e.get('type')!r}") -def find_invalid_node_types(graph: Dict[str, object]) -> List[Dict[str, object]]: - return [ - n - for n in graph.get("nodes", []) # type: ignore[union-attr] - if n.get("type") not in NOTE_TYPES - ] - - def print_invalid_node_types(invalid: List[Dict[str, object]]) -> None: if not invalid: print("Validation passed: all node types are in the controlled vocabulary") return - print(f"Validation failed: {len(invalid)} note(s) with unknown type") print(f" allowed: {', '.join(sorted(NOTE_TYPES))}") for n in invalid: print(f"- {n.get('id')} ({n.get('path')}): type={n.get('type')!r}") +def _existing_generated_at(output: Path, candidate: GraphBuild) -> str: + """Keep the old timestamp when the written graph would be unchanged.""" + if not output.exists(): + return "" + try: + existing = json.loads(output.read_text(encoding="utf-8")) + except (json.JSONDecodeError, OSError, TypeError, ValueError): + return "" + if ( + candidate.graph["nodes"] == existing.get("nodes", []) + and candidate.graph["edges"] == existing.get("edges", []) + ): + return str(existing.get("generated_at", "")) + return "" + + def main() -> None: args = parse_args() root = args.vault_root output = root / "index" / "graph.json" should_write = args.write or not args.validate + build = build_graph(root) if should_write: + stamp = _existing_generated_at(output, build) + if stamp: + build = build_graph(root, existing_generated_at=stamp) output.parent.mkdir(parents=True, exist_ok=True) - - existing_generated_at = "" - if output.exists(): - try: - existing = json.loads(output.read_text(encoding="utf-8")) - existing_nodes = existing.get("nodes", []) - existing_edges = existing.get("edges", []) - candidate_graph, unresolved, wiki_uses = generate_graph(root) - if ( - candidate_graph["nodes"] == existing_nodes - and candidate_graph["edges"] == existing_edges - ): - existing_generated_at = str(existing.get("generated_at", "")) - graph, unresolved, wiki_uses = generate_graph( - root, existing_generated_at=existing_generated_at - ) - except (json.JSONDecodeError, OSError, TypeError, ValueError): - graph, unresolved, wiki_uses = generate_graph(root) - else: - graph, unresolved, wiki_uses = generate_graph(root) - - if should_write: - output.write_text(json.dumps(graph, indent=2) + "\n", encoding="utf-8") + output.write_text(json.dumps(build.graph, indent=2) + "\n", encoding="utf-8") print(f"Wrote {output}") if args.validate: - print_unresolved(unresolved) - invalid = find_invalid_types(graph) - print_invalid_types(invalid) - invalid_nodes = find_invalid_node_types(graph) - print_invalid_node_types(invalid_nodes) - print_wiki_links(wiki_uses) - if unresolved or invalid or invalid_nodes or wiki_uses: + print_unresolved(build.unresolved) + print_invalid_types(build.invalid_edge_types) + print_invalid_node_types(build.invalid_node_types) + print_wiki_links(build.wiki_uses) + if not build.ok: sys.exit(1) diff --git a/scripts/mcp-server.py b/scripts/mcp-server.py index b920bf6..c5d46bb 100755 --- a/scripts/mcp-server.py +++ b/scripts/mcp-server.py @@ -4,7 +4,9 @@ This server is the single interface an agent needs to work the vault over stdio with JSON-RPC 2.0. It provides framework operating instructions (read_instructions), and exposes graph and note *querying* (graph_summary, -graph_search_nodes, graph_neighbors, read_note, refresh_graph), *writing* +graph_search_nodes, graph_neighbors, read_note — all served by the shared +`kb_framework` package over a graph built in memory from the vault's Markdown +and cached until the next write; refresh_graph rebuilds it), *writing* (write_note for text, capture_raw for verbatim bytes from a local file or an http(s) download) and *publishing* (edit_note, git_commit, git_push), so an agent without shell or filesystem tools can ingest a resource end to end. @@ -27,21 +29,35 @@ import subprocess import sys import urllib.error import urllib.request -from collections import Counter from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, List -from vault_root import FRAMEWORK_ROOT, parse_vault_args +from vault_root import FRAMEWORK_ROOT, parse_vault_args # noqa: E402 + +from kb_framework import graph as kb_graph # noqa: E402 +from kb_framework import query_tools # noqa: E402 +from kb_framework import vault as kb_vault # noqa: E402 SERVER_NAME = "knowledge-base-mcp" SERVER_VERSION = "0.1.0" +@dataclass @dataclass class AppState: + """The selected vault plus the graph built from it, cached until a write.""" + root: Path - graph_path: Path + graph_cache: Dict[str, Any] | None = None + + def graph(self) -> Dict[str, Any]: + if self.graph_cache is None: + self.graph_cache = kb_graph.build_graph(self.root).graph + return self.graph_cache + + def invalidate(self) -> None: + self.graph_cache = None class _StdinReader: @@ -140,7 +156,6 @@ def _write_message_ndjson(stdout: Any, payload: Dict[str, Any]) -> None: stdout.buffer.flush() - def _response(message_id: Any, result: Dict[str, Any]) -> Dict[str, Any]: return {"jsonrpc": "2.0", "id": message_id, "result": result} @@ -153,47 +168,9 @@ def _error(message_id: Any, code: int, message: str) -> Dict[str, Any]: } -def _load_graph(state: AppState) -> Dict[str, Any]: - if not state.graph_path.exists(): - raise FileNotFoundError( - f"graph not found at {state.graph_path}. Run scripts/generate-graph.py first." - ) - return json.loads(state.graph_path.read_text(encoding="utf-8")) - - -def _nodes_by_id(graph: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: - nodes = graph.get("nodes", []) - return {str(node.get("id")): node for node in nodes if isinstance(node, dict)} - - -def _match_filters(node: Dict[str, Any], args: Dict[str, Any]) -> bool: - repo = args.get("repo") - note_type = args.get("type") - status = args.get("status") - tag = args.get("tag") - query = str(args.get("query", "")).strip().lower() - - if repo and node.get("repo") != repo: - return False - if note_type and node.get("type") != note_type: - return False - if status and node.get("status") != status: - return False - if tag and tag not in (node.get("tags") or []): - return False - if query: - haystacks = [ - str(node.get("id", "")).lower(), - str(node.get("title", "")).lower(), - str(node.get("path", "")).lower(), - " ".join(str(t).lower() for t in (node.get("tags") or [])), - ] - if not any(query in h for h in haystacks): - return False - return True - - _WILDCARD_PATHS = {".", "./", "-A", "--all", "*", "-a", ":/"} +# Reserved for operator-written graph dumps (generate-graph.py --write); never a note. +_GENERATED_GRAPH_PATH = "index/graph.json" # Ceiling for a single capture_raw call; papers and slide decks fit, a video # download that would bloat the vault does not. @@ -202,31 +179,15 @@ _DOWNLOAD_TIMEOUT_SECONDS = 60 # graph_summary defaults: enough most-linked notes to spot the hubs, and enough # tags to cover the vocabulary an agent actually filters on. -_DEFAULT_TOP_LINKED = 10 -_DEFAULT_TOP_TAGS = 50 def _resolve_in_root(state: AppState, raw_path: Any, *, label: str = "path") -> Path: - """Turn a vault-relative path into an absolute path inside the repo root. - - Rejects empty values, absolute paths and anything escaping the root, so a - tool call can never touch a file outside the vault. - """ - value = str(raw_path or "").strip() - if not value: - raise ValueError(f"'{label}' is required") - candidate = Path(value) - if candidate.is_absolute(): - raise ValueError(f"'{label}' must be vault-relative, got absolute '{value}'") - root = state.root.resolve() - target = (root / candidate).resolve() - if target != root and not target.is_relative_to(root): - raise ValueError(f"'{label}' resolves outside repository root: '{value}'") - return target + """Vault-relative path -> absolute path inside the repo root (never outside).""" + return kb_vault.resolve_in_root(state.root, raw_path, label=label) def _relative_to_root(state: AppState, target: Path) -> str: - return target.relative_to(state.root.resolve()).as_posix() + return kb_vault.relative_to_root(state.root, target) def _run_git(state: AppState, args: List[str]) -> subprocess.CompletedProcess: @@ -258,99 +219,15 @@ def _git_result(payload: Dict[str, Any], exit_code: int) -> Dict[str, Any]: def _tool_definitions() -> List[Dict[str, Any]]: return [ - { - "name": "read_instructions", - "description": ( - "Read framework AGENTS.md and AGENT_QUERY_GUIDE.md before working " - "with the vault. These operating documents live outside an " - "external vault and do not require a generated graph." - ), - "inputSchema": { - "type": "object", - "properties": {}, - "additionalProperties": False, - }, - }, - { - "name": "graph_summary", - "description": ( - "Orientation for a cold start: node and edge counts, the repo " - "scopes, the most-linked notes, and the tag vocabulary as " - "`by_tag` (most frequent first) plus `tag_count`, the number of " - "distinct tags. Read `by_tag` before filtering " - "`graph_search_nodes` by `tag` \u2014 that filter matches a tag " - "verbatim, so a guessed spelling returns nothing." - ), - "inputSchema": { - "type": "object", - "properties": { - "topLinked": { - "type": "integer", - "minimum": 1, - "maximum": 50, - "description": "Most-linked notes to return (default 10).", - }, - "topTags": { - "type": "integer", - "minimum": 1, - "maximum": 500, - "description": "Tags to return, most frequent first (default 50).", - }, - }, - "additionalProperties": False, - }, - }, - { - "name": "graph_search_nodes", - "description": "Search graph nodes by metadata and free text.", - "inputSchema": { - "type": "object", - "properties": { - "repo": {"type": "string"}, - "type": {"type": "string"}, - "status": {"type": "string"}, - "tag": {"type": "string"}, - "query": {"type": "string"}, - "limit": {"type": "integer", "minimum": 1, "maximum": 200}, - }, - }, - }, - { - "name": "graph_neighbors", - "description": ( - "Show inbound and outbound links for a node ID. Each neighbor " - "includes the edge `type` (relationship, e.g. supersedes, " - "depends-on), `anchor_text` (the link's display text), and " - "`annotation` (what to expect in that note)." - ), - "inputSchema": { - "type": "object", - "required": ["id"], - "properties": { - "id": {"type": "string"}, - "limit": {"type": "integer", "minimum": 1, "maximum": 200}, - }, - }, - }, - { - "name": "read_note", - "description": ( - "Read note content by graph node ID or vault-relative path. " - "Provide exactly one of `id` or `path`. " - "If both or neither are provided, the call returns an error." - ), - "inputSchema": { - "type": "object", - "properties": { - "id": {"type": "string", "description": "Graph node ID (e.g. 'AGENTS')."}, - "path": {"type": "string", "description": "Vault-relative file path (e.g. 'README.md')."}, - }, - "additionalProperties": False, - }, - }, + *query_tools.definitions(), { "name": "refresh_graph", - "description": "Regenerate graph.json using scripts/generate-graph.py.", + "description": ( + "Rebuild the in-memory graph from the vault's Markdown. With " + "validate=true also run the commit gate (unresolved links, " + "unknown edge/note types, legacy wiki links) and return an error " + "result listing the problems." + ), "inputSchema": { "type": "object", "properties": { @@ -496,187 +373,20 @@ def _text_result(text: str, structured: Dict[str, Any] | None = None) -> Dict[st return result -def _tool_read_instructions() -> Dict[str, Any]: - documents = [ - {"path": name, "content": (FRAMEWORK_ROOT / name).read_text(encoding="utf-8")} - for name in ("AGENTS.md", "AGENT_QUERY_GUIDE.md") - ] - payload = {"framework_root": str(FRAMEWORK_ROOT), "documents": documents} - return _text_result(json.dumps(payload, indent=2), structured=payload) - - -def _tool_graph_summary(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: - graph = _load_graph(state) - nodes = graph.get("nodes", []) - edges = graph.get("edges", []) - top_n = int(args.get("topLinked", _DEFAULT_TOP_LINKED)) - top_tags = int(args.get("topTags", _DEFAULT_TOP_TAGS)) - - by_repo: Counter[str] = Counter(str(node.get("repo", "unknown")) for node in nodes) - tag_counts: Counter[str] = Counter( - str(tag) for node in nodes for tag in (node.get("tags") or []) - ) - # Frequent tags first, alphabetical within a tie, so a truncated list is - # both the most useful slice and stable across calls. - ranked_tags = sorted(tag_counts.items(), key=lambda item: (-item[1], item[0])) - - sorted_nodes = sorted( - nodes, - key=lambda n: int(n.get("inbound_count", 0)), - reverse=True, - )[:top_n] - payload = { - "version": graph.get("version"), - "generated_at": graph.get("generated_at"), - "node_count": len(nodes), - "edge_count": len(edges), - "by_repo": dict(by_repo), - "tag_count": len(tag_counts), - "by_tag": [ - {"tag": tag, "count": count} for tag, count in ranked_tags[:top_tags] - ], - "top_linked": [ - { - "id": node.get("id"), - "title": node.get("title"), - "path": node.get("path"), - "inbound_count": node.get("inbound_count", 0), - } - for node in sorted_nodes - ], - } - return _text_result(json.dumps(payload, indent=2), structured=payload) - - -def _tool_graph_search_nodes(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: - graph = _load_graph(state) - nodes = graph.get("nodes", []) - limit = int(args.get("limit", 30)) - filtered = [node for node in nodes if _match_filters(node, args)] - filtered.sort(key=lambda n: str(n.get("id", ""))) - sliced = filtered[:limit] - payload = { - "total_matches": len(filtered), - "returned": len(sliced), - "items": sliced, - } - return _text_result(json.dumps(payload, indent=2), structured=payload) - - -def _tool_graph_neighbors(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: - graph = _load_graph(state) - node_id = str(args["id"]) - limit = int(args.get("limit", 50)) - nodes = _nodes_by_id(graph) - if node_id not in nodes: - raise ValueError(f"node '{node_id}' not found") - - edges = graph.get("edges", []) - outbound_edges = [e for e in edges if e.get("from") == node_id] - inbound_edges = [e for e in edges if e.get("to") == node_id] - - def describe(edge_list: List[Dict[str, Any]], direction: str) -> List[Dict[str, Any]]: - # neighbor id is the "other end" relative to node_id - other_key = "to" if direction == "outbound" else "from" - result: List[Dict[str, Any]] = [] - for edge in edge_list: - item_id = str(edge.get(other_key)) - node = nodes.get(item_id) - entry = { - "id": node.get("id") if node else item_id, - "title": node.get("title") if node else "", - "path": node.get("path") if node else "", - "direction": direction, - "type": edge.get("type", "related"), - "anchor_text": edge.get("anchor_text", ""), - "annotation": edge.get("annotation", ""), - } - result.append(entry) - return result[:limit] - - payload = { - "node": nodes[node_id], - "inbound_count": len(inbound_edges), - "outbound_count": len(outbound_edges), - "inbound": describe(inbound_edges, "inbound"), - "outbound": describe(outbound_edges, "outbound"), - } - return _text_result(json.dumps(payload, indent=2), structured=payload) - - -def _tool_read_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: - has_id = bool(args.get("id")) - has_path = bool(args.get("path")) - - if has_id and has_path: - raise ValueError("provide exactly one of 'id' or 'path', not both") - if not has_id and not has_path: - raise ValueError("exactly one of 'id' or 'path' is required") - - path_value = args.get("path") - - if has_id: - nodes = _nodes_by_id(_load_graph(state)) - node = nodes.get(str(args["id"])) - if not node: - raise ValueError(f"node '{args['id']}' not found") - path_value = node.get("path") - - target = (state.root / str(path_value)).resolve() - if not target.is_relative_to(state.root.resolve()): - raise ValueError("note path resolves outside repository root") - if not target.exists(): - raise FileNotFoundError(f"note not found at {target}") - - relative = str(target.relative_to(state.root).as_posix()) - data = target.read_bytes() - try: - text = data.decode("utf-8") - except UnicodeDecodeError: - payload = { - "path": relative, - "binary": True, - "bytes": len(data), - "sha256": hashlib.sha256(data).hexdigest(), - "hint": ( - f"{target} is not UTF-8 text (e.g. a PDF or image captured under " - "raw/). Open it with a file-reading tool that handles binaries; " - "this server returns text only." - ), - } - return _text_result(json.dumps(payload, indent=2), structured=payload) - - payload = { - "path": relative, - "binary": False, - "content": text, - } - return _text_result(json.dumps(payload, indent=2), structured=payload) - - def _tool_refresh_graph(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: - command = [ - sys.executable, str(FRAMEWORK_ROOT / "scripts/generate-graph.py"), - "--vault-root", str(state.root), "--write", - ] - if bool(args.get("validate", False)): - command.append("--validate") - - proc = subprocess.run( - command, - cwd=state.root, - check=False, - capture_output=True, - text=True, - ) - + """Rebuild the in-memory graph; with validate=true also apply the commit gate.""" + build = kb_graph.build_graph(state.root) + state.graph_cache = build.graph + problems = kb_graph.validation_problems(build) if bool(args.get("validate", False)) else [] payload = { - "command": command, - "exit_code": proc.returncode, - "stdout": proc.stdout, - "stderr": proc.stderr, + "vault_root": str(state.root), + "generated_at": build.graph.get("generated_at"), + "node_count": len(build.graph.get("nodes", [])), + "edge_count": len(build.graph.get("edges", [])), + "validated": bool(args.get("validate", False)), + "problems": problems, } - if proc.returncode != 0: + if problems: return { "isError": True, "content": [{"type": "text", "text": json.dumps(payload, indent=2)}], @@ -691,7 +401,7 @@ def _tool_write_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: content = str(args["content"]) target = _resolve_in_root(state, args.get("path")) relative = _relative_to_root(state, target) - if relative == state.graph_path.relative_to(state.root.resolve()).as_posix(): + if relative == _GENERATED_GRAPH_PATH: raise ValueError( "index/graph.json is generated \u2014 use refresh_graph instead of writing it" ) @@ -706,6 +416,7 @@ def _tool_write_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") + state.invalidate() payload = { "path": relative, "created": not existed, @@ -742,6 +453,7 @@ def _tool_edit_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: updated = text.replace(old_string, new_string) target.write_text(updated, encoding="utf-8") + state.invalidate() payload = { "path": relative, "replacements": occurrences if replace_all else 1, @@ -818,6 +530,7 @@ def _tool_capture_raw(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(data) source_label = url + state.invalidate() payload = { "path": relative, @@ -918,16 +631,8 @@ def _tool_git_push(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]: def _call_tool(state: AppState, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: - if name == "read_instructions": - return _tool_read_instructions() - if name == "graph_summary": - return _tool_graph_summary(state, arguments) - if name == "graph_search_nodes": - return _tool_graph_search_nodes(state, arguments) - if name == "graph_neighbors": - return _tool_graph_neighbors(state, arguments) - if name == "read_note": - return _tool_read_note(state, arguments) + if name in query_tools.READ_ONLY_TOOL_NAMES: + return query_tools.call(name, arguments, graph=state.graph(), root=state.root) if name == "refresh_graph": return _tool_refresh_graph(state, arguments) if name == "write_note": @@ -949,7 +654,7 @@ def _list_resources(state: AppState) -> Dict[str, Any]: { "uri": "repo://graph", "name": "Repository Graph", - "description": "Raw index/graph.json", + "description": "The graph built in memory from the vault's Markdown", "mimeType": "application/json", }, { @@ -970,7 +675,7 @@ def _list_resources(state: AppState) -> Dict[str, Any]: def _read_resource(state: AppState, uri: str) -> Dict[str, Any]: if uri == "repo://graph": - text = state.graph_path.read_text(encoding="utf-8") + text = json.dumps(state.graph(), indent=2) return { "contents": [ { @@ -1038,7 +743,7 @@ def _handle_request(state: AppState, message: Dict[str, Any]) -> Dict[str, Any] def main() -> None: args = parse_vault_args(argparse.ArgumentParser(description=__doc__)) root = args.vault_root - state = AppState(root=root, graph_path=root / "index" / "graph.json") + state = AppState(root=root) reader = _StdinReader(sys.stdin.buffer) try: diff --git a/scripts/pre-commit b/scripts/pre-commit index 20d3ffb..2235abb 100755 --- a/scripts/pre-commit +++ b/scripts/pre-commit @@ -1,15 +1,9 @@ #!/usr/bin/env bash -# Generate and validate the selected vault graph before accepting a commit. +# Validate the selected vault graph before accepting a commit. +# The graph is derived in memory; nothing is written and nothing must be staged. set -euo pipefail framework_root="$(cd -- "$(dirname -- "$(readlink -f -- "${BASH_SOURCE[0]}")")/.." && pwd)" vault_root="${1:-${KNOWLEDGE_BASE_ROOT:-$framework_root}}" -echo "[pre-commit] Generating and validating graph..." -"${PYTHON:-python3}" "$framework_root/scripts/generate-graph.py" --vault-root "$vault_root" --write --validate - -if ! git -C "$vault_root" ls-files --error-unmatch index/graph.json >/dev/null 2>&1 || \ - ! git -C "$vault_root" diff --quiet -- index/graph.json; then - echo "[pre-commit] index/graph.json was modified by generation." - echo "[pre-commit] Stage the generated index/graph.json before committing." - exit 1 -fi +echo "[pre-commit] Validating graph..." +"${PYTHON:-python3}" "$framework_root/scripts/generate-graph.py" --vault-root "$vault_root" --validate diff --git a/scripts/vault_root.py b/scripts/vault_root.py index 9e44e4f..1acf5a7 100644 --- a/scripts/vault_root.py +++ b/scripts/vault_root.py @@ -1,11 +1,18 @@ -"""Resolve the selected data directory consistently for framework command-line tools.""" +"""Resolve the selected data directory consistently for framework command-line tools. + +Also puts the framework root on `sys.path` so every script can import the +`kb_framework` package from a plain checkout without installing it. +""" import argparse import os +import sys from pathlib import Path FRAMEWORK_ROOT = Path(__file__).resolve().parents[1] +if str(FRAMEWORK_ROOT) not in sys.path: + sys.path.insert(0, str(FRAMEWORK_ROOT)) def parse_vault_args(parser: argparse.ArgumentParser) -> argparse.Namespace: diff --git a/skills/knowledge-base-ingest-resource/SKILL.md b/skills/knowledge-base-ingest-resource/SKILL.md index dc84896..46996a4 100644 --- a/skills/knowledge-base-ingest-resource/SKILL.md +++ b/skills/knowledge-base-ingest-resource/SKILL.md @@ -23,7 +23,7 @@ not that framework or caller instructions cease to apply. The `knowledge-database` MCP server is the only interface for vault access: use `graph_summary`, `graph_search_nodes`, `graph_neighbors` and `read_note` to find context; `capture_raw` for source bytes; `write_note` and -`edit_note` for notes; `refresh_graph(validate=true)` to regenerate and validate. +`edit_note` for notes; `refresh_graph(validate=true)` to rebuild the in-memory graph and validate. Use `git_commit` / `git_push` only when the caller explicitly authorizes them. Skills stay in the framework. Never infer the vault from this skill's location. @@ -95,7 +95,7 @@ before writing content notes. Never invent transcript text or run shell extracto ## 4. Read the captured content Use `read_note(path="03_resources//raw/")` for captured -text. Path reads access the file directly and work before graph regeneration. +text. Path reads access the file directly and work even before the graph picks the file up. For a captured binary, use the invoking agent’s explicitly permitted inspection tool on the absolute path returned by `capture_raw`; if unavailable, report it. This — not the live URL — is the source of truth for the rest of the steps, so the note @@ -233,17 +233,17 @@ So the resource is reachable in the graph, not just reaching out: lists them, and link the new note from there (create the hub if it does not exist, with `type: hub`, `tags: [resource, hub]`). -## 9. Regenerate and validate the graph +## 9. Rebuild and validate the graph Use `refresh_graph(validate=true)`. -This regenerates `index/graph.json` and validates links and edge relationship -types. If validation reports unresolved links, fix the targets (typos, wrong `id`, +This rebuilds the in-memory graph from the Markdown and validates links, edge +relationship types and note types; the result lists every problem under +`problems` and is an error result while any remain. If validation reports unresolved links, fix the targets (typos, wrong `id`, or a link to a note that does not exist yet — either create it or downgrade it to plain text / external URL); if it reports an unknown edge `type`, correct it to a -value in the controlled vocabulary. Re-run until clean. The pre-commit hook rejects -a stale or -invalid graph, so this must pass. +value in the controlled vocabulary. Re-run until clean. The pre-commit hook runs the same validation, so this must +pass. There is no graph file to stage. ## 10. Stage (do not commit unless asked) @@ -251,7 +251,7 @@ Without explicit publishing authorization, leave the changes uncommitted and report the touched paths. The MCP interface has no stage-only tool. When a commit is authorized, use `git_commit(message=..., paths=[...])` with an -explicit list covering the resource folder, backlinked notes and `index/graph.json`. +explicit list covering the resource folder and every backlinked note. This tool stages those paths and commits them; do not use wildcard paths. Use `git_push` only when pushing is explicitly authorized. Follow the caller's branch and review policy. diff --git a/skills/knowledge-base-read/SKILL.md b/skills/knowledge-base-read/SKILL.md index 304d91c..52e564a 100644 --- a/skills/knowledge-base-read/SKILL.md +++ b/skills/knowledge-base-read/SKILL.md @@ -1,14 +1,14 @@ --- name: knowledge-base-read -description: Read and query the knowledge-base vault — a personal docs-as-code knowledge database (PARA-organized). Use when the user asks what is known about a repo, policy, decision (ADR), area, or project; wants context before working in a repo; or asks to find/navigate notes by tag, type, repo, or link. Resolves the question against the Markdown notes and the generated graph index. +description: Read and query the knowledge-base vault — a personal docs-as-code knowledge database (PARA-organized). Use when the user asks what is known about a repo, policy, decision (ADR), area, or project; wants context before working in a repo; or asks to find/navigate notes by tag, type, repo, or link. Resolves the question against the Markdown notes and the graph the MCP server builds from them in memory. --- # Read the Knowledge Base This vault is a personal docs-as-code knowledge database, organized with the [PARA method](https://fortelabs.com/blog/para/). -Markdown is the source of truth; `index/graph.json` is a generated navigation -index. Use this skill to answer questions FROM the vault and to surface the +Markdown is the source of truth; the graph is a navigation index the MCP +server builds from it in memory (nothing on disk). Use this skill to answer questions FROM the vault and to surface the exact notes that back the answer. Call `read_instructions` on `knowledge-database` to read the framework's @@ -20,7 +20,7 @@ This skill is read-only — do not modify notes here. The `knowledge-database` MCP server is the only interface for vault access. Use `graph_summary` to orient, `graph_search_nodes` to find notes, `graph_neighbors` to traverse links, and `read_note` to read the source. -Use `refresh_graph` when the generated index is stale. Skills and operational +Use `refresh_graph` when the graph looks stale. Skills and operational documentation belong to the framework; never infer the data vault from this skill's file location. @@ -34,7 +34,6 @@ skill's file location. 03_resources/repos// per-repo info.md (+ decisions/) 03_resources/repos/repo-landscape.md index of documented repos 04_archives/ inactive items -index/graph.json generated graph (nodes + edges) ``` @@ -79,8 +78,8 @@ policies, decisions, and repo info. Use each edge's `type` and `annotation` to judge which links are worth following; links are the intended navigation, so traverse them rather than guessing. -If the graph looks stale or a path is missing, use `refresh_graph`. This updates -only the generated index; it does not touch Markdown. Do not commit. +If the graph looks stale or a path is missing, use `refresh_graph`. This rebuilds +the in-memory graph; it does not touch Markdown. Nothing to commit. ## 4. Answer with citations diff --git a/tests/fixtures/graph_golden.json b/tests/fixtures/graph_golden.json new file mode 100644 index 0000000..dbac6dd --- /dev/null +++ b/tests/fixtures/graph_golden.json @@ -0,0 +1,105 @@ +{ + "version": 1, + "nodes": [ + { + "id": "README", + "title": "README", + "path": "README.md", + "repo": "global", + "type": "note", + "status": "active", + "tags": [], + "last_reviewed": "", + "outbound_links": [], + "inbound_count": 0 + }, + { + "id": "area-ops", + "title": "Area Ops", + "path": "02_areas/ops.md", + "repo": "global", + "type": "area", + "status": "active", + "tags": [ + "ops" + ], + "last_reviewed": "", + "outbound_links": [ + "project-alpha" + ], + "inbound_count": 2 + }, + { + "id": "project-alpha", + "title": "Project Alpha", + "path": "01_projects/alpha.md", + "repo": "alpha", + "type": "project", + "status": "active", + "tags": [ + "alpha", + "demo" + ], + "last_reviewed": "2026-01-01", + "outbound_links": [ + "area-ops", + "resource-demo" + ], + "inbound_count": 1 + }, + { + "id": "resource-demo", + "title": "Demo Resource", + "path": "03_resources/demo/index.md", + "repo": "global", + "type": "reference", + "status": "active", + "tags": [ + "demo" + ], + "last_reviewed": "", + "outbound_links": [ + "area-ops" + ], + "inbound_count": 1 + } + ], + "edges": [ + { + "from": "area-ops", + "to": "project-alpha", + "kind": "markdown_link", + "type": "part-of", + "anchor_text": "Project Alpha", + "annotation": "", + "weight": 1.0 + }, + { + "from": "project-alpha", + "to": "area-ops", + "kind": "markdown_link", + "type": "depends-on", + "anchor_text": "Area Ops", + "annotation": "operations area", + "weight": 1.0 + }, + { + "from": "project-alpha", + "to": "resource-demo", + "kind": "markdown_link", + "type": "related", + "anchor_text": "the demo resource", + "annotation": "resource landing page.", + "weight": 1.0 + }, + { + "from": "resource-demo", + "to": "area-ops", + "kind": "markdown_link", + "type": "related", + "anchor_text": "Area Ops", + "annotation": "", + "weight": 1.0 + } + ] +} \ No newline at end of file diff --git a/tests/fixtures/stdio_golden.json b/tests/fixtures/stdio_golden.json new file mode 100644 index 0000000..ee8fead --- /dev/null +++ b/tests/fixtures/stdio_golden.json @@ -0,0 +1,259 @@ +{ + "graph_summary": { + "tool": "graph_summary", + "args": { + "topLinked": 3, + "topTags": 2 + }, + "structuredContent": { + "version": 1, + "node_count": 4, + "edge_count": 4, + "by_repo": { + "global": 3, + "alpha": 1 + }, + "tag_count": 3, + "by_tag": [ + { + "tag": "demo", + "count": 2 + }, + { + "tag": "alpha", + "count": 1 + } + ], + "top_linked": [ + { + "id": "area-ops", + "title": "Area Ops", + "path": "02_areas/ops.md", + "inbound_count": 2 + }, + { + "id": "project-alpha", + "title": "Project Alpha", + "path": "01_projects/alpha.md", + "inbound_count": 1 + }, + { + "id": "resource-demo", + "title": "Demo Resource", + "path": "03_resources/demo/index.md", + "inbound_count": 1 + } + ] + } + }, + "graph_search_nodes": { + "tool": "graph_search_nodes", + "args": { + "query": "alpha", + "limit": 5 + }, + "structuredContent": { + "total_matches": 1, + "returned": 1, + "items": [ + { + "id": "project-alpha", + "title": "Project Alpha", + "path": "01_projects/alpha.md", + "repo": "alpha", + "type": "project", + "status": "active", + "tags": [ + "alpha", + "demo" + ], + "last_reviewed": "2026-01-01", + "outbound_links": [ + "area-ops", + "resource-demo" + ], + "inbound_count": 1 + } + ] + } + }, + "graph_neighbors": { + "tool": "graph_neighbors", + "args": { + "id": "area-ops" + }, + "structuredContent": { + "node": { + "id": "area-ops", + "title": "Area Ops", + "path": "02_areas/ops.md", + "repo": "global", + "type": "area", + "status": "active", + "tags": [ + "ops" + ], + "last_reviewed": "", + "outbound_links": [ + "project-alpha" + ], + "inbound_count": 2 + }, + "inbound_count": 2, + "outbound_count": 1, + "inbound": [ + { + "id": "project-alpha", + "title": "Project Alpha", + "path": "01_projects/alpha.md", + "direction": "inbound", + "type": "depends-on", + "anchor_text": "Area Ops", + "annotation": "operations area" + }, + { + "id": "resource-demo", + "title": "Demo Resource", + "path": "03_resources/demo/index.md", + "direction": "inbound", + "type": "related", + "anchor_text": "Area Ops", + "annotation": "" + } + ], + "outbound": [ + { + "id": "project-alpha", + "title": "Project Alpha", + "path": "01_projects/alpha.md", + "direction": "outbound", + "type": "part-of", + "anchor_text": "Project Alpha", + "annotation": "" + } + ] + } + }, + "read_note_by_id": { + "tool": "read_note", + "args": { + "id": "resource-demo" + }, + "structuredContent": { + "path": "03_resources/demo/index.md", + "binary": false, + "content": "---\nid: resource-demo\ntitle: Demo Resource\ntype: reference\ntags: [demo]\n---\n\n# Demo Resource\n\nRaw original kept in `raw/`. Related: [Area Ops](02_areas/ops.md).\n" + } + }, + "read_note_by_path": { + "tool": "read_note", + "args": { + "path": "02_areas/ops.md" + }, + "structuredContent": { + "path": "02_areas/ops.md", + "binary": false, + "content": "---\nid: area-ops\ntitle: Area Ops\ntype: area\ntags: [ops]\n---\n\n# Area Ops\n\nParent hub: [Project Alpha](project-alpha){\"type\":\"part-of\"}.\n" + } + }, + "tool_definitions": [ + { + "name": "read_instructions", + "description": "Read framework AGENTS.md and AGENT_QUERY_GUIDE.md before working with the vault. These operating documents live outside an external vault and do not require a generated graph.", + "inputSchema": { + "type": "object", + "properties": {}, + "additionalProperties": false + } + }, + { + "name": "graph_summary", + "description": "Orientation for a cold start: node and edge counts, the repo scopes, the most-linked notes, and the tag vocabulary as `by_tag` (most frequent first) plus `tag_count`, the number of distinct tags. Read `by_tag` before filtering `graph_search_nodes` by `tag` \u2014 that filter matches a tag verbatim, so a guessed spelling returns nothing.", + "inputSchema": { + "type": "object", + "properties": { + "topLinked": { + "type": "integer", + "minimum": 1, + "maximum": 50, + "description": "Most-linked notes to return (default 10)." + }, + "topTags": { + "type": "integer", + "minimum": 1, + "maximum": 500, + "description": "Tags to return, most frequent first (default 50)." + } + }, + "additionalProperties": false + } + }, + { + "name": "graph_search_nodes", + "description": "Search graph nodes by metadata and free text.", + "inputSchema": { + "type": "object", + "properties": { + "repo": { + "type": "string" + }, + "type": { + "type": "string" + }, + "status": { + "type": "string" + }, + "tag": { + "type": "string" + }, + "query": { + "type": "string" + }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 200 + } + } + } + }, + { + "name": "graph_neighbors", + "description": "Show inbound and outbound links for a node ID. Each neighbor includes the edge `type` (relationship, e.g. supersedes, depends-on), `anchor_text` (the link's display text), and `annotation` (what to expect in that note).", + "inputSchema": { + "type": "object", + "required": [ + "id" + ], + "properties": { + "id": { + "type": "string" + }, + "limit": { + "type": "integer", + "minimum": 1, + "maximum": 200 + } + } + } + }, + { + "name": "read_note", + "description": "Read note content by graph node ID or vault-relative path. Provide exactly one of `id` or `path`. If both or neither are provided, the call returns an error.", + "inputSchema": { + "type": "object", + "properties": { + "id": { + "type": "string", + "description": "Graph node ID (e.g. 'AGENTS')." + }, + "path": { + "type": "string", + "description": "Vault-relative file path (e.g. 'README.md')." + } + }, + "additionalProperties": false + } + } + ] +} \ No newline at end of file diff --git a/tests/fixtures/vault/01_projects/alpha.md b/tests/fixtures/vault/01_projects/alpha.md new file mode 100644 index 0000000..0ca44c8 --- /dev/null +++ b/tests/fixtures/vault/01_projects/alpha.md @@ -0,0 +1,15 @@ +--- +id: project-alpha +title: Project Alpha +type: project +status: active +tags: [alpha, demo] +repo: alpha +last_reviewed: 2026-01-01 +--- + +# Project Alpha + +Depends on [Area Ops](area-ops){"type":"depends-on","link-info":"operations area"}. +See [the demo resource](resource-demo) — resource landing page. +External: [MCP spec](https://modelcontextprotocol.io/). diff --git a/tests/fixtures/vault/02_areas/ops.md b/tests/fixtures/vault/02_areas/ops.md new file mode 100644 index 0000000..23d49e6 --- /dev/null +++ b/tests/fixtures/vault/02_areas/ops.md @@ -0,0 +1,10 @@ +--- +id: area-ops +title: Area Ops +type: area +tags: [ops] +--- + +# Area Ops + +Parent hub: [Project Alpha](project-alpha){"type":"part-of"}. diff --git a/tests/fixtures/vault/03_resources/demo/index.md b/tests/fixtures/vault/03_resources/demo/index.md new file mode 100644 index 0000000..049907b --- /dev/null +++ b/tests/fixtures/vault/03_resources/demo/index.md @@ -0,0 +1,10 @@ +--- +id: resource-demo +title: Demo Resource +type: reference +tags: [demo] +--- + +# Demo Resource + +Raw original kept in `raw/`. Related: [Area Ops](02_areas/ops.md). diff --git a/tests/fixtures/vault/03_resources/demo/raw/original.md b/tests/fixtures/vault/03_resources/demo/raw/original.md new file mode 100644 index 0000000..a130dc3 --- /dev/null +++ b/tests/fixtures/vault/03_resources/demo/raw/original.md @@ -0,0 +1,2 @@ +not a note +[broken](../missing.md) diff --git a/tests/fixtures/vault/03_resources/demo/raw/paper.pdf b/tests/fixtures/vault/03_resources/demo/raw/paper.pdf new file mode 100644 index 0000000..90ff851 Binary files /dev/null and b/tests/fixtures/vault/03_resources/demo/raw/paper.pdf differ diff --git a/tests/fixtures/vault/README.md b/tests/fixtures/vault/README.md new file mode 100644 index 0000000..7e59600 --- /dev/null +++ b/tests/fixtures/vault/README.md @@ -0,0 +1 @@ +# README diff --git a/tests/test_external_vault.py b/tests/test_external_vault.py index 5c2feed..052e5d0 100644 --- a/tests/test_external_vault.py +++ b/tests/test_external_vault.py @@ -20,6 +20,8 @@ class ExternalVaultTests(unittest.TestCase): self.base = Path(self.tmp.name) self.framework = self.base / "framework space" shutil.copytree(FRAMEWORK / "scripts", self.framework / "scripts") + shutil.copytree(FRAMEWORK / "kb_framework", self.framework / "kb_framework", + ignore=shutil.ignore_patterns("__pycache__")) shutil.copy2(FRAMEWORK / "Makefile", self.framework / "Makefile") self.vault = self.base / "vault space" self.other = self.base / "other" @@ -115,15 +117,14 @@ class ExternalVaultTests(unittest.TestCase): ("write_note", {"path": "../escape.md", "content": "escape"})]: self.assertTrue(self.tool(tool, args).get("isError")) - def test_refresh_uses_absolute_framework_generator_and_writes_when_validating(self): + def test_refresh_validates_in_memory_and_never_runs_vault_tools(self): (self.vault / "scripts").mkdir() (self.vault / "scripts/generate-graph.py").write_text("raise RuntimeError('vault tool used')\n") result = self.tool("refresh_graph", {"validate": True}) self.assertFalse(result.get("isError"), result) - command = result["structuredContent"]["command"] - self.assertEqual(command[1], str(self.framework / "scripts/generate-graph.py")) - self.assertEqual(command[command.index("--vault-root") + 1], str(self.vault)) - self.assertTrue((self.vault / "index/graph.json").exists()) + self.assertEqual(result["structuredContent"]["vault_root"], str(self.vault)) + self.assertEqual(result["structuredContent"]["node_count"], 1) + self.assertFalse((self.vault / "index/graph.json").exists()) summary = self.tool("graph_summary") self.assertFalse(summary.get("isError"), summary) @@ -178,16 +179,17 @@ class ExternalVaultTests(unittest.TestCase): def make(*args): return subprocess.run(["make", "-C", str(self.framework), f"VAULT_ROOT={self.vault}", *args], env=self.env, capture_output=True, text=True) - proc = make("graph", "install-hooks") + proc = make("validate-graph", "install-hooks") self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) hook = self.vault / ".git/hooks/pre-commit" self.assertTrue(hook.exists()) - self.git(self.vault, "add", "index/graph.json") proc = subprocess.run([str(hook)], cwd=self.other, env=self.env, capture_output=True, text=True) self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) - (self.vault / "03_resources/new.md").write_text("# new\n") + (self.vault / "03_resources/new.md").write_text("# new\n[gone](nowhere)\n") proc = subprocess.run([str(hook)], cwd=self.other, env=self.env, capture_output=True, text=True) self.assertNotEqual(proc.returncode, 0) + self.assertIn("nowhere", proc.stdout + proc.stderr) + self.assertFalse((self.vault / "index/graph.json").exists()) self.assertFalse((self.framework / "index/graph.json").exists()) def test_installer_preserves_unmanaged_hooks_and_reinstalls_managed_hook(self): diff --git a/tests/test_generate_graph.py b/tests/test_generate_graph.py index 9fc8013..ea94e07 100644 --- a/tests/test_generate_graph.py +++ b/tests/test_generate_graph.py @@ -63,12 +63,7 @@ class TestValidateExitCode(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name).resolve() / "vault" - (self.root / "scripts").mkdir(parents=True) - (self.root / "03_resources").mkdir() - for name in ("generate-graph.py", "vault_root.py"): - (self.root / "scripts" / name).write_text( - (SCRIPTS / name).read_text(encoding="utf-8"), encoding="utf-8", - ) + (self.root / "03_resources").mkdir(parents=True) def tearDown(self) -> None: self._tmp.cleanup() @@ -81,7 +76,7 @@ class TestValidateExitCode(unittest.TestCase): def validate(self) -> subprocess.CompletedProcess: return subprocess.run( - [sys.executable, "scripts/generate-graph.py", "--validate"], + [sys.executable, str(SCRIPTS / "generate-graph.py"), "--vault-root", str(self.root), "--validate"], cwd=self.root, capture_output=True, text=True, diff --git a/tests/test_in_memory_graph.py b/tests/test_in_memory_graph.py new file mode 100644 index 0000000..d49d3c1 --- /dev/null +++ b/tests/test_in_memory_graph.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +"""Tests for the graph-in-memory contract of the stdio server and the hook. + +`index/graph.json` is no longer a committed artifact: the MCP server builds +the graph lazily from Markdown, invalidates it on writes, and `refresh_graph` +rebuilds (and optionally validates) in memory. The pre-commit hook only +validates. These tests pin that nothing in the default flow writes the file. +""" + +from __future__ import annotations + +import importlib +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +FRAMEWORK = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(FRAMEWORK / "scripts")) +mcp = importlib.import_module("mcp-server") + + +def write_note(root: Path, rel: str, note_id: str, body: str = "") -> None: + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"---\nid: {note_id}\ntitle: {note_id}\ntype: note\n---\n# {note_id}\n{body}\n", encoding="utf-8") + + +class InMemoryVault(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name).resolve() / "vault" + self.root.mkdir() + write_note(self.root, "03_resources/a.md", "a", "[b](b)") + write_note(self.root, "03_resources/b.md", "b") + self.state = mcp.AppState(root=self.root) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def call(self, name: str, **args) -> dict: + return mcp._call_tool(self.state, name, args) + + +class TestStdioServerBuildsGraphLazily(InMemoryVault): + def test_summary_without_graph_file(self) -> None: + payload = self.call("graph_summary")["structuredContent"] + self.assertEqual(payload["node_count"], 2) + self.assertEqual(payload["edge_count"], 1) + self.assertFalse((self.root / "index").exists()) + + def test_write_note_invalidates_cache(self) -> None: + self.call("graph_summary") + self.call("write_note", path="03_resources/c.md", content="---\nid: c\n---\n# c\n") + self.assertEqual(self.call("graph_summary")["structuredContent"]["node_count"], 3) + + def test_edit_note_invalidates_cache(self) -> None: + self.call("graph_summary") + self.call("edit_note", path="03_resources/b.md", old_string="# b", new_string="# b\n[a](a)") + self.assertEqual(self.call("graph_summary")["structuredContent"]["edge_count"], 2) + + def test_refresh_rebuilds_and_validates_in_memory(self) -> None: + self.call("graph_summary") + write_note(self.root, "03_resources/d.md", "d", "[gone](nowhere)") + ok = self.call("refresh_graph") + self.assertFalse(ok.get("isError"), ok) + self.assertEqual(ok["structuredContent"]["node_count"], 3) + bad = self.call("refresh_graph", validate=True) + self.assertTrue(bad.get("isError"), bad) + self.assertIn("nowhere", json.dumps(bad["structuredContent"])) + self.assertFalse((self.root / "index").exists()) + + def test_graph_resource_is_served_from_memory(self) -> None: + result = mcp._read_resource(self.state, "repo://graph") + graph = json.loads(result["contents"][0]["text"]) + self.assertEqual(len(graph["nodes"]), 2) + self.assertFalse((self.root / "index").exists()) + + def test_read_note_by_id_uses_in_memory_graph(self) -> None: + payload = self.call("read_note", id="b")["structuredContent"] + self.assertEqual(payload["path"], "03_resources/b.md") + + +class TestGeneratorAndHookValidateOnly(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.root = Path(self._tmp.name).resolve() / "vault" + self.root.mkdir() + write_note(self.root, "03_resources/a.md", "a") + self.env = {k: v for k, v in os.environ.items() if k != "KNOWLEDGE_BASE_ROOT"} + + def tearDown(self) -> None: + self._tmp.cleanup() + + def run_cmd(self, *cmd: str) -> subprocess.CompletedProcess: + return subprocess.run(list(cmd), cwd=self.root, env=self.env, capture_output=True, text=True) + + def test_validate_alone_writes_nothing(self) -> None: + proc = self.run_cmd(sys.executable, str(FRAMEWORK / "scripts/generate-graph.py"), "--vault-root", str(self.root), "--validate") + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertFalse((self.root / "index").exists()) + + def test_write_still_produces_a_file_for_operators(self) -> None: + proc = self.run_cmd(sys.executable, str(FRAMEWORK / "scripts/generate-graph.py"), "--vault-root", str(self.root), "--write") + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertTrue((self.root / "index/graph.json").exists()) + + def test_hook_validates_without_requiring_a_staged_graph(self) -> None: + hook = FRAMEWORK / "scripts/pre-commit" + proc = self.run_cmd("bash", str(hook), str(self.root)) + self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) + self.assertFalse((self.root / "index").exists()) + write_note(self.root, "03_resources/b.md", "b", "[x](missing)") + proc = self.run_cmd("bash", str(hook), str(self.root)) + self.assertNotEqual(proc.returncode, 0) + self.assertIn("missing", proc.stdout + proc.stderr) + self.assertFalse((self.root / "index").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_kb_framework.py b/tests/test_kb_framework.py new file mode 100644 index 0000000..c8ce59f --- /dev/null +++ b/tests/test_kb_framework.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Tests for the importable `kb_framework` package. + +The package is the contract both the stdio MCP server and the hosted connector +consume: build the graph in memory from a vault's Markdown, answer the query +tools over that graph, and read framework instructions. These tests pin the +graph shape against a golden fixture produced by the file-writing generator, +and the tool payloads against goldens captured from the stdio server, so the +two front ends cannot drift apart. +""" + +from __future__ import annotations + +import json +import sys +import unittest +from pathlib import Path + +FRAMEWORK = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(FRAMEWORK)) +FIXTURES = FRAMEWORK / "tests" / "fixtures" +VAULT = FIXTURES / "vault" + +from kb_framework import graph as kbg # noqa: E402 +from kb_framework import query as kbq # noqa: E402 +from kb_framework import query_tools as kbt # noqa: E402 +from kb_framework import vault as kbv # noqa: E402 +from kb_framework import instructions as kbi # noqa: E402 + + +def golden_graph() -> dict: + return json.loads((FIXTURES / "graph_golden.json").read_text(encoding="utf-8")) + + +def golden_stdio() -> dict: + return json.loads((FIXTURES / "stdio_golden.json").read_text(encoding="utf-8")) + + +class TestBuildGraphInMemory(unittest.TestCase): + def test_matches_golden_without_touching_disk(self) -> None: + result = kbg.build_graph(VAULT) + self.assertEqual(result.graph["nodes"], golden_graph()["nodes"]) + self.assertEqual(result.graph["edges"], golden_graph()["edges"]) + self.assertEqual(result.graph["version"], 1) + self.assertTrue(result.graph["generated_at"]) + self.assertFalse((VAULT / "index").exists()) + + def test_reports_no_problems_for_clean_fixture(self) -> None: + result = kbg.build_graph(VAULT) + self.assertEqual(result.unresolved, []) + self.assertEqual(result.wiki_uses, []) + self.assertEqual(kbg.validation_problems(result), []) + + def test_validation_reports_unresolved_and_bad_types(self) -> None: + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + (root / "a.md").write_text( + "---\nid: a\ntype: descision\n---\n[x](missing){\"type\":\"bogus\"}\n[[legacy]]\n", + encoding="utf-8", + ) + result = kbg.build_graph(root) + problems = "\n".join(kbg.validation_problems(result)) + self.assertIn("missing", problems) + self.assertIn("descision", problems) + self.assertIn("legacy", problems) + self.assertFalse(result.ok) + + +class TestQueryFunctions(unittest.TestCase): + def setUp(self) -> None: + self.graph = golden_graph() + self.stdio = golden_stdio() + + def strip(self, payload: dict) -> dict: + payload = dict(payload) + payload.pop("generated_at", None) + return payload + + def test_summary_matches_stdio_golden(self) -> None: + case = self.stdio["graph_summary"] + payload = kbq.summary(self.graph, **case["args"]) + self.assertEqual(self.strip(payload), case["structuredContent"]) + + def test_search_matches_stdio_golden(self) -> None: + case = self.stdio["graph_search_nodes"] + self.assertEqual(kbq.search_nodes(self.graph, case["args"]), case["structuredContent"]) + + def test_neighbors_matches_stdio_golden(self) -> None: + case = self.stdio["graph_neighbors"] + self.assertEqual(kbq.neighbors(self.graph, **case["args"]), case["structuredContent"]) + + def test_neighbors_unknown_id_raises(self) -> None: + with self.assertRaises(ValueError): + kbq.neighbors(self.graph, "nope") + + def test_node_path_resolves_id(self) -> None: + self.assertEqual(kbq.node_path(self.graph, "area-ops"), "02_areas/ops.md") + with self.assertRaises(ValueError): + kbq.node_path(self.graph, "nope") + + +class TestVaultHelpers(unittest.TestCase): + def test_resolve_in_root_rejects_escape_and_absolute(self) -> None: + for bad in ("../x.md", "/etc/passwd", "", " "): + with self.subTest(bad=bad), self.assertRaises(ValueError): + kbv.resolve_in_root(VAULT, bad) + self.assertEqual(kbv.resolve_in_root(VAULT, "README.md"), VAULT / "README.md") + + def test_read_note_payload_text_and_binary(self) -> None: + text = kbv.read_note_payload(VAULT, "02_areas/ops.md") + self.assertEqual(text["binary"], False) + self.assertIn("# Area Ops", text["content"]) + binary = kbv.read_note_payload(VAULT, "03_resources/demo/raw/paper.pdf") + self.assertTrue(binary["binary"]) + self.assertEqual(set(binary), {"path", "binary", "bytes", "sha256", "hint"}) + with self.assertRaises(FileNotFoundError): + kbv.read_note_payload(VAULT, "nope.md") + + +class TestQueryTools(unittest.TestCase): + """The five read-only tools, as the stdio server and the connector expose them.""" + + def setUp(self) -> None: + self.graph = golden_graph() + self.stdio = golden_stdio() + + def test_definitions_match_stdio(self) -> None: + self.assertEqual(kbt.definitions(), self.stdio["tool_definitions"]) + + def test_call_returns_stdio_result_shape(self) -> None: + for key in ("graph_summary", "graph_search_nodes", "graph_neighbors", "read_note_by_id", "read_note_by_path"): + case = self.stdio[key] + with self.subTest(tool=key): + result = kbt.call(case["tool"], case["args"], graph=self.graph, root=VAULT) + sc = dict(result["structuredContent"]) + sc.pop("generated_at", None) + self.assertEqual(sc, case["structuredContent"]) + self.assertEqual(result["content"][0]["type"], "text") + self.assertEqual(json.loads(result["content"][0]["text"]).get("path", sc.get("path")), sc.get("path")) + + def test_read_note_requires_exactly_one_selector(self) -> None: + for args in ({}, {"id": "area-ops", "path": "README.md"}): + with self.subTest(args=args), self.assertRaises(ValueError): + kbt.call("read_note", args, graph=self.graph, root=VAULT) + + def test_unknown_tool_raises(self) -> None: + with self.assertRaises(ValueError): + kbt.call("write_note", {}, graph=self.graph, root=VAULT) + + +class TestInstructions(unittest.TestCase): + def test_reads_framework_documents(self) -> None: + payload = kbi.read_instructions() + self.assertEqual([d["path"] for d in payload["documents"]], ["AGENTS.md", "AGENT_QUERY_GUIDE.md"]) + self.assertIn("AGENT_QUERY_GUIDE", payload["documents"][0]["content"]) + self.assertTrue(Path(payload["framework_root"]).is_dir()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 8297831..25b592e 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -40,9 +40,7 @@ class ToolTestCase(unittest.TestCase): self._tmp = tempfile.TemporaryDirectory() self.root = Path(self._tmp.name).resolve() / "vault" self.root.mkdir() - self.state = mcp.AppState( - root=self.root, graph_path=self.root / "index" / "graph.json" - ) + self.state = mcp.AppState(root=self.root) _git(self.root, "init", "-q", "-b", "main") _git(self.root, "config", "user.email", "test@example.com") _git(self.root, "config", "user.name", "Test") @@ -440,18 +438,13 @@ class TestGraphSummary(ToolTestCase): """ def write_graph(self, nodes: list) -> None: - self.state.graph_path.parent.mkdir(parents=True, exist_ok=True) - self.state.graph_path.write_text( - json.dumps( - { - "version": 1, - "generated_at": "2026-09-14T00:00:00Z", - "nodes": nodes, - "edges": [], - } - ), - encoding="utf-8", - ) + # The graph lives in memory; seed the cache the way a build would. + self.state.graph_cache = { + "version": 1, + "generated_at": "2026-09-14T00:00:00Z", + "nodes": nodes, + "edges": [], + } def node(self, node_id: str, **fields) -> dict: base = {