lachtan/nanobot-runtime
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

provozni zaloha

Browse Source
This commit is contained in:
lachtan
2026-06-24 08:11:12 +02:00
parent 9295dba19f
commit 1db3ec4756
97 changed files with 7698 additions and 817 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
skills/llm-wiki/scripts/init_wiki.py Normal file
View File

@@ -0,0 +1,206 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
init_wiki.py — Bootstrap or upgrade an LLM Wiki structure in a project.
Plain init creates the directory layout and drops in templates for SCHEMA.md,
index.md, log.md, the page template, and the optional graph layer
(graph/ontology.yaml, graph/README.md, graph/.gitignore). It is idempotent:
re-running won't clobber existing files.
`--upgrade` mode is for wikis bootstrapped under an older plugin version. It
does the same idempotent file creation, then inspects the existing SCHEMA.md
for sections introduced in newer versions and prints clear instructions for
what to merge by hand. It never overwrites SCHEMA.md — the schema is
co-evolved with the user.
Usage:
python init_wiki.py <project-root> [--wiki-dir wiki] [--raw-dir raw] [--upgrade]
Examples:
python init_wiki.py .
python init_wiki.py . --upgrade
python init_wiki.py ~/research --wiki-dir kb --raw-dir sources
"""
import argparse
import sys
from pathlib import Path
from datetime import date
SKILL_ROOT = Path(__file__).resolve().parent.parent
TEMPLATES = SKILL_ROOT / "assets"
SUBDIRS = ["sources", "entities", "concepts", "synthesis", "graph"]
# Markers used by --upgrade to detect SCHEMA.md sections introduced in
# specific plugin versions. Each entry: (heading_marker, version_label,
# template_anchor, blurb).
SCHEMA_SECTION_MARKERS = [
{
"marker": "## Optional graph metadata",
"version": "0.3.0",
"anchor": "## Optional graph metadata",
"label": "Optional graph metadata (Frontmatter section)",
},
{
"marker": "## Graph layer",
"version": "0.3.0",
"anchor": "## Graph layer",
"label": "Graph layer (canonical-vs-generated artifact policy)",
},
{
"marker": "Graph lint + extract",
"version": "0.3.0",
"anchor": "- Graph lint + extract: after every ingest that adds typed `graph.relationships`.",
"label": "Graph lint + extract cadence (Lint cadence section)",
},
]
def copy_template(src: Path, dst: Path, substitutions: dict | None = None) -> bool:
"""Copy a template file to dst. Returns True if file was created, False if it already existed."""
if dst.exists():
return False
text = src.read_text()
if substitutions:
for key, value in substitutions.items():
text = text.replace(key, value)
dst.parent.mkdir(parents=True, exist_ok=True)
dst.write_text(text)
return True
def detect_schema_gaps(schema_path: Path) -> list[dict]:
"""Return the SCHEMA_SECTION_MARKERS entries missing from the user's SCHEMA.md."""
if not schema_path.exists():
return []
text = schema_path.read_text(encoding="utf-8")
return [m for m in SCHEMA_SECTION_MARKERS if m["marker"] not in text]
def print_schema_upgrade_guidance(schema_path: Path, gaps: list[dict]) -> None:
template_path = TEMPLATES / "SCHEMA.md.template"
print()
print("=" * 64)
print(f"Upgrade required: {schema_path}")
print("=" * 64)
print(
"Your SCHEMA.md predates one or more sections introduced by newer\n"
"plugin versions. The graph layer itself is opt-in, but to make Claude\n"
"aware of it, merge the sections below by hand. SCHEMA.md is co-evolved\n"
"with you — this script never overwrites it."
)
print()
print("Missing sections:")
for m in gaps:
print(f" - [{m['version']}] {m['label']}")
print()
print(f"Reference template: {template_path}")
print(
"Diff your SCHEMA.md against the template and copy the missing\n"
"sections in. Or run /wiki:upgrade and Claude will propose the edits\n"
"interactively (one section at a time, never silent)."
)
def init_wiki(project_root: Path, wiki_dir: str, raw_dir: str, upgrade: bool = False) -> None:
project_root = project_root.resolve()
if not project_root.exists():
print(f"Error: project root does not exist: {project_root}", file=sys.stderr)
sys.exit(1)
wiki = project_root / wiki_dir
raw = project_root / raw_dir
mode = "Upgrading" if upgrade else "Initializing"
print(f"{mode} LLM Wiki in: {project_root}")
print(f" Wiki directory: {wiki}")
print(f" Raw directory: {raw}")
print()
created = []
skipped = []
# Create wiki subdirs
for subdir in SUBDIRS:
d = wiki / subdir
if not d.exists():
d.mkdir(parents=True)
created.append(f"{wiki_dir}/{subdir}/")
else:
skipped.append(f"{wiki_dir}/{subdir}/")
# Create raw + raw/assets
for d, label in [(raw, raw_dir), (raw / "assets", f"{raw_dir}/assets")]:
if not d.exists():
d.mkdir(parents=True)
created.append(f"{label}/")
else:
skipped.append(f"{label}/")
# Copy templates
template_map = [
("SCHEMA.md.template", wiki / "SCHEMA.md"),
("index.md.template", wiki / "index.md"),
("log.md.template", wiki / "log.md"),
("page.md.template", wiki / ".page-template.md"),
("ontology.yaml.template", wiki / "graph" / "ontology.yaml"),
("graph_README.md.template", wiki / "graph" / "README.md"),
("graph_gitignore.template", wiki / "graph" / ".gitignore"),
]
for src_name, dst in template_map:
src = TEMPLATES / src_name
if not src.exists():
print(f"Warning: template missing: {src}", file=sys.stderr)
continue
if copy_template(src, dst):
created.append(str(dst.relative_to(project_root)))
else:
skipped.append(str(dst.relative_to(project_root)))
# Report
if created:
print("Created:")
for path in created:
print(f" + {path}")
if skipped:
print("Already existed (skipped):")
for path in skipped:
print(f" = {path}")
if upgrade:
gaps = detect_schema_gaps(wiki / "SCHEMA.md")
if gaps:
print_schema_upgrade_guidance(wiki / "SCHEMA.md", gaps)
else:
print()
print("SCHEMA.md is up to date with the current template — no manual merge needed.")
return
print()
print("Next steps:")
print(f" 1. Read {wiki_dir}/SCHEMA.md and customize it for your domain.")
print(f" 2. (Optional) Edit {wiki_dir}/graph/ontology.yaml to add domain-specific predicates.")
print(f" 3. Drop your first source into {raw_dir}/.")
print(f" 4. Ask Claude to ingest it.")
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("project_root", type=Path, help="Project root directory.")
parser.add_argument("--wiki-dir", default="wiki", help="Name of the wiki subdirectory (default: wiki).")
parser.add_argument("--raw-dir", default="raw", help="Name of the raw sources subdirectory (default: raw).")
parser.add_argument("--upgrade", action="store_true",
help="Upgrade an existing wiki: add missing files idempotently and surface SCHEMA.md sections to merge by hand.")
args = parser.parse_args()
init_wiki(args.project_root, args.wiki_dir, args.raw_dir, upgrade=args.upgrade)
if __name__ == "__main__":
main()

169
skills/llm-wiki/scripts/wiki_compile.py Executable file
View File

@@ -0,0 +1,169 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["nanobot-ai"]
# ///
"""wiki_compile.py — dávkový compile nasbíraných zdrojů z cml/raw/ do cml/wiki/.
Spouštěn systémovým cronem každou minutu. Capture (interaktivní) hází zdroje do
cml/raw/ a hned potvrdí; těžký raw→wiki compile (čtení zdrojů, psaní stránek,
rozhodování) běží mimo interaktivní tah přes LLM agenta — tady, na pozadí.
Tok:
1. Levná pre-kontrola (BEZ LLM): jsou v cml/raw/ nezpracované zdroje
(regulérní soubory mimo _done/, _hard/, assets/)? Žádné → exit 0, agenta
vůbec neinstancuj.
2. Lockfile (cml/.compile.lock, PID + start-timestamp): běží jiný compile?
→ exit 0 (neduplikovat). Stale lock (mrtvý proces / > STALE_SECONDS) se
přebere, ať se to nezasekne po pádu.
3. Jinak Nanobot.from_config() + bot.run(<drain goal>) — vyprázdní VŠECHNO
nasbírané v jednom dávkovém běhu (jeden index/graph update pro víc zdrojů).
4. Tiše: jen append do log/wiki_compile_cron.log; žádný Telegram.
Vzor = skills/detach/scripts/tasks-daemon.py (shebang uv run, deps nanobot-ai,
Nanobot.from_config + asyncio.wait_for(bot.run(...), timeout)).
"""
import asyncio
import json
import os
import sys
import traceback
from datetime import datetime
from pathlib import Path
# Skript žije v workspace/skills/llm-wiki/scripts/ → parents[3] = workspace.
WORKSPACE = Path(__file__).resolve().parents[3]
CML = WORKSPACE / "cml"
RAW = CML / "raw"
LOCK = CML / ".compile.lock"
LOG = WORKSPACE / "log" / "wiki_compile_cron.log"
# Podadresáře v raw/, které NEjsou pending zdroje.
RESERVED_DIRS = {"_done", "_hard", "assets"}
TIMEOUT_SECONDS = 25 * 60
STALE_SECONDS = 30 * 60
DRAIN_GOAL = (
"Pomocí skillu llm-wiki (operace Compile/drain) zkompiluj VŠECHNY nezpracované zdroje "
"v `cml/raw/` (regulérní soubory přímo v `cml/raw/`, mimo `_done/`, `_hard/`, `assets/`) "
"do wiki v `cml/wiki/`. Pro každý zdroj proveď plný ingest podle "
"references/ingest-workflow.md: source/entity/concept stránky s frontmatterem a `[[odkazy]]`, "
"aktualizuj `cml/wiki/index.md` a `cml/wiki/log.md`. Po zpracování všech zdrojů regeneruj graph "
"(`wiki_graph_lint.py` + `wiki_graph_extract.py` na `cml/wiki/`). Každý úspěšně zpracovaný "
"zdroj přesuň do `cml/raw/_done/`. Ambiguózní/konfliktní zdroj NEcompiluj natvrdo — nech ho "
"v `cml/raw/` (nebo přesuň do `cml/raw/_hard/`) a důvod zaznamenej do `cml/wiki/log.md`. "
"Lint je report-only: žádné destruktivní úpravy existujících stránek bez potvrzení. "
"Idempotence: pokud pro zdroj už stránky existují (byl zkompilován dřív, jen nepřesunut), "
"NEcykluj reconciliací — ber ho jako hotový, přesuň raw soubor do `cml/raw/_done/` a pokračuj. "
"Každý vyřízený zdroj VŽDY přesuň z `cml/raw/` pryč, ať ho příští cron tik nezpracovává znovu. "
"Běžíš v izolované session na pozadí, bez interakce s uživatelem."
)
def log(message: str) -> None:
LOG.parent.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().astimezone().isoformat(timespec="seconds")
with LOG.open("a", encoding="utf-8") as handle:
handle.write(f"{stamp} {message}\n")
def pending_sources() -> list[Path]:
"""Regulérní soubory přímo v cml/raw/ (mimo skryté a rezervované podadresáře)."""
if not RAW.exists():
return []
return [p for p in sorted(RAW.iterdir()) if p.is_file() and not p.name.startswith(".")]
def _pid_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def _lock_is_stale() -> bool:
"""Lock je mrtvý, když ho nelze přečíst, proces neběží, nebo je starší než STALE_SECONDS."""
try:
data = json.loads(LOCK.read_text())
pid = int(data["pid"])
started = datetime.fromisoformat(data["started"])
except (OSError, ValueError, KeyError):
return True
if not _pid_alive(pid):
return True
age = (datetime.now().astimezone() - started).total_seconds()
return age > STALE_SECONDS
def acquire_lock() -> bool:
"""Atomicky vytvoř lock. Vrať False, když už běží živý compile."""
for _ in range(2):
try:
fd = os.open(LOCK, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
except FileExistsError:
if not _lock_is_stale():
return False
log("stale lock, reclaiming")
LOCK.unlink(missing_ok=True)
continue
payload = {"pid": os.getpid(), "started": datetime.now().astimezone().isoformat()}
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(payload, handle)
return True
return False
async def run_compile(goal: str) -> str:
# Heavy import deferred: the per-minute pre-check (no pending work) must not
# pay the nanobot import cost — only an actual compile run needs it.
from nanobot import Nanobot
bot = Nanobot.from_config()
result = await bot.run(goal, session_key="wiki-compile")
return result.content or ""
def main() -> int:
dry_run = "--dry-run" in sys.argv[1:]
pending = pending_sources()
if not pending:
return 0
if not acquire_lock():
log(f"SKIP compile already running ({len(pending)} pending)")
return 0
if dry_run:
names = ", ".join(p.name for p in pending)
log(f"DRY-RUN would compile {len(pending)} pending: {names}")
LOCK.unlink(missing_ok=True)
return 0
started = datetime.now().astimezone()
log(f"START compile {len(pending)} pending: {', '.join(p.name for p in pending)}")
try:
result_text = asyncio.run(
asyncio.wait_for(run_compile(DRAIN_GOAL), timeout=TIMEOUT_SECONDS)
)
summary = result_text.strip().splitlines()[0][:200] if result_text.strip() else "(prázdný výstup)"
duration = int((datetime.now().astimezone() - started).total_seconds())
log(f"END compile duration={duration}s remaining={len(pending_sources())} :: {summary}")
return 0
except asyncio.TimeoutError:
log(f"TIMEOUT compile po {TIMEOUT_SECONDS // 60} min")
return 1
except Exception as error:
log(f"EXCEPTION compile: {error}\n{traceback.format_exc()}")
return 1
finally:
LOCK.unlink(missing_ok=True)
if __name__ == "__main__":
sys.exit(main())

541
skills/llm-wiki/scripts/wiki_graph_extract.py Normal file
View File

@@ -0,0 +1,541 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["pyyaml"]
# ///
"""
wiki_graph_extract.py — Compile the markdown wiki into a queryable graph.
Markdown remains canonical. This script reads every wiki page, derives nodes
and edges (typed semantic edges from `graph.relationships`, plus implicit
`mentions`, `sourced_from`, `summarizes_raw` edges), and emits artifacts under
`<wiki>/graph/` that can be deleted and rebuilt at any time.
Requires PyYAML (`pip install pyyaml`) — the new graph layer uses real YAML
parsing for its nested frontmatter, unlike the stdlib-only lint/search/stats
scripts.
Usage:
python wiki_graph_extract.py <wiki-dir> [options]
Options:
--out <dir> Output directory (default: <wiki-dir>/graph)
--formats jsonl,sqlite,... Comma-list of formats to emit
(jsonl, sqlite, graphml; default: all three)
--ontology <path> Override ontology path
(default: <wiki-dir>/graph/ontology.yaml)
Examples:
python wiki_graph_extract.py wiki/
python wiki_graph_extract.py wiki/ --out wiki/graph --formats jsonl,sqlite
"""
import argparse
import hashlib
import json
import re
import sqlite3
import sys
import xml.etree.ElementTree as ET
from collections import defaultdict
from pathlib import Path
try:
import yaml
except ImportError:
print(
"wiki_graph_extract.py requires PyYAML.\n"
"Install with: pip install pyyaml",
file=sys.stderr,
)
sys.exit(2)
WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]")
FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
SKIP_TOP_LEVEL_FILES = {"SCHEMA.md", "index.md", "log.md", "README.md"}
SKIP_TOP_LEVEL_DIRS = {"indexes", "graph"}
DEFAULT_FORMATS = ["jsonl", "sqlite", "graphml"]
# ---------------------------------------------------------------------------
# Page collection
# ---------------------------------------------------------------------------
def parse_frontmatter(text: str) -> tuple[dict, str]:
"""Extract YAML frontmatter using PyYAML. Returns (meta, body)."""
m = FRONTMATTER_RE.match(text)
if not m:
return {}, text
fm_text = m.group(1)
body = text[m.end():]
try:
meta = yaml.safe_load(fm_text) or {}
except yaml.YAMLError:
meta = {}
if not isinstance(meta, dict):
meta = {}
return meta, body
def collect_pages(wiki_root: Path) -> list[dict]:
pages = []
for md_path in sorted(wiki_root.rglob("*.md")):
rel = md_path.relative_to(wiki_root)
if rel.parts[0] in SKIP_TOP_LEVEL_FILES or rel.parts[0] in SKIP_TOP_LEVEL_DIRS:
continue
if rel.name.startswith("."):
continue
try:
text = md_path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
continue
meta, body = parse_frontmatter(text)
links = [m.group(1).strip() for m in WIKILINK_RE.finditer(body)]
pages.append({
"path": str(md_path),
"rel_path": str(rel).replace("\\", "/"),
"slug": md_path.stem,
"meta": meta,
"body": body,
"links": links,
})
return pages
# ---------------------------------------------------------------------------
# Ontology
# ---------------------------------------------------------------------------
def load_ontology(path: Path) -> dict:
if not path.exists():
return {"node_types": {}, "predicates": {}}
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
except yaml.YAMLError as e:
print(f"Ontology parse error ({path}): {e}", file=sys.stderr)
sys.exit(2)
data.setdefault("node_types", {})
data.setdefault("predicates", {})
return data
def derive_node_type(meta: dict, ontology: dict) -> str | None:
"""Map a page's frontmatter to a node_type using ontology[node_types][*].maps_from."""
page_type = meta.get("type")
page_kind = meta.get("kind")
explicit = (meta.get("graph") or {}).get("node_type") if isinstance(meta.get("graph"), dict) else None
if explicit:
return explicit
# Try (type, kind) match first, then type-only.
type_kind_match = None
type_only_match = None
for nt_name, nt_def in ontology["node_types"].items():
maps = (nt_def or {}).get("maps_from") or {}
m_type = maps.get("type")
m_kind = maps.get("kind")
if m_type and m_type == page_type:
if m_kind and m_kind == page_kind:
type_kind_match = nt_name
break
if not m_kind and type_only_match is None:
type_only_match = nt_name
return type_kind_match or type_only_match
# ---------------------------------------------------------------------------
# Node + edge construction
# ---------------------------------------------------------------------------
def build_nodes(pages: list[dict], ontology: dict) -> tuple[list[dict], dict, list[dict]]:
"""Build the node list + slug→node_id index + alias rows. Returns (nodes, slug_to_id, aliases)."""
nodes: list[dict] = []
slug_to_id: dict[str, str] = {}
aliases: list[dict] = []
seen_ids: set[str] = set()
for p in pages:
meta = p["meta"]
graph_meta = meta.get("graph") if isinstance(meta.get("graph"), dict) else {}
node_type = derive_node_type(meta, ontology) or "concept"
explicit_id = graph_meta.get("node_id")
node_id = explicit_id or f"{node_type}:{p['slug']}"
# Skip duplicates — first one wins; lint will flag this.
if node_id in seen_ids:
continue
seen_ids.add(node_id)
node = {
"id": node_id,
"slug": p["slug"],
"title": meta.get("title") or p["slug"],
"page_type": meta.get("type") or "",
"node_type": node_type,
"kind": meta.get("kind") or "",
"tags": list(meta.get("tags") or []),
"aliases": list(graph_meta.get("aliases") or []),
"path": p["rel_path"],
"created": meta.get("created") or "",
"updated": meta.get("updated") or "",
"canonical": bool(graph_meta.get("canonical", False)),
}
nodes.append(node)
slug_to_id[p["slug"]] = node_id
for alias in node["aliases"]:
aliases.append({"alias": str(alias), "node_id": node_id})
return nodes, slug_to_id, aliases
def edge_id(subject: str, predicate: str, obj: str, source: str | None, evidence: str | None) -> str:
# Truncated to 96 bits — collision risk is negligible at any plausible
# wiki scale and shorter ids keep the JSONL/sqlite/graphml outputs readable.
h = hashlib.sha256()
parts = [subject or "", predicate or "", obj or "", source or "", evidence or ""]
h.update("\x1f".join(parts).encode("utf-8"))
return h.hexdigest()[:24]
def make_edge(*, subject, predicate, obj, source, evidence, confidence, status,
extraction_method, page, extras: dict | None = None) -> dict:
return {
"id": edge_id(subject, predicate, obj, source, evidence),
"subject": subject,
"predicate": predicate,
"object": obj,
"source": source or "",
"evidence": evidence or "",
"confidence": confidence or "",
"status": status or "",
"extraction_method": extraction_method,
"page": page,
"extras": extras or {},
}
def build_edges(pages: list[dict], slug_to_id: dict[str, str]) -> list[dict]:
edges: list[dict] = []
seen_ids: set[str] = set()
def push(edge: dict) -> None:
if edge["id"] in seen_ids:
return
seen_ids.add(edge["id"])
edges.append(edge)
for p in pages:
slug = p["slug"]
subject_id = slug_to_id.get(slug)
if not subject_id:
continue
meta = p["meta"]
graph_meta = meta.get("graph") if isinstance(meta.get("graph"), dict) else {}
# 1. Typed semantic edges from graph.relationships[].
for rel in graph_meta.get("relationships") or []:
if not isinstance(rel, dict):
continue
obj = rel.get("object")
predicate = rel.get("predicate")
if not (obj and predicate):
continue
extras = {
k: rel[k] for k in ("valid_from", "valid_to", "notes", "raw_ref",
"contradicts", "supersedes")
if k in rel and rel[k] is not None
}
push(make_edge(
subject=subject_id,
predicate=str(predicate),
obj=str(obj),
source=rel.get("source"),
evidence=rel.get("evidence"),
confidence=rel.get("confidence"),
status=rel.get("status"),
extraction_method="explicit_graph_frontmatter",
page=p["rel_path"],
extras=extras,
))
# 2. Mentions edges from body wikilinks.
seen_targets: set[str] = set()
for link in p["links"]:
target_slug = link.split("#")[0].strip()
if not target_slug or target_slug == slug:
continue
target_id = slug_to_id.get(target_slug)
if not target_id or target_id in seen_targets:
continue
seen_targets.add(target_id)
push(make_edge(
subject=subject_id,
predicate="mentions",
obj=target_id,
source=None,
evidence=None,
confidence="low",
status="current",
extraction_method="body_wikilink",
page=p["rel_path"],
))
# 3. sourced_from edges from frontmatter `sources:` (skip on source pages themselves).
if meta.get("type") != "source":
for src_slug in meta.get("sources") or []:
src_id = slug_to_id.get(str(src_slug))
if not src_id:
continue
push(make_edge(
subject=subject_id,
predicate="sourced_from",
obj=src_id,
source=str(src_slug),
evidence=None,
confidence="high",
status="current",
extraction_method="frontmatter_sources",
page=p["rel_path"],
))
# 4. summarizes_raw edges from source pages' raw: field.
if meta.get("type") == "source":
raw_path = meta.get("raw")
if raw_path:
push(make_edge(
subject=subject_id,
predicate="summarizes_raw",
obj=f"raw:{raw_path}",
source=None,
evidence=None,
confidence="high",
status="current",
extraction_method="frontmatter_raw",
page=p["rel_path"],
))
return edges
# ---------------------------------------------------------------------------
# Output writers
# ---------------------------------------------------------------------------
def _normalize_for_json(value):
if hasattr(value, "isoformat"):
return value.isoformat()
if isinstance(value, list):
return [_normalize_for_json(v) for v in value]
if isinstance(value, dict):
return {k: _normalize_for_json(v) for k, v in value.items()}
return value
def write_jsonl(out_dir: Path, nodes: list[dict], edges: list[dict]) -> None:
nodes_sorted = sorted(nodes, key=lambda n: n["id"])
edges_sorted = sorted(edges, key=lambda e: e["id"])
with (out_dir / "nodes.jsonl").open("w", encoding="utf-8") as f:
for n in nodes_sorted:
f.write(json.dumps(_normalize_for_json(n), sort_keys=True, ensure_ascii=False))
f.write("\n")
with (out_dir / "edges.jsonl").open("w", encoding="utf-8") as f:
for e in edges_sorted:
f.write(json.dumps(_normalize_for_json(e), sort_keys=True, ensure_ascii=False))
f.write("\n")
def write_sqlite(out_dir: Path, nodes: list[dict], aliases: list[dict], edges: list[dict]) -> None:
db_path = out_dir / "graph.sqlite"
if db_path.exists():
db_path.unlink()
conn = sqlite3.connect(db_path)
try:
conn.executescript("""
CREATE TABLE nodes (
id TEXT PRIMARY KEY,
slug TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
page_type TEXT NOT NULL,
node_type TEXT NOT NULL,
kind TEXT,
path TEXT NOT NULL,
created TEXT,
updated TEXT,
metadata_json TEXT NOT NULL
);
CREATE TABLE aliases (
alias TEXT NOT NULL,
node_id TEXT NOT NULL,
PRIMARY KEY (alias, node_id),
FOREIGN KEY (node_id) REFERENCES nodes(id)
);
CREATE TABLE edges (
id TEXT PRIMARY KEY,
subject TEXT NOT NULL,
predicate TEXT NOT NULL,
object TEXT NOT NULL,
source TEXT,
evidence TEXT,
confidence TEXT,
status TEXT,
extraction_method TEXT NOT NULL,
page TEXT NOT NULL,
metadata_json TEXT NOT NULL
);
CREATE INDEX idx_edges_subject ON edges(subject);
CREATE INDEX idx_edges_object ON edges(object);
CREATE INDEX idx_edges_predicate ON edges(predicate);
CREATE INDEX idx_edges_source ON edges(source);
""")
for n in sorted(nodes, key=lambda n: n["id"]):
metadata_json = json.dumps(_normalize_for_json({
"tags": n.get("tags", []),
"aliases": n.get("aliases", []),
"canonical": n.get("canonical", False),
}), sort_keys=True, ensure_ascii=False)
conn.execute(
"INSERT INTO nodes (id, slug, title, page_type, node_type, kind, path, created, updated, metadata_json) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
n["id"], n["slug"], n["title"], n["page_type"], n["node_type"],
n.get("kind") or None, n["path"],
str(n.get("created") or "") or None,
str(n.get("updated") or "") or None,
metadata_json,
),
)
for a in sorted(aliases, key=lambda a: (a["alias"], a["node_id"])):
conn.execute(
"INSERT OR IGNORE INTO aliases (alias, node_id) VALUES (?, ?)",
(a["alias"], a["node_id"]),
)
for e in sorted(edges, key=lambda e: e["id"]):
metadata_json = json.dumps(_normalize_for_json(e.get("extras") or {}),
sort_keys=True, ensure_ascii=False)
conn.execute(
"INSERT INTO edges (id, subject, predicate, object, source, evidence, "
"confidence, status, extraction_method, page, metadata_json) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
e["id"], e["subject"], e["predicate"], e["object"],
e.get("source") or None, e.get("evidence") or None,
e.get("confidence") or None, e.get("status") or None,
e["extraction_method"], e["page"], metadata_json,
),
)
conn.commit()
finally:
conn.close()
def write_graphml(out_dir: Path, nodes: list[dict], edges: list[dict]) -> None:
ns = "http://graphml.graphdrawing.org/xmlns"
ET.register_namespace("", ns)
root = ET.Element(f"{{{ns}}}graphml")
keys = [
("d_title", "node", "title", "string"),
("d_node_type", "node", "node_type", "string"),
("d_page_type", "node", "page_type", "string"),
("d_path", "node", "path", "string"),
("d_predicate", "edge", "predicate", "string"),
("d_confidence", "edge", "confidence", "string"),
("d_status", "edge", "status", "string"),
("d_source", "edge", "source", "string"),
]
for kid, kfor, kname, ktype in keys:
k = ET.SubElement(root, f"{{{ns}}}key")
k.set("id", kid)
k.set("for", kfor)
k.set("attr.name", kname)
k.set("attr.type", ktype)
graph = ET.SubElement(root, f"{{{ns}}}graph")
graph.set("id", "wiki")
graph.set("edgedefault", "directed")
for n in sorted(nodes, key=lambda n: n["id"]):
node_el = ET.SubElement(graph, f"{{{ns}}}node")
node_el.set("id", n["id"])
for kid, kfor, kname, _ in keys:
if kfor != "node":
continue
data = ET.SubElement(node_el, f"{{{ns}}}data")
data.set("key", kid)
data.text = str(n.get(kname) or "")
for e in sorted(edges, key=lambda e: e["id"]):
edge_el = ET.SubElement(graph, f"{{{ns}}}edge")
edge_el.set("id", e["id"])
edge_el.set("source", e["subject"])
edge_el.set("target", e["object"])
for kid, kfor, kname, _ in keys:
if kfor != "edge":
continue
data = ET.SubElement(edge_el, f"{{{ns}}}data")
data.set("key", kid)
data.text = str(e.get(kname) or "")
tree = ET.ElementTree(root)
ET.indent(tree, space=" ")
tree.write(out_dir / "graph.graphml", encoding="utf-8", xml_declaration=True)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("wiki", type=Path, help="Wiki directory.")
parser.add_argument("--out", type=Path, help="Output directory (default: <wiki>/graph)")
parser.add_argument("--formats", default=",".join(DEFAULT_FORMATS),
help="Comma-list: jsonl, sqlite, graphml")
parser.add_argument("--ontology", type=Path, help="Ontology file (default: <wiki>/graph/ontology.yaml)")
args = parser.parse_args()
if not args.wiki.exists():
print(f"Wiki directory not found: {args.wiki}", file=sys.stderr)
sys.exit(1)
out_dir = args.out or (args.wiki / "graph")
out_dir.mkdir(parents=True, exist_ok=True)
ontology_path = args.ontology or (args.wiki / "graph" / "ontology.yaml")
ontology = load_ontology(ontology_path)
formats = [f.strip().lower() for f in args.formats.split(",") if f.strip()]
unknown = [f for f in formats if f not in DEFAULT_FORMATS]
if unknown:
print(f"Unknown formats: {unknown}. Allowed: {DEFAULT_FORMATS}", file=sys.stderr)
sys.exit(1)
pages = collect_pages(args.wiki)
nodes, slug_to_id, aliases = build_nodes(pages, ontology)
edges = build_edges(pages, slug_to_id)
if "jsonl" in formats:
write_jsonl(out_dir, nodes, edges)
if "sqlite" in formats:
write_sqlite(out_dir, nodes, aliases, edges)
if "graphml" in formats:
write_graphml(out_dir, nodes, edges)
print(f"Extracted {len(nodes)} nodes, {len(edges)} edges → {out_dir}")
breakdown = defaultdict(int)
for e in edges:
breakdown[e["predicate"]] += 1
for pred, count in sorted(breakdown.items(), key=lambda x: (-x[1], x[0])):
print(f" {pred:20s} {count}")
if __name__ == "__main__":
main()

418
skills/llm-wiki/scripts/wiki_graph_lint.py Normal file
View File

@@ -0,0 +1,418 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["pyyaml"]
# ///
"""
wiki_graph_lint.py — Validate the typed graph metadata in a wiki.
Reads every page's `graph:` frontmatter, cross-checks against the ontology
(`wiki/graph/ontology.yaml`), and reports problems. Conservative by design:
reports only, never edits.
Requires PyYAML (`pip install pyyaml`).
Checks:
- Unique `graph.node_id` values across pages.
- All relationship `object` ids resolve to known nodes (or are allowed
string-literal targets for predicates whose object_types include "*").
- All predicates exist in `graph/ontology.yaml`.
- Predicate subject/object types match ontology.
- Typed semantic edges (anything except mentions/sourced_from/summarizes_raw
and predicates with `requires_evidence: false`) carry `source` and
`evidence`.
- `source` references resolve to an existing source page.
- `confidence` is one of high|medium|low; `status` is one of
current|historical|proposed|disputed|superseded.
- No duplicate canonical nodes for the same node id.
- Aliases do not collide across distinct canonical nodes.
- `contradicts` / `supersedes` references resolve to known node/edge ids.
- Generated graph has no orphan typed nodes (nodes with no inbound or
outbound typed edges) except for `source` nodes (allowed source-only).
Usage:
python wiki_graph_lint.py [<wiki-dir>] [--json]
"""
import argparse
import json
import re
import sys
from collections import defaultdict
from pathlib import Path
try:
import yaml
except ImportError:
print(
"wiki_graph_lint.py requires PyYAML.\n"
"Install with: pip install pyyaml",
file=sys.stderr,
)
sys.exit(2)
# Same module is imported by extract; we re-use its build_nodes/build_edges to
# guarantee lint sees exactly what extract would emit.
SCRIPT_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(SCRIPT_DIR))
import wiki_graph_extract as _extract # noqa: E402
FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]")
SKIP_TOP_LEVEL_FILES = {"SCHEMA.md", "index.md", "log.md", "README.md"}
SKIP_TOP_LEVEL_DIRS = {"indexes", "graph"}
ALLOWED_CONFIDENCE = {"high", "medium", "low"}
ALLOWED_STATUS = {"current", "historical", "proposed", "disputed", "superseded"}
IMPLICIT_PREDICATES = {"mentions", "sourced_from", "summarizes_raw"}
def parse_frontmatter(text: str) -> tuple[dict, str]:
m = FRONTMATTER_RE.match(text)
if not m:
return {}, text
fm_text = m.group(1)
body = text[m.end():]
try:
meta = yaml.safe_load(fm_text) or {}
except yaml.YAMLError:
meta = {}
if not isinstance(meta, dict):
meta = {}
return meta, body
def collect_pages(wiki_root: Path) -> list[dict]:
pages = []
for md_path in sorted(wiki_root.rglob("*.md")):
rel = md_path.relative_to(wiki_root)
if rel.parts[0] in SKIP_TOP_LEVEL_FILES or rel.parts[0] in SKIP_TOP_LEVEL_DIRS:
continue
if rel.name.startswith("."):
continue
try:
text = md_path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
continue
meta, body = parse_frontmatter(text)
pages.append({
"path": str(md_path),
"rel_path": str(rel).replace("\\", "/"),
"slug": md_path.stem,
"meta": meta,
"body": body,
"links": [m.group(1).strip() for m in WIKILINK_RE.finditer(body)],
})
return pages
def derive_node_type(meta: dict, ontology: dict) -> str | None:
page_type = meta.get("type")
page_kind = meta.get("kind")
explicit = (meta.get("graph") or {}).get("node_type") if isinstance(meta.get("graph"), dict) else None
if explicit:
return explicit
type_kind_match = None
type_only_match = None
for nt_name, nt_def in ontology.get("node_types", {}).items():
maps = (nt_def or {}).get("maps_from") or {}
m_type = maps.get("type")
m_kind = maps.get("kind")
if m_type and m_type == page_type:
if m_kind and m_kind == page_kind:
type_kind_match = nt_name
break
if not m_kind and type_only_match is None:
type_only_match = nt_name
return type_kind_match or type_only_match
def derive_node_id(meta: dict, slug: str, ontology: dict) -> str:
graph_meta = meta.get("graph") if isinstance(meta.get("graph"), dict) else {}
explicit = graph_meta.get("node_id")
if explicit:
return str(explicit)
node_type = derive_node_type(meta, ontology) or "concept"
return f"{node_type}:{slug}"
def types_match(allowed: list[str] | None, actual: str | None) -> bool:
if not allowed:
return True
if "*" in allowed:
return True
return actual in allowed
def lint(pages: list[dict], ontology: dict) -> dict:
findings = {
"duplicate_node_ids": [],
"unknown_predicates": [],
"broken_object_refs": [],
"subject_type_mismatch": [],
"object_type_mismatch": [],
"missing_evidence": [],
"missing_source_field": [],
"broken_source_refs": [],
"invalid_confidence": [],
"invalid_status": [],
"duplicate_canonical": [],
"alias_collisions": [],
"broken_contradicts": [],
"broken_supersedes": [],
"orphan_typed_nodes": [],
"summary": {},
}
predicates = ontology.get("predicates", {})
node_types = ontology.get("node_types", {})
# Build node index
node_by_id: dict[str, dict] = {}
duplicates: dict[str, list[str]] = defaultdict(list)
for p in pages:
nid = derive_node_id(p["meta"], p["slug"], ontology)
if nid in node_by_id:
duplicates[nid].append(p["rel_path"])
duplicates[nid].append(node_by_id[nid]["rel_path"])
continue
node_type = derive_node_type(p["meta"], ontology) or "concept"
graph_meta = p["meta"].get("graph") if isinstance(p["meta"].get("graph"), dict) else {}
node_by_id[nid] = {
"id": nid,
"node_type": node_type,
"rel_path": p["rel_path"],
"slug": p["slug"],
"page_type": p["meta"].get("type"),
"canonical": bool(graph_meta.get("canonical", False)),
"aliases": list(graph_meta.get("aliases") or []),
"graph": graph_meta,
}
for nid, paths in duplicates.items():
findings["duplicate_node_ids"].append({"node_id": nid, "paths": sorted(set(paths))})
# Source pages by slug — used to validate `source:` refs on edges.
source_slugs = {p["slug"] for p in pages if p["meta"].get("type") == "source"}
# Aliases
alias_to_canonicals: dict[str, set[str]] = defaultdict(set)
canonical_by_id: dict[str, list[str]] = defaultdict(list)
for n in node_by_id.values():
if n["canonical"]:
canonical_by_id[n["id"]].append(n["rel_path"])
for alias in n["aliases"]:
alias_to_canonicals[str(alias)].add(n["id"])
for nid, paths in canonical_by_id.items():
if len(paths) > 1:
findings["duplicate_canonical"].append({"node_id": nid, "paths": paths})
for alias, owners in alias_to_canonicals.items():
if len(owners) > 1:
findings["alias_collisions"].append({"alias": alias, "owners": sorted(owners)})
# Walk relationships
for p in pages:
graph_meta = p["meta"].get("graph") if isinstance(p["meta"].get("graph"), dict) else {}
subject_id = derive_node_id(p["meta"], p["slug"], ontology)
subject_type = node_by_id.get(subject_id, {}).get("node_type")
for idx, rel in enumerate(graph_meta.get("relationships") or []):
if not isinstance(rel, dict):
continue
predicate = rel.get("predicate")
obj = rel.get("object")
here = {"page": p["rel_path"], "predicate": predicate,
"object": obj, "index": idx}
if not predicate or predicate not in predicates:
findings["unknown_predicates"].append({**here})
continue
pdef = predicates[predicate] or {}
# Object resolution. Allow string-literal objects only when
# ontology lists "*" in object_types (e.g. summarizes_raw).
object_types = pdef.get("object_types") or []
allows_wildcard_obj = "*" in object_types
if obj and obj not in node_by_id:
if not allows_wildcard_obj:
findings["broken_object_refs"].append({**here})
# Subject type check
if not types_match(pdef.get("subject_types"), subject_type):
findings["subject_type_mismatch"].append({
**here,
"subject": subject_id,
"subject_type": subject_type,
"allowed": pdef.get("subject_types"),
})
# Object type check (only if object resolves to a node)
obj_node = node_by_id.get(obj) if obj else None
obj_type = obj_node["node_type"] if obj_node else None
if obj_node and not types_match(pdef.get("object_types"), obj_type):
findings["object_type_mismatch"].append({
**here,
"object_type": obj_type,
"allowed": pdef.get("object_types"),
})
requires_evidence = pdef.get("requires_evidence", True)
if requires_evidence:
if not rel.get("evidence"):
findings["missing_evidence"].append({**here})
if not rel.get("source"):
findings["missing_source_field"].append({**here})
# source field must reference an existing source page slug
src = rel.get("source")
if src and str(src) not in source_slugs:
findings["broken_source_refs"].append({**here, "source": src})
confidence = rel.get("confidence")
if confidence and confidence not in ALLOWED_CONFIDENCE:
findings["invalid_confidence"].append({**here, "confidence": confidence})
status = rel.get("status")
if status and status not in ALLOWED_STATUS:
findings["invalid_status"].append({**here, "status": status})
# contradicts / supersedes resolution
for ref_field, bucket in (("contradicts", "broken_contradicts"),
("supersedes", "broken_supersedes")):
ref = rel.get(ref_field)
if ref:
ref_str = str(ref)
if ref_str not in node_by_id and ref_str not in source_slugs:
findings[bucket].append({**here, ref_field: ref_str})
# Orphan typed nodes — pages that declared `graph:` frontmatter but end
# up with no typed (non-implicit) edge touching them after extraction.
# Source nodes are exempt (they participate via implicit edges).
extracted_edges = _extract.build_edges(pages, {n["slug"]: n["id"] for n in node_by_id.values()})
typed_node_refs: set[str] = set()
for e in extracted_edges:
if e["predicate"] in IMPLICIT_PREDICATES:
continue
typed_node_refs.add(e["subject"])
if e["object"] in node_by_id:
typed_node_refs.add(e["object"])
for n in node_by_id.values():
if n["node_type"] == "source":
continue
graph_meta = n.get("graph") or {}
if not graph_meta:
continue # Pages without graph metadata are valid; they're text-only nodes.
if n["id"] in typed_node_refs:
continue
findings["orphan_typed_nodes"].append({
"node_id": n["id"],
"path": n["rel_path"],
})
# Summary
findings["summary"] = {
"pages_scanned": len(pages),
"nodes": len(node_by_id),
**{k: len(v) for k, v in findings.items() if isinstance(v, list)},
}
return findings
def render_text(findings: dict) -> str:
out = []
s = findings["summary"]
out.append("=" * 60)
out.append("Wiki Graph Lint Report")
out.append("=" * 60)
out.append(f"Pages scanned: {s['pages_scanned']} Nodes: {s['nodes']}")
out.append("")
sections = [
("duplicate_node_ids", "Duplicate node ids",
lambda f: f" - {f['node_id']}: {', '.join(f['paths'])}"),
("unknown_predicates", "Unknown predicates (not in ontology)",
lambda f: f" - {f['page']}#rel[{f['index']}] predicate={f['predicate']!r}"),
("broken_object_refs", "Broken object references",
lambda f: f" - {f['page']}#rel[{f['index']}] {f['predicate']} → {f['object']!r}"),
("subject_type_mismatch", "Subject type does not match ontology",
lambda f: f" - {f['page']}#rel[{f['index']}] {f['predicate']}: subject={f['subject_type']} (allowed: {f['allowed']})"),
("object_type_mismatch", "Object type does not match ontology",
lambda f: f" - {f['page']}#rel[{f['index']}] {f['predicate']}: object={f['object_type']} (allowed: {f['allowed']})"),
("missing_evidence", "Missing evidence on typed edge",
lambda f: f" - {f['page']}#rel[{f['index']}] {f['predicate']} → {f['object']}"),
("missing_source_field", "Missing source on typed edge",
lambda f: f" - {f['page']}#rel[{f['index']}] {f['predicate']} → {f['object']}"),
("broken_source_refs", "source: does not match any source page",
lambda f: f" - {f['page']}#rel[{f['index']}] source={f['source']!r}"),
("invalid_confidence", "Invalid confidence value",
lambda f: f" - {f['page']}#rel[{f['index']}] confidence={f['confidence']!r}"),
("invalid_status", "Invalid status value",
lambda f: f" - {f['page']}#rel[{f['index']}] status={f['status']!r}"),
("duplicate_canonical", "Duplicate canonical nodes",
lambda f: f" - {f['node_id']}: {', '.join(f['paths'])}"),
("alias_collisions", "Alias used by multiple canonical nodes",
lambda f: f" - {f['alias']!r}: {', '.join(f['owners'])}"),
("broken_contradicts", "Broken contradicts reference",
lambda f: f" - {f['page']}#rel[{f['index']}] contradicts={f.get('contradicts')}"),
("broken_supersedes", "Broken supersedes reference",
lambda f: f" - {f['page']}#rel[{f['index']}] supersedes={f.get('supersedes')}"),
("orphan_typed_nodes", "Orphan typed nodes (no inbound or outbound typed edges)",
lambda f: f" - {f['node_id']} ({f['path']})"),
]
healthy = True
for key, label, formatter in sections:
items = findings[key]
if not items:
continue
healthy = False
out.append(f"{label} ({len(items)}):")
for item in items[:50]:
out.append(formatter(item))
if len(items) > 50:
out.append(f" ... and {len(items) - 50} more")
out.append("")
if healthy:
out.append("No graph issues found.")
return "\n".join(out)
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("wiki", nargs="?", type=Path, default=Path("cml/wiki"))
parser.add_argument("--ontology", type=Path, help="Ontology file (default: <wiki>/graph/ontology.yaml)")
parser.add_argument("--json", action="store_true")
args = parser.parse_args()
if not args.wiki.exists():
print(f"Wiki directory not found: {args.wiki}", file=sys.stderr)
sys.exit(1)
ontology_path = args.ontology or (args.wiki / "graph" / "ontology.yaml")
if not ontology_path.exists():
print(f"Ontology not found: {ontology_path}", file=sys.stderr)
print("Did you forget to seed wiki/graph/ontology.yaml? See assets/ontology.yaml.template.",
file=sys.stderr)
sys.exit(1)
try:
ontology = yaml.safe_load(ontology_path.read_text(encoding="utf-8")) or {}
except yaml.YAMLError as e:
print(f"Ontology parse error: {e}", file=sys.stderr)
sys.exit(2)
pages = collect_pages(args.wiki)
findings = lint(pages, ontology)
if args.json:
print(json.dumps(findings, indent=2, default=str))
else:
print(render_text(findings))
if __name__ == "__main__":
main()

267
skills/llm-wiki/scripts/wiki_graph_query.py Normal file
View File

@@ -0,0 +1,267 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
wiki_graph_query.py — Query the compiled wiki graph (graph.sqlite).
Use this to accelerate navigation: find what's connected to a node, list
typed edges around a subject, find a path between two nodes, or dump every
fact about a node. The graph is a navigation index — for high-stakes
claims, follow the `source` field back to the wiki page and the raw file.
Subcommands:
neighbors --node <id> List nodes one hop away from <id>
edges --subject <id> List all outbound edges from <id>
[--predicate <p>] Filter by predicate
path --from <id> --to <id> Shortest directed path (BFS, max depth 6)
[--max-depth N]
facts --about <id> Outbound + inbound edges for <id>
Common options:
--db <path> Path to graph.sqlite (default: <wiki>/graph/graph.sqlite)
--json Emit JSON instead of text
Examples:
python wiki_graph_query.py wiki/ neighbors --node product:konvy
python wiki_graph_query.py wiki/ edges --subject person:stephanie-emmanouel
python wiki_graph_query.py wiki/ path --from person:praney-behl --to product:konvy
python wiki_graph_query.py wiki/ facts --about product:konvy
"""
import argparse
import json
import sqlite3
import sys
from collections import deque
from pathlib import Path
EVIDENCE_SNIPPET_LEN = 140
def open_db(path: Path) -> sqlite3.Connection:
if not path.exists():
print(f"graph.sqlite not found at {path}.", file=sys.stderr)
print("Run wiki_graph_extract.py first.", file=sys.stderr)
sys.exit(1)
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
return conn
def fetch_node(conn: sqlite3.Connection, node_id: str) -> dict | None:
row = conn.execute("SELECT * FROM nodes WHERE id = ?", (node_id,)).fetchone()
return dict(row) if row else None
def edges_from(conn: sqlite3.Connection, subject: str, predicate: str | None = None) -> list[dict]:
q = "SELECT * FROM edges WHERE subject = ?"
params: list = [subject]
if predicate:
q += " AND predicate = ?"
params.append(predicate)
q += " ORDER BY predicate, object"
return [dict(r) for r in conn.execute(q, params).fetchall()]
def edges_to(conn: sqlite3.Connection, obj: str, predicate: str | None = None) -> list[dict]:
q = "SELECT * FROM edges WHERE object = ?"
params: list = [obj]
if predicate:
q += " AND predicate = ?"
params.append(predicate)
q += " ORDER BY predicate, subject"
return [dict(r) for r in conn.execute(q, params).fetchall()]
def truncate(text: str | None) -> str:
if not text:
return ""
if len(text) <= EVIDENCE_SNIPPET_LEN:
return text
return text[: EVIDENCE_SNIPPET_LEN - 1].rstrip() + "…"
def render_edge_row(e: dict) -> str:
pieces = [
f" {e['subject']} --[{e['predicate']}]--> {e['object']}",
]
confidence = e.get("confidence") or "-"
status = e.get("status") or "-"
src = e.get("source") or "-"
pieces.append(f" via {src} conf={confidence} status={status}")
if e.get("evidence"):
pieces.append(f" evidence: {truncate(e['evidence'])}")
pieces.append(f" (page: {e['page']})")
return "\n".join(pieces)
def cmd_neighbors(conn: sqlite3.Connection, args) -> dict:
node = fetch_node(conn, args.node)
if not node:
print(f"node not found: {args.node}", file=sys.stderr)
sys.exit(1)
out_edges = edges_from(conn, args.node)
in_edges = edges_to(conn, args.node)
neighbors: dict[str, dict] = {}
for e in out_edges:
neighbors.setdefault(e["object"], {"node_id": e["object"], "out": [], "in": []})
neighbors[e["object"]]["out"].append(e)
for e in in_edges:
neighbors.setdefault(e["subject"], {"node_id": e["subject"], "out": [], "in": []})
neighbors[e["subject"]]["in"].append(e)
# Resolve neighbor titles where possible
for nid, slot in neighbors.items():
target = fetch_node(conn, nid)
slot["title"] = target["title"] if target else nid
slot["path"] = target["path"] if target else None
return {
"node": node,
"neighbors": sorted(neighbors.values(), key=lambda n: n["node_id"]),
}
def cmd_edges(conn: sqlite3.Connection, args) -> dict:
es = edges_from(conn, args.subject, args.predicate)
return {"subject": args.subject, "predicate": args.predicate, "edges": es}
def cmd_facts(conn: sqlite3.Connection, args) -> dict:
node = fetch_node(conn, args.about)
if not node:
print(f"node not found: {args.about}", file=sys.stderr)
sys.exit(1)
return {
"node": node,
"outbound": edges_from(conn, args.about),
"inbound": edges_to(conn, args.about),
}
def cmd_path(conn: sqlite3.Connection, args) -> dict:
src = fetch_node(conn, getattr(args, "from"))
dst = fetch_node(conn, args.to)
if not src:
print(f"from-node not found: {getattr(args, 'from')}", file=sys.stderr)
sys.exit(1)
if not dst:
print(f"to-node not found: {args.to}", file=sys.stderr)
sys.exit(1)
start, goal = getattr(args, "from"), args.to
queue = deque([(start, [start], [])])
visited = {start}
while queue:
node, node_path, edge_path = queue.popleft()
if node == goal:
return {"from": start, "to": goal, "path_nodes": node_path, "path_edges": edge_path}
if len(node_path) - 1 >= args.max_depth:
continue
for e in edges_from(conn, node):
nxt = e["object"]
if nxt in visited:
continue
visited.add(nxt)
queue.append((nxt, node_path + [nxt], edge_path + [e]))
return {"from": start, "to": goal, "path_nodes": [], "path_edges": []}
def render(result: dict, command: str) -> str:
out: list[str] = []
if command == "neighbors":
n = result["node"]
out.append(f"Node: {n['id']} ({n['title']}) {n['node_type']} [{n['path']}]")
out.append(f"Neighbors: {len(result['neighbors'])}")
for nb in result["neighbors"]:
out.append("")
out.append(f" → {nb['node_id']} ({nb['title']})")
for e in nb.get("out", []):
out.append(f" out [{e['predicate']}] conf={e.get('confidence') or '-'} src={e.get('source') or '-'}")
for e in nb.get("in", []):
out.append(f" in [{e['predicate']}] from {e['subject']} src={e.get('source') or '-'}")
elif command == "edges":
out.append(f"Edges from {result['subject']}"
+ (f" with predicate {result['predicate']}" if result['predicate'] else ""))
for e in result["edges"]:
out.append("")
out.append(render_edge_row(e))
elif command == "facts":
n = result["node"]
out.append(f"Facts about {n['id']} ({n['title']}) [{n['path']}]")
out.append("")
out.append(f"Outbound ({len(result['outbound'])}):")
for e in result["outbound"]:
out.append(render_edge_row(e))
out.append("")
out.append(f"Inbound ({len(result['inbound'])}):")
for e in result["inbound"]:
out.append(render_edge_row(e))
elif command == "path":
if not result["path_nodes"]:
out.append(f"No path found from {result['from']} to {result['to']} within depth limit.")
else:
out.append(f"Path from {result['from']} to {result['to']} ({len(result['path_edges'])} hops):")
for e in result["path_edges"]:
out.append("")
out.append(render_edge_row(e))
return "\n".join(out)
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("wiki", type=Path, help="Wiki directory.")
parser.add_argument("--db", type=Path, help="Path to graph.sqlite (default: <wiki>/graph/graph.sqlite)")
parser.add_argument("--json", action="store_true")
sub = parser.add_subparsers(dest="command", required=True)
p_n = sub.add_parser("neighbors")
p_n.add_argument("--node", required=True)
p_e = sub.add_parser("edges")
p_e.add_argument("--subject", required=True)
p_e.add_argument("--predicate")
p_p = sub.add_parser("path")
p_p.add_argument("--from", dest="from", required=True)
p_p.add_argument("--to", required=True)
p_p.add_argument("--max-depth", type=int, default=6)
p_f = sub.add_parser("facts")
p_f.add_argument("--about", required=True)
args = parser.parse_args()
if not args.wiki.exists():
print(f"Wiki directory not found: {args.wiki}", file=sys.stderr)
sys.exit(1)
db_path = args.db or (args.wiki / "graph" / "graph.sqlite")
conn = open_db(db_path)
try:
if args.command == "neighbors":
result = cmd_neighbors(conn, args)
elif args.command == "edges":
result = cmd_edges(conn, args)
elif args.command == "path":
result = cmd_path(conn, args)
elif args.command == "facts":
result = cmd_facts(conn, args)
else:
parser.print_help()
sys.exit(1)
finally:
conn.close()
if args.json:
print(json.dumps(result, indent=2, default=str))
else:
print(render(result, args.command))
if __name__ == "__main__":
main()

319
skills/llm-wiki/scripts/wiki_lint.py Normal file
View File

@@ -0,0 +1,319 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
wiki_lint.py — Structural health check for an LLM Wiki.
Reports orphan pages, broken wikilinks, oversized pages, frontmatter issues,
stale pages, duplicate slugs, and (with --suggest-pages) terms that appear in
many pages without their own page.
Conservative by design: reports findings, never edits.
Usage:
python wiki_lint.py [<wiki-dir>] [options]
Options:
--soft-cap N Page-size soft cap in lines (default: 400)
--hard-cap N Page-size hard cap in lines (default: 800)
--required-fm a,b Required frontmatter fields (default: type,title,tags,created,updated)
--suggest-pages Surface terms appearing in many pages without a page
--suggest-min N Minimum occurrences for --suggest-pages (default: 5)
--json Emit JSON instead of text
Examples:
python wiki_lint.py wiki/
python wiki_lint.py wiki/ --suggest-pages
python wiki_lint.py wiki/ --json > lint.json
"""
import argparse
import json
import re
import sys
from collections import Counter, defaultdict
from datetime import date, datetime
from pathlib import Path
WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]")
FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
CAPITALIZED_PHRASE_RE = re.compile(r"\b([A-Z][a-zA-Z0-9]+(?:\s+[A-Z][a-zA-Z0-9]+){0,3})\b")
SKIP_TOP_LEVEL_FILES = {"SCHEMA.md", "index.md", "log.md", "README.md"}
SKIP_TOP_LEVEL_DIRS = {"indexes", "graph"}
def parse_frontmatter(text: str) -> tuple[dict, str, bool]:
"""Returns (metadata, body, malformed). malformed=True if frontmatter was attempted but unparseable."""
if not text.startswith("---"):
return {}, text, False
m = FRONTMATTER_RE.match(text)
if not m:
return {}, text, True
fm_text = m.group(1)
body = text[m.end():]
meta = {}
current_key = None
for line in fm_text.split("\n"):
if not line.strip():
continue
kv = re.match(r"^([a-zA-Z_]+):\s*(.*)$", line)
if kv:
key, value = kv.group(1), kv.group(2).strip()
if value.startswith("[") and value.endswith("]"):
items = [x.strip().strip('"').strip("'") for x in value[1:-1].split(",") if x.strip()]
meta[key] = items
elif value:
meta[key] = value.strip('"').strip("'")
else:
meta[key] = []
current_key = key
elif line.startswith(" - ") and current_key:
meta[current_key].append(line[4:].strip().strip('"').strip("'"))
return meta, body, False
def collect_pages(wiki_root: Path) -> list[dict]:
pages = []
for md_path in wiki_root.rglob("*.md"):
rel = md_path.relative_to(wiki_root)
if rel.parts[0] in SKIP_TOP_LEVEL_FILES or rel.parts[0] in SKIP_TOP_LEVEL_DIRS:
continue
if rel.name.startswith("."):
continue
try:
text = md_path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError) as e:
pages.append({
"path": str(md_path),
"rel_path": str(rel),
"slug": md_path.stem,
"read_error": str(e),
})
continue
meta, body, malformed = parse_frontmatter(text)
line_count = text.count("\n") + 1
links = [m.group(1).strip() for m in WIKILINK_RE.finditer(body)]
pages.append({
"path": str(md_path),
"rel_path": str(rel),
"slug": md_path.stem,
"meta": meta,
"body": body,
"line_count": line_count,
"links": links,
"malformed_fm": malformed,
})
return pages
def parse_date(s):
if not s or not isinstance(s, str):
return None
try:
return datetime.strptime(s[:10], "%Y-%m-%d").date()
except (ValueError, TypeError):
return None
def lint(pages: list[dict], soft_cap: int, hard_cap: int, required_fm: list[str], suggest_pages: bool, suggest_min: int) -> dict:
findings = {
"orphans": [],
"broken_links": [],
"oversized_hard": [],
"oversized_soft": [],
"missing_frontmatter": [],
"malformed_frontmatter": [],
"duplicate_slugs": [],
"stale_pages": [],
"read_errors": [],
"suggested_pages": [],
"summary": {},
}
# Read errors
for p in pages:
if "read_error" in p:
findings["read_errors"].append({"path": p["rel_path"], "error": p["read_error"]})
pages = [p for p in pages if "read_error" not in p]
# Slugs
slug_to_pages = defaultdict(list)
for p in pages:
slug_to_pages[p["slug"]].append(p["rel_path"])
for slug, paths in slug_to_pages.items():
if len(paths) > 1:
findings["duplicate_slugs"].append({"slug": slug, "paths": paths})
# Inbound link map
inbound = defaultdict(set)
all_slugs = set(slug_to_pages.keys())
for p in pages:
for link in p["links"]:
inbound[link].add(p["slug"])
# Orphans, broken links, oversize, frontmatter, staleness
for p in pages:
# Orphans
if not inbound.get(p["slug"]):
findings["orphans"].append({"slug": p["slug"], "path": p["rel_path"]})
# Broken links
for link in p["links"]:
if link not in all_slugs:
findings["broken_links"].append({
"from": p["slug"],
"from_path": p["rel_path"],
"to": link,
})
# Oversize
if p["line_count"] > hard_cap:
findings["oversized_hard"].append({"path": p["rel_path"], "lines": p["line_count"]})
elif p["line_count"] > soft_cap:
findings["oversized_soft"].append({"path": p["rel_path"], "lines": p["line_count"]})
# Frontmatter
if p["malformed_fm"]:
findings["malformed_frontmatter"].append({"path": p["rel_path"]})
else:
missing = [field for field in required_fm if field not in p["meta"] or p["meta"].get(field) in ("", None, [])]
if missing:
findings["missing_frontmatter"].append({"path": p["rel_path"], "missing": missing})
# Staleness: heuristic — page hasn't been updated in 90 days AND has been touched by recent ingests.
# Approximate: if updated > 90d ago and the page is well-linked (a hub), flag it.
updated = parse_date(p["meta"].get("updated"))
if updated:
age_days = (date.today() - updated).days
if age_days > 90 and len(inbound.get(p["slug"], [])) >= 3:
findings["stale_pages"].append({
"path": p["rel_path"],
"updated": p["meta"].get("updated"),
"age_days": age_days,
"inbound_count": len(inbound.get(p["slug"], [])),
})
# Suggested pages: capitalized multi-word phrases appearing in many pages without a page
if suggest_pages:
phrase_pages = defaultdict(set)
for p in pages:
seen = set()
for m in CAPITALIZED_PHRASE_RE.finditer(p["body"]):
phrase = m.group(1).strip()
seen.add(phrase)
for phrase in seen:
phrase_pages[phrase].add(p["slug"])
# Title set for filtering
existing_titles = {p["meta"].get("title", "").lower() for p in pages}
existing_slugs_normalized = {s.lower().replace("-", " ") for s in all_slugs}
candidates = []
for phrase, page_set in phrase_pages.items():
if len(page_set) < suggest_min:
continue
if phrase.lower() in existing_titles:
continue
if phrase.lower() in existing_slugs_normalized:
continue
# Filter out section header garbage
if phrase.split()[0] in {"Section", "Where", "Sources", "Tags", "Type", "Title"}:
continue
candidates.append({"phrase": phrase, "page_count": len(page_set), "pages": sorted(page_set)[:5]})
candidates.sort(key=lambda x: -x["page_count"])
findings["suggested_pages"] = candidates[:30]
findings["summary"] = {
"total_pages": len(pages),
"orphans": len(findings["orphans"]),
"broken_links": len(findings["broken_links"]),
"oversized_hard": len(findings["oversized_hard"]),
"oversized_soft": len(findings["oversized_soft"]),
"missing_frontmatter": len(findings["missing_frontmatter"]),
"malformed_frontmatter": len(findings["malformed_frontmatter"]),
"duplicate_slugs": len(findings["duplicate_slugs"]),
"stale_pages": len(findings["stale_pages"]),
"read_errors": len(findings["read_errors"]),
"suggested_pages": len(findings["suggested_pages"]),
}
return findings
def render_text(findings: dict) -> str:
out = []
s = findings["summary"]
out.append("=" * 60)
out.append("Wiki Lint Report")
out.append("=" * 60)
out.append(f"Total pages scanned: {s['total_pages']}")
out.append("")
sections = [
("orphans", "Orphan pages (no inbound links)", lambda f: f" - {f['slug']} ({f['path']})"),
("broken_links", "Broken wikilinks", lambda f: f" - [[{f['to']}]] referenced from {f['from_path']}"),
("oversized_hard", "OVERSIZE (over hard cap — must split)", lambda f: f" - {f['path']} ({f['lines']} lines)"),
("oversized_soft", "Oversize (over soft cap — consider splitting)", lambda f: f" - {f['path']} ({f['lines']} lines)"),
("missing_frontmatter", "Missing frontmatter fields", lambda f: f" - {f['path']} missing: {', '.join(f['missing'])}"),
("malformed_frontmatter", "Malformed frontmatter", lambda f: f" - {f['path']}"),
("duplicate_slugs", "Duplicate slugs", lambda f: f" - {f['slug']}: {', '.join(f['paths'])}"),
("stale_pages", "Stale pages (well-linked but not updated in 90+ days)", lambda f: f" - {f['path']} (updated {f['updated']}, {f['age_days']}d ago, {f['inbound_count']} inbound)"),
("read_errors", "Read errors", lambda f: f" - {f['path']}: {f['error']}"),
]
for key, label, formatter in sections:
items = findings[key]
if not items:
continue
out.append(f"{label} ({len(items)}):")
for item in items[:50]:
out.append(formatter(item))
if len(items) > 50:
out.append(f" ... and {len(items) - 50} more")
out.append("")
if findings["suggested_pages"]:
out.append(f"Suggested page candidates ({len(findings['suggested_pages'])}):")
out.append(" Phrases appearing in many pages without a dedicated page:")
for item in findings["suggested_pages"]:
out.append(f" - \"{item['phrase']}\" ({item['page_count']} pages)")
out.append("")
if all(v == 0 for k, v in s.items() if k != "total_pages"):
out.append("No issues found. Wiki is healthy.")
return "\n".join(out)
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("wiki", nargs="?", type=Path, default=Path("cml/wiki"), help="Wiki directory (default: cml/wiki).")
parser.add_argument("--soft-cap", type=int, default=400, help="Page-size soft cap (lines).")
parser.add_argument("--hard-cap", type=int, default=800, help="Page-size hard cap (lines).")
parser.add_argument("--required-fm", default="type,title,tags,created,updated", help="Required frontmatter fields, comma-separated.")
parser.add_argument("--suggest-pages", action="store_true", help="Surface page candidates.")
parser.add_argument("--suggest-min", type=int, default=5, help="Minimum page count for suggestions.")
parser.add_argument("--json", action="store_true", help="Emit JSON.")
args = parser.parse_args()
if not args.wiki.exists():
print(f"Wiki directory not found: {args.wiki}", file=sys.stderr)
sys.exit(1)
pages = collect_pages(args.wiki)
required_fm = [f.strip() for f in args.required_fm.split(",") if f.strip()]
findings = lint(pages, args.soft_cap, args.hard_cap, required_fm, args.suggest_pages, args.suggest_min)
if args.json:
print(json.dumps(findings, indent=2, default=str))
else:
print(render_text(findings))
if __name__ == "__main__":
main()

270
skills/llm-wiki/scripts/wiki_search.py Normal file
View File

@@ -0,0 +1,270 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
wiki_search.py — BM25 search over wiki pages with frontmatter filters.
Fallback for when index-first navigation doesn't surface the right pages.
Pure-Python implementation (no dependencies beyond stdlib) so it runs anywhere.
Usage:
python wiki_search.py "query terms" [options]
Options:
--wiki <dir> Wiki directory (default: cml/wiki)
--top N Return top N results (default: 10)
--type <type> Filter by frontmatter type (source|entity|concept|synthesis|...)
--tag <tag> Filter by tag (repeatable)
--since YYYY-MM-DD Only pages updated on or after this date
--backlinks <slug> Find pages that link to <slug>; ignores the query
--top-linked N Show the N most-linked-to pages (hubs); ignores the query
--cache <path> Persist the BM25 index to disk for faster reruns
Examples:
python wiki_search.py "diffusion training stability" --top 5
python wiki_search.py "alignment" --type concept --tag safety
python wiki_search.py "" --backlinks transformer
python wiki_search.py "" --top-linked 10
"""
import argparse
import json
import math
import re
import sys
from collections import Counter, defaultdict
from datetime import date, datetime
from pathlib import Path
WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]")
FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
TOKEN_RE = re.compile(r"[a-z0-9]+")
def parse_frontmatter(text: str) -> tuple[dict, str]:
"""Lightweight YAML-ish frontmatter parser. Returns (metadata, body)."""
m = FRONTMATTER_RE.match(text)
if not m:
return {}, text
fm_text = m.group(1)
body = text[m.end():]
meta = {}
current_key = None
for line in fm_text.split("\n"):
if not line.strip():
continue
# Inline list: tags: [a, b, c]
kv = re.match(r"^([a-zA-Z_]+):\s*(.*)$", line)
if kv:
key, value = kv.group(1), kv.group(2).strip()
if value.startswith("[") and value.endswith("]"):
items = [x.strip().strip('"').strip("'") for x in value[1:-1].split(",") if x.strip()]
meta[key] = items
elif value:
meta[key] = value.strip('"').strip("'")
else:
meta[key] = []
current_key = key
elif line.startswith(" - ") and current_key:
meta[current_key].append(line[4:].strip().strip('"').strip("'"))
return meta, body
def tokenize(text: str) -> list[str]:
return TOKEN_RE.findall(text.lower())
def slug_from_path(path: Path, wiki_root: Path) -> str:
return path.stem
def extract_wikilinks(body: str) -> list[str]:
return [m.group(1).strip() for m in WIKILINK_RE.finditer(body)]
def collect_pages(wiki_root: Path) -> list[dict]:
"""Walk the wiki and return [{path, slug, meta, body, tokens, links}]."""
pages = []
for md_path in wiki_root.rglob("*.md"):
# Skip the schema, index, log, and template files
rel = md_path.relative_to(wiki_root)
if rel.parts[0] in {"SCHEMA.md", "index.md", "log.md"} or rel.name.startswith("."):
continue
if rel.parts[0] in {"indexes", "graph"}:
continue
try:
text = md_path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
continue
meta, body = parse_frontmatter(text)
pages.append({
"path": str(md_path),
"rel_path": str(rel),
"slug": slug_from_path(md_path, wiki_root),
"meta": meta,
"body": body,
"tokens": tokenize(body + " " + meta.get("title", "")),
"links": extract_wikilinks(body),
})
return pages
def build_bm25(pages: list[dict]) -> dict:
"""Build a BM25 index. Returns {df, avgdl, N, doc_lens, term_freqs}."""
N = len(pages)
df = Counter()
doc_lens = []
term_freqs = []
for page in pages:
tokens = page["tokens"]
doc_lens.append(len(tokens))
tf = Counter(tokens)
term_freqs.append(tf)
for term in tf:
df[term] += 1
avgdl = sum(doc_lens) / N if N else 0
return {"N": N, "df": df, "avgdl": avgdl, "doc_lens": doc_lens, "term_freqs": term_freqs}
def bm25_score(query_tokens: list[str], doc_idx: int, idx: dict, k1: float = 1.5, b: float = 0.75) -> float:
score = 0.0
N = idx["N"]
df = idx["df"]
avgdl = idx["avgdl"]
dl = idx["doc_lens"][doc_idx]
tf = idx["term_freqs"][doc_idx]
for term in query_tokens:
if term not in df:
continue
idf = math.log(1 + (N - df[term] + 0.5) / (df[term] + 0.5))
f = tf.get(term, 0)
if f == 0:
continue
denom = f + k1 * (1 - b + b * (dl / avgdl if avgdl else 1))
score += idf * (f * (k1 + 1)) / denom
return score
def parse_date(s: str | None) -> date | None:
if not s:
return None
try:
return datetime.strptime(s[:10], "%Y-%m-%d").date()
except (ValueError, TypeError):
return None
def passes_filters(page: dict, args) -> bool:
meta = page["meta"]
if args.type and meta.get("type") != args.type:
return False
if args.tag:
page_tags = set(meta.get("tags", []) or [])
if not all(t in page_tags for t in args.tag):
return False
if args.since:
since = parse_date(args.since)
updated = parse_date(meta.get("updated"))
if since and updated and updated < since:
return False
if since and not updated:
return False
return True
def cmd_search(args, pages: list[dict]) -> None:
filtered = [p for p in pages if passes_filters(p, args)]
if not filtered:
print("No pages matched the filters.", file=sys.stderr)
return
idx = build_bm25(filtered)
query_tokens = tokenize(args.query)
if not query_tokens:
print("Empty query.", file=sys.stderr)
return
scored = [(bm25_score(query_tokens, i, idx), i) for i in range(len(filtered))]
scored.sort(key=lambda x: -x[0])
top = [(s, filtered[i]) for s, i in scored[:args.top] if s > 0]
if not top:
print("No matches.", file=sys.stderr)
return
print(f"Top {len(top)} results for: {args.query!r}")
print()
for score, page in top:
title = page["meta"].get("title") or page["slug"]
page_type = page["meta"].get("type", "?")
print(f" [{score:6.2f}] [{page_type:9}] {title}")
print(f" {page['rel_path']}")
def cmd_backlinks(args, pages: list[dict]) -> None:
target = args.backlinks
inbound = []
for page in pages:
if target in page["links"]:
inbound.append(page)
if not inbound:
print(f"No pages link to [[{target}]].", file=sys.stderr)
return
print(f"Pages linking to [[{target}]] ({len(inbound)}):")
for page in inbound:
title = page["meta"].get("title") or page["slug"]
print(f" - {title} ({page['rel_path']})")
def cmd_top_linked(args, pages: list[dict]) -> None:
inbound_count = Counter()
for page in pages:
for link in page["links"]:
inbound_count[link] += 1
top = inbound_count.most_common(args.top_linked)
if not top:
print("No links found in the wiki.", file=sys.stderr)
return
print(f"Top {len(top)} most-linked-to pages (hubs):")
for slug, count in top:
# Try to find the page for the title
match = next((p for p in pages if p["slug"] == slug), None)
title = (match["meta"].get("title") if match else None) or slug
marker = "" if match else " [BROKEN LINK]"
print(f" {count:4d} {title} ({slug}){marker}")
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("query", nargs="?", default="", help="Query terms.")
parser.add_argument("--wiki", type=Path, default=Path("cml/wiki"), help="Wiki directory (default: cml/wiki).")
parser.add_argument("--top", type=int, default=10, help="Top N results (default: 10).")
parser.add_argument("--type", help="Filter by frontmatter type.")
parser.add_argument("--tag", action="append", default=[], help="Filter by tag (repeatable).")
parser.add_argument("--since", help="Only pages updated on or after YYYY-MM-DD.")
parser.add_argument("--backlinks", help="Find pages linking to this slug.")
parser.add_argument("--top-linked", type=int, help="Show the N most-linked-to pages.")
parser.add_argument("--cache", type=Path, help="(reserved) Cache path for BM25 index.")
args = parser.parse_args()
if not args.wiki.exists():
print(f"Wiki directory not found: {args.wiki}", file=sys.stderr)
sys.exit(1)
pages = collect_pages(args.wiki)
if not pages:
print(f"No wiki pages found under {args.wiki}", file=sys.stderr)
sys.exit(0)
if args.backlinks:
cmd_backlinks(args, pages)
elif args.top_linked:
cmd_top_linked(args, pages)
elif args.query:
cmd_search(args, pages)
else:
parser.print_help()
sys.exit(1)
if __name__ == "__main__":
main()

159
skills/llm-wiki/scripts/wiki_stats.py Normal file
View File

@@ -0,0 +1,159 @@
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""
wiki_stats.py — Quick summary of wiki size, shape, and link density.
Useful for deciding when to shard the index or split pages.
Usage:
python wiki_stats.py [<wiki-dir>]
Example:
python wiki_stats.py wiki/
"""
import argparse
import re
import sys
from collections import Counter
from pathlib import Path
WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]")
FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
SKIP_TOP_LEVEL_FILES = {"SCHEMA.md", "log.md", "README.md"}
SKIP_TOP_LEVEL_DIRS = {"indexes", "graph"}
def parse_type(text: str) -> str | None:
m = FRONTMATTER_RE.match(text)
if not m:
return None
fm = m.group(1)
for line in fm.split("\n"):
kv = re.match(r"^type:\s*(.*)$", line)
if kv:
return kv.group(1).strip().strip('"').strip("'")
return None
def main():
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("wiki", nargs="?", type=Path, default=Path("cml/wiki"))
args = parser.parse_args()
if not args.wiki.exists():
print(f"Wiki directory not found: {args.wiki}", file=sys.stderr)
sys.exit(1)
total_pages = 0
total_lines = 0
total_words = 0
total_links = 0
pages_by_type = Counter()
pages_by_dir = Counter()
largest = []
most_linked_in = Counter()
index_lines = 0
for md_path in args.wiki.rglob("*.md"):
rel = md_path.relative_to(args.wiki)
try:
text = md_path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
continue
if rel.name == "index.md" and len(rel.parts) == 1:
index_lines = text.count("\n") + 1
continue
if rel.parts[0] in SKIP_TOP_LEVEL_FILES:
continue
if rel.parts[0] in SKIP_TOP_LEVEL_DIRS:
continue
if rel.name.startswith("."):
continue
total_pages += 1
line_count = text.count("\n") + 1
word_count = len(text.split())
total_lines += line_count
total_words += word_count
# Strip frontmatter before counting wikilinks; frontmatter uses bare slugs.
body = FRONTMATTER_RE.sub("", text, count=1) if text.startswith("---") else text
links = WIKILINK_RE.findall(body)
total_links += len(links)
for link in links:
target = link.split("|")[0].strip()
most_linked_in[target] += 1
page_type = parse_type(text) or "(none)"
pages_by_type[page_type] += 1
if len(rel.parts) > 1:
pages_by_dir[rel.parts[0]] += 1
else:
pages_by_dir["(root)"] += 1
largest.append((line_count, str(rel)))
largest.sort(reverse=True)
print("=" * 60)
print(f"Wiki Stats: {args.wiki}")
print("=" * 60)
print(f"Pages: {total_pages}")
print(f"Total lines: {total_lines:,}")
print(f"Total words: {total_words:,}")
print(f"Total links: {total_links:,}")
if total_pages:
print(f"Avg page: {total_lines // total_pages} lines / {total_words // total_pages} words")
print(f"Link density: {total_links / total_pages:.1f} links per page")
print(f"index.md: {index_lines} lines" + (" ← shard recommended (>300)" if index_lines > 300 else ""))
print()
print("Pages by type:")
for t, n in pages_by_type.most_common():
print(f" {t:15s} {n}")
print()
print("Pages by directory:")
for d, n in pages_by_dir.most_common():
print(f" {d:15s} {n}")
print()
if largest:
print("Largest pages:")
for lines, path in largest[:10]:
warn = ""
if lines > 800:
warn = " ← OVER HARD CAP"
elif lines > 400:
warn = " ← over soft cap"
print(f" {lines:5d} {path}{warn}")
print()
if most_linked_in:
print("Most-linked-to pages (hubs):")
for slug, count in most_linked_in.most_common(10):
print(f" {count:4d} [[{slug}]]")
print()
# Scaling recommendations
print("Scaling thresholds:")
if total_pages < 50:
print(" → Below first threshold. Flat structure is fine.")
elif total_pages < 150 and index_lines < 300:
print(" → Below shard threshold. Continue with single index.md.")
elif (total_pages >= 150 or index_lines >= 300) and not (args.wiki / "indexes").exists():
print(" → AT SHARD THRESHOLD. Consider sharding index.md into wiki/indexes/<type>.md.")
print(" See references/scaling-playbook.md.")
elif total_pages >= 300:
print(" → Past 300 pages. Use scripts/wiki_search.py as a routine fallback.")
if total_pages >= 500:
print(" → Past 500 pages. Run lint weekly or per-N-ingests.")
if __name__ == "__main__":
main()