530 lines
18 KiB
Python
530 lines
18 KiB
Python
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = ["requests", "pyyaml", "markdown-it-py", "sqlite-vec==0.1.6"]
|
|
# ///
|
|
"""Offline indexer for the wiki skill. Runs every minute from the nanobot crontab.
|
|
|
|
Indexing never happens inside an agent turn: the `exec` tool times out at 60 s while a
|
|
full index of 10^4 chunks takes ~93 s. The agent only ever reads a finished index.
|
|
|
|
Per tick:
|
|
|
|
1. take `wiki/.sync.lock`; a live holder means exit 0 in silence
|
|
2. detect changes cheaply — `git ls-remote` (no fetch) for git sources, a
|
|
path+size+mtime walk for workspace sources, sha256 only on a stat mismatch
|
|
3. nothing changed and nothing pending -> exit 0 (the overwhelming majority of ticks)
|
|
4. re-chunk changed files, then drain every chunk with `embedded_at IS NULL` — which
|
|
is also the way back out of degraded mode after Ollama returns
|
|
5. record indexed_rev / last_sync_at
|
|
6. log a coverage line naming top-level directories no source covers
|
|
7. append the run summary to log/wiki_sync.log
|
|
|
|
Network failure is per-source: a timeout or non-zero git exit logs WARN, skips that
|
|
source with its `indexed_rev` untouched, and lets the others finish. Without the
|
|
timeouts a hanging `ls-remote` would hold the lock and block workspace sources that
|
|
have nothing to do with the network.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
import wiki_store as store
|
|
from wiki_chunker import parse_markdown
|
|
from wiki_config import (
|
|
GIT_KIND,
|
|
LOCK_PATH,
|
|
SYNC_LOG_PATH,
|
|
WIKI_DIR,
|
|
ConfigError,
|
|
SourceConfig,
|
|
WikiConfig,
|
|
db_path,
|
|
load_config,
|
|
source_root,
|
|
)
|
|
from wiki_embed import (
|
|
EmbeddingUnavailable,
|
|
OllamaEmbedder,
|
|
expected_meta,
|
|
meta_mismatches,
|
|
)
|
|
|
|
STALE_SECONDS = 30 * 60
|
|
LS_REMOTE_TIMEOUT = 20
|
|
GIT_TIMEOUT = 300
|
|
GIT_ERROR_CHARS = 300
|
|
SKIP_DIRS = frozenset({".git"})
|
|
|
|
|
|
class GitError(RuntimeError):
|
|
"""A git subprocess failed or timed out."""
|
|
|
|
|
|
@dataclass
|
|
class SourcePlan:
|
|
"""What one source needs done this tick."""
|
|
|
|
changed: list[str] = field(default_factory=list)
|
|
deleted: list[str] = field(default_factory=list)
|
|
stat_refresh: list[tuple[str, int, float]] = field(default_factory=list)
|
|
new_rev: str | None = None
|
|
|
|
@property
|
|
def has_work(self) -> bool:
|
|
return bool(self.changed or self.deleted or self.stat_refresh)
|
|
|
|
|
|
def log(message: str) -> None:
|
|
SYNC_LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
stamp = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with SYNC_LOG_PATH.open("a", encoding="utf-8") as handle:
|
|
handle.write(f"{stamp} {message}\n")
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(UTC).isoformat(timespec="seconds")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# lock
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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:
|
|
"""A lock is dead if unreadable, its pid is gone, or it is older than STALE_SECONDS."""
|
|
try:
|
|
data = json.loads(LOCK_PATH.read_text(encoding="utf-8"))
|
|
pid = int(data["pid"])
|
|
started = datetime.fromisoformat(data["started_at"])
|
|
except (OSError, ValueError, KeyError):
|
|
return True
|
|
if not _pid_alive(pid):
|
|
return True
|
|
return (datetime.now().astimezone() - started).total_seconds() > STALE_SECONDS
|
|
|
|
|
|
def acquire_lock() -> bool:
|
|
"""Atomically create the lock. False when a live sync already runs.
|
|
|
|
Reclaiming a stale lock is not optional: without it a crashed run (OOM, reboot)
|
|
would make every later cron tick exit 0 in silence, forever.
|
|
"""
|
|
LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
for _ in range(2):
|
|
try:
|
|
handle = os.open(LOCK_PATH, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
|
|
except FileExistsError:
|
|
if not _lock_is_stale():
|
|
return False
|
|
log("WARN stale lock, reclaiming")
|
|
LOCK_PATH.unlink(missing_ok=True)
|
|
continue
|
|
payload = {"pid": os.getpid(), "started_at": datetime.now().astimezone().isoformat()}
|
|
with os.fdopen(handle, "w", encoding="utf-8") as file:
|
|
json.dump(payload, file)
|
|
return True
|
|
return False
|
|
|
|
|
|
def release_lock() -> None:
|
|
LOCK_PATH.unlink(missing_ok=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# filesystem
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for block in iter(lambda: handle.read(65536), b""):
|
|
digest.update(block)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def walk_source(source: SourceConfig, root: Path) -> dict[str, tuple[int, float]]:
|
|
"""Covered files under `root`, as rel_path -> (size, mtime). Prunes uncovered dirs."""
|
|
found: dict[str, tuple[int, float]] = {}
|
|
if not root.is_dir():
|
|
return found
|
|
for dirpath, dirnames, filenames in os.walk(root):
|
|
rel_dir = os.path.relpath(dirpath, root)
|
|
rel_dir = "" if rel_dir == "." else rel_dir
|
|
dirnames[:] = [
|
|
name for name in dirnames if name not in SKIP_DIRS and source.covers_dir(f"{rel_dir}/{name}".lstrip("/"))
|
|
]
|
|
for name in filenames:
|
|
rel_path = f"{rel_dir}/{name}".lstrip("/")
|
|
if not source.covers(rel_path):
|
|
continue
|
|
file_path = Path(dirpath) / name
|
|
if not file_path.is_file():
|
|
continue
|
|
stat = file_path.stat()
|
|
found[rel_path] = (stat.st_size, stat.st_mtime)
|
|
return found
|
|
|
|
|
|
def uncovered_top_level_dirs(source: SourceConfig, root: Path) -> list[str]:
|
|
"""Top-level directories holding markdown that no source path reaches.
|
|
|
|
This is the safety net under the whitelist: a directory that silently fails to be
|
|
indexed shows up here instead of nowhere. `wiki/` is skipped because it holds this
|
|
skill's own clones and index — never a candidate, so reporting it is pure noise.
|
|
"""
|
|
if not root.is_dir():
|
|
return []
|
|
uncovered = []
|
|
for entry in sorted(root.iterdir()):
|
|
if not entry.is_dir() or entry.name in SKIP_DIRS or entry == WIKI_DIR:
|
|
continue
|
|
if source.covers_dir(entry.name):
|
|
continue
|
|
if next(entry.rglob("*.md"), None) is not None:
|
|
uncovered.append(entry.name)
|
|
return uncovered
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# git driver
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _git(args: list[str], cwd: Path | None = None, timeout: int = GIT_TIMEOUT) -> str:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
cwd=cwd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
except subprocess.TimeoutExpired as exc:
|
|
raise GitError(f"git {' '.join(args)} timed out after {timeout}s") from exc
|
|
if result.returncode != 0:
|
|
raise GitError(f"git {' '.join(args)} failed: {_one_line(result.stderr)}")
|
|
return result.stdout
|
|
|
|
|
|
def _one_line(text: str) -> str:
|
|
"""Squash git's multi-line stderr into one capped line.
|
|
|
|
A permanently unreachable source warns on every tick, so a six-line stderr would
|
|
put thousands of lines a day into the log and drown everything else.
|
|
"""
|
|
collapsed = " ".join(text.split())
|
|
return collapsed[:GIT_ERROR_CHARS] + ("…" if len(collapsed) > GIT_ERROR_CHARS else "")
|
|
|
|
|
|
def remote_head(url: str) -> str:
|
|
"""Remote HEAD without fetching — the right tool for a per-minute cadence."""
|
|
output = _git(["ls-remote", "--symref", url, "HEAD"], timeout=LS_REMOTE_TIMEOUT)
|
|
for line in output.splitlines():
|
|
parts = line.split()
|
|
if len(parts) == 2 and parts[1] == "HEAD":
|
|
return parts[0]
|
|
raise GitError(f"no HEAD in ls-remote output for {url}")
|
|
|
|
|
|
def _diff_names(root: Path, base_rev: str) -> tuple[list[str], list[str]] | None:
|
|
"""(changed, deleted) between base_rev and FETCH_HEAD, or None if base_rev is gone."""
|
|
try:
|
|
output = _git(["diff", "--name-status", f"{base_rev}..FETCH_HEAD"], cwd=root)
|
|
except GitError:
|
|
return None
|
|
changed, deleted = [], []
|
|
for line in output.splitlines():
|
|
fields = line.split("\t")
|
|
if len(fields) < 2:
|
|
continue
|
|
status = fields[0]
|
|
if status.startswith("R") and len(fields) >= 3:
|
|
deleted.append(fields[1])
|
|
changed.append(fields[2])
|
|
elif status.startswith("D"):
|
|
deleted.append(fields[1])
|
|
else:
|
|
changed.append(fields[1])
|
|
return changed, deleted
|
|
|
|
|
|
def plan_git_source(conn, source: SourceConfig, root: Path, full: bool) -> SourcePlan:
|
|
rev = remote_head(str(source.url))
|
|
row = store.get_source(conn, source.source_id)
|
|
indexed_rev = row["indexed_rev"] if row else None
|
|
|
|
if not root.exists():
|
|
_git(["clone", str(source.url), str(root)])
|
|
indexed_rev = None
|
|
if not full and indexed_rev == rev:
|
|
return SourcePlan(new_rev=rev)
|
|
|
|
_git(["fetch", "--prune"], cwd=root)
|
|
diff = None if (full or not indexed_rev) else _diff_names(root, indexed_rev)
|
|
_git(["reset", "--hard", "FETCH_HEAD"], cwd=root)
|
|
|
|
indexed = store.list_source_files(conn, source.source_id)
|
|
on_disk = walk_source(source, root)
|
|
if diff is None:
|
|
# Fresh clone, --full, or an indexed_rev the repo no longer has: reindex it all.
|
|
return SourcePlan(
|
|
changed=sorted(on_disk),
|
|
deleted=sorted(path for path in indexed if path not in on_disk),
|
|
new_rev=rev,
|
|
)
|
|
raw_changed, raw_deleted = diff
|
|
changed = sorted({path for path in raw_changed if path in on_disk})
|
|
deleted = sorted(
|
|
{path for path in raw_deleted if path in indexed} | {path for path in indexed if path not in on_disk}
|
|
)
|
|
return SourcePlan(changed=changed, deleted=[p for p in deleted if p not in changed], new_rev=rev)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# workspace driver
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def plan_workspace_source(conn, source: SourceConfig, root: Path, full: bool) -> SourcePlan:
|
|
on_disk = walk_source(source, root)
|
|
indexed = store.list_source_files(conn, source.source_id)
|
|
plan = SourcePlan()
|
|
for rel_path, (size, mtime) in sorted(on_disk.items()):
|
|
row = indexed.get(rel_path)
|
|
if full or row is None:
|
|
plan.changed.append(rel_path)
|
|
continue
|
|
if row["size"] == size and row["mtime"] == mtime:
|
|
continue
|
|
if _sha256(root / rel_path) == row["sha256"]:
|
|
# Content is identical; refresh the stats so the next tick stays cheap.
|
|
plan.stat_refresh.append((rel_path, size, mtime))
|
|
else:
|
|
plan.changed.append(rel_path)
|
|
plan.deleted = sorted(path for path in indexed if path not in on_disk)
|
|
return plan
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# indexing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def index_file(conn, source_id: str, root: Path, rel_path: str, now: str) -> int:
|
|
"""Re-chunk one file. Returns the chunk count, or -1 when it could not be read."""
|
|
file_path = root / rel_path
|
|
try:
|
|
text = file_path.read_text(encoding="utf-8")
|
|
stat = file_path.stat()
|
|
except (OSError, UnicodeDecodeError):
|
|
return -1
|
|
parsed = parse_markdown(text, rel_path)
|
|
store.upsert_file(
|
|
conn,
|
|
source_id,
|
|
rel_path,
|
|
parsed.title,
|
|
parsed.tags,
|
|
parsed.headings,
|
|
_sha256(file_path),
|
|
stat.st_size,
|
|
stat.st_mtime,
|
|
now,
|
|
)
|
|
store.replace_chunks(conn, source_id, rel_path, [(c.breadcrumb, c.text) for c in parsed.chunks])
|
|
return len(parsed.chunks)
|
|
|
|
|
|
def apply_plan(conn, source: SourceConfig, root: Path, plan: SourcePlan) -> dict[str, int]:
|
|
counts = {"indexed": 0, "deleted": 0, "unreadable": 0, "chunks": 0}
|
|
now = _now()
|
|
for rel_path in plan.deleted:
|
|
with store.tx(conn):
|
|
store.delete_file(conn, source.source_id, rel_path)
|
|
counts["deleted"] += 1
|
|
for rel_path in plan.changed:
|
|
with store.tx(conn):
|
|
chunks = index_file(conn, source.source_id, root, rel_path, now)
|
|
if chunks < 0:
|
|
counts["unreadable"] += 1
|
|
log(f"WARN {source.source_id}: unreadable {rel_path}")
|
|
continue
|
|
counts["indexed"] += 1
|
|
counts["chunks"] += chunks
|
|
if plan.stat_refresh:
|
|
with store.tx(conn):
|
|
for rel_path, size, mtime in plan.stat_refresh:
|
|
store.touch_file_stat(conn, source.source_id, rel_path, size, mtime)
|
|
with store.tx(conn):
|
|
store.mark_synced(conn, source.source_id, plan.new_rev, now)
|
|
return counts
|
|
|
|
|
|
def drain_pending(conn, config: WikiConfig) -> tuple[int, str | None]:
|
|
"""Embed every chunk still missing a vector. Returns (embedded, warning)."""
|
|
if not store.has_pending(conn):
|
|
return 0, None
|
|
embedder = OllamaEmbedder(config.embedding)
|
|
embedded = 0
|
|
while True:
|
|
batch = store.pending_chunks(conn, config.embedding.batch)
|
|
if not batch:
|
|
return embedded, None
|
|
try:
|
|
vectors = embedder.embed_documents([row["text"] for row in batch])
|
|
except EmbeddingUnavailable as exc:
|
|
return embedded, f"embeddings unavailable, staying FTS-only: {exc}"
|
|
now = _now()
|
|
with store.tx(conn):
|
|
for row, vector in zip(batch, vectors, strict=True):
|
|
store.store_embedding(conn, row["id"], vector, now)
|
|
embedded += len(batch)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def ensure_meta(conn, config: WikiConfig, full: bool) -> str | None:
|
|
"""Write the index identity, or report a mismatch that only --full can resolve."""
|
|
stored = store.read_meta(conn)
|
|
mismatches = meta_mismatches(stored, config.embedding)
|
|
if mismatches and not full:
|
|
return f"index identity mismatch on {', '.join(mismatches)} — run wiki_sync.py --full"
|
|
with store.tx(conn):
|
|
store.write_meta(conn, expected_meta(config.embedding))
|
|
return None
|
|
|
|
|
|
def run(config: WikiConfig, args: argparse.Namespace) -> int:
|
|
sources = [s for s in config.sources if not args.source or s.source_id == args.source]
|
|
if not sources:
|
|
log(f"WARN unknown source {args.source!r}")
|
|
return 1
|
|
|
|
with store.connection(db_path()) as conn:
|
|
problem = ensure_meta(conn, config, args.full)
|
|
if problem:
|
|
log(f"WARN {problem}")
|
|
return 1
|
|
|
|
for source in sources:
|
|
with store.tx(conn):
|
|
store.upsert_source(conn, source.source_id, source.kind)
|
|
|
|
warnings: list[str] = []
|
|
did_work = args.full
|
|
for source in sources:
|
|
root = source_root(source)
|
|
try:
|
|
if source.kind == GIT_KIND:
|
|
plan = plan_git_source(conn, source, root, args.full)
|
|
else:
|
|
plan = plan_workspace_source(conn, source, root, args.full)
|
|
except GitError as exc:
|
|
# indexed_rev stays untouched, so the next tick retries this source.
|
|
warnings.append(f"{source.source_id}: {exc}")
|
|
log(f"WARN {source.source_id}: {exc}")
|
|
continue
|
|
|
|
row = store.get_source(conn, source.source_id)
|
|
indexed_rev = row["indexed_rev"] if row else None
|
|
rev_moved = plan.new_rev is not None and plan.new_rev != indexed_rev
|
|
if not plan.has_work and not rev_moved:
|
|
continue
|
|
|
|
did_work = True
|
|
if not plan.has_work:
|
|
# Upstream moved but touched nothing we index — record the rev so the
|
|
# next tick stops fetching.
|
|
with store.tx(conn):
|
|
store.mark_synced(conn, source.source_id, plan.new_rev, _now())
|
|
log(f"{source.source_id}: rev moved to {plan.new_rev}, no indexed file changed")
|
|
continue
|
|
|
|
counts = apply_plan(conn, source, root, plan)
|
|
log(
|
|
f"{source.source_id}: indexed {counts['indexed']} files "
|
|
f"({counts['chunks']} chunks), deleted {counts['deleted']}"
|
|
)
|
|
|
|
embedded, embed_warning = drain_pending(conn, config)
|
|
if embed_warning:
|
|
warnings.append(embed_warning)
|
|
log(f"WARN {embed_warning}")
|
|
|
|
if not did_work and not embedded and not warnings:
|
|
return 0
|
|
|
|
for source in sources:
|
|
uncovered = uncovered_top_level_dirs(source, source_root(source))
|
|
if uncovered:
|
|
log(f"coverage {source.source_id}: markdown outside paths in {', '.join(uncovered)}")
|
|
|
|
stats = store.index_stats(conn)
|
|
log(
|
|
f"done files={stats['files']} chunks={stats['chunks']} vectors={stats['vectors']} "
|
|
f"pending={stats['pending']} embedded={embedded}"
|
|
)
|
|
return 1 if warnings else 0
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Index the wiki sources into wiki/index.sqlite.")
|
|
parser.add_argument(
|
|
"--full",
|
|
action="store_true",
|
|
help="wipe and rebuild the index (needed after a model or chunker change)",
|
|
)
|
|
parser.add_argument("--source", help="limit the run to one source id")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
try:
|
|
config = load_config()
|
|
except ConfigError as exc:
|
|
log(f"WARN {exc}")
|
|
print(f"config error: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
WIKI_DIR.mkdir(parents=True, exist_ok=True)
|
|
if not acquire_lock():
|
|
return 0
|
|
try:
|
|
if args.full:
|
|
log("full reindex requested")
|
|
with store.transaction(db_path()) as conn:
|
|
store.reset_index(conn)
|
|
return run(config, args)
|
|
finally:
|
|
release_lock()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|