#!/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())