diff --git a/corpus/areas.py b/corpus/areas.py new file mode 100644 index 0000000..5465135 --- /dev/null +++ b/corpus/areas.py @@ -0,0 +1,94 @@ +"""Which resume area a harvested document feeds. + +The payload `area` is what lets the query path filter *before* it searches, so a +harvested file has to be classified once, deterministically, and never promoted +to make a page read better. + +The classification is keyword evidence, in this order: the path, the repository +name, the description and topics, then the languages. The result is constrained +to the platform guide's `fits` list: a platform cannot feed an area it does not +declare. When nothing matches, the first `fits` entry wins, which is why every +guide lists its strongest area first. + +This is a heuristic and it is labelled as one. It is not a claim about Sam's +career; the curated `content/` tree carries the claims. +""" + +from __future__ import annotations + +import re + +from .payloads import AREAS + +# Order matters: the first area with a hit wins, so the more specific evidence +# (an ESP32 repository is edge, not platform) is checked before the general. +_RULES = ( + ("edge", ( + "esp32", "esp8266", "arduino", "lora", "lorawan", "mqtt", "sensor", + "zigbee", "z-wave", "relay", "gpio", "pir", "modbus", "iot", + "firmware", "breadboard", "raspberry pi", "rpi", "camera", "cam", + "temperature", "watering", "fish feeder", "driveway", "light", + "snapcast", "mopidy", "audio", "voice bridge", + )), + ("ai", ( + "ai", "llm", "agent", "agentic", "rag", "embedding", "vector", + "model", "ml", "prompt", "graphify", "omniroute", "paperclip", + "langgraph", "prefect", "whisper", "piper", "chat", "memory", + "knowledge graph", "vision", "triage", + )), + ("platform", ( + "nixos", "nix", "docker", "compose", "kubernetes", "k8s", "deploy", + "deployment", "backup", "borg", "kopia", "restic", "network", "dns", + "email", "caddy", "proxmox", "postgres", "terraform", "ansible", + "linux", "systemd", "homelab", "server", "infrastructure", "git", + "gitea", "monitor", "logging", "security", "firewall", "vpn", + "tailscale", "storage", "garage", "s3", + )), + ("products", ( + "product", "app", "web", "dashboard", "react", "laravel", "flutter", + "kotlin", "calendar", "photo", "kontra", "store", "resume", "site", + "gps", "tracker", "api", "frontend", "ui", "ux", + )), + ("history", ( + "history", "agency", "client", "archive", "old", "legacy", "2019", + "2018", "2020", "2021", "career", "resume copy", + )), + ("about", ("about", "profile", "bio", "personal", "cv")), +) + +_WORD_SPLIT = re.compile(r"[^0-9a-z]+") + + +def area_for(evidence: str, fits, fallback: str = None) -> str: + """Classify one document from its path, name, description and topics.""" + haystack = " " + _WORD_SPLIT.sub(" ", (evidence or "").lower()) + " " + allowed = [area for area in (fits or AREAS) if area in AREAS] + if not allowed: + allowed = list(AREAS) + for area, needles in _RULES: + if area not in allowed: + continue + for needle in needles: + if f" {needle} " in haystack or f" {needle}s " in haystack: + return area + if fallback in allowed: + return fallback + return allowed[0] + + +def area_for_repo(repo: dict, fits) -> str: + topics = " ".join(repo.get("topics") or []) + evidence = " ".join( + str(part) + for part in ( + repo.get("full_name", ""), + repo.get("name", ""), + repo.get("description", ""), + topics, + ) + ) + return area_for(evidence, fits, fallback=(fits or [None])[0]) + + +def area_for_note(path: str, fits) -> str: + return area_for(path, fits, fallback=(fits or [None])[0]) diff --git a/corpus/chunker.py b/corpus/chunker.py index f6ac30c..fe31cc7 100644 --- a/corpus/chunker.py +++ b/corpus/chunker.py @@ -285,3 +285,124 @@ def verify_no_split(chunks: list, regions: list) -> None: f"chunk {chunk.index} splits protected lines {first}-{last} " f"(chunk covers {chunk.line_start}-{chunk.line_end})" ) + + +# -- code ----------------------------------------------------------------- + + +def _code_depth(line: str, depth: int) -> int: + """Approximate bracket depth for one line, ignoring strings and comments. + + This is a heuristic lexer, not a parser: it is here to answer one question — + "is this a safe line to break after?" — and it errs towards *not* breaking. + A line with an ambiguous quote or an unterminated bracket simply keeps the + chunk open, which can only produce a larger chunk, never a broken one. + """ + index = 0 + quote = None + length = len(line) + while index < length: + char = line[index] + if quote: + if char == "\\": + index += 2 + continue + if line.startswith(quote, index): + index += len(quote) + quote = None + continue + index += 1 + continue + if line.startswith("//", index) or line.startswith("#", index) and line.lstrip().startswith("#"): + break + if char in "\"'`": + triple = char * 3 + if line.startswith(triple, index): + quote = triple + index += 3 + continue + quote = char + index += 1 + continue + if char in "([{": + depth += 1 + elif char in ")]}": + depth = max(0, depth - 1) + index += 1 + return depth + + +def code_segments(text: str) -> list: + """Split code into whole top-level segments. A function is never cut.""" + lines = text.split("\n") + segments = [] + current = [] + depth = 0 + start = 1 + for number, line in enumerate(lines, start=1): + if not current: + start = number + depth = _code_depth(line, depth) + current.append(line) + # A blank line at bracket depth zero is the only automatic boundary. + if depth == 0 and not line.strip() and len(current) > 1: + segments.append((start, number, "\n".join(current))) + current = [] + if current: + segments.append((start, len(lines), "\n".join(current))) + return [(start, end, body) for start, end, body in segments if body.strip()] + + +def chunk_code(text: str) -> list: + """Chunk a source file without cutting a function, class or block. + + Segments are whole top-level units. A unit larger than the ceiling is kept + as one oversize chunk and flagged, exactly like an oversized Markdown fence: + cutting it would break the rule this function exists to keep. + + Overlap repeats the previous unit, never part of one, so an overlapping + chunk is still syntactically whole. + """ + segments = code_segments(text) + if not segments: + raise ChunkError("source file produced no segments (empty?)") + + chunks = [] + current = [] # list of (start, end, body) + tokens = 0 + + def flush(): + nonlocal current, tokens + if not current: + return + body = "\n".join(segment[2] for segment in current).strip("\n") + chunks.append( + Chunk( + index=len(chunks), + text=body, + tokens=estimate_tokens(body), + line_start=current[0][0], + line_end=current[-1][1], + blocks=tuple( + Block("code", segment[2], estimate_tokens(segment[2]), segment[0], segment[1]) + for segment in current + ), + oversize=len(current) == 1 and estimate_tokens(current[0][2]) > TOKEN_MAX, + ) + ) + current = [] + tokens = 0 + + for segment in segments: + segment_tokens = estimate_tokens(segment[2]) + if current and tokens + segment_tokens > TOKEN_MAX: + previous = current[-1] + flush() + budget = int(round(TOKEN_TARGET * OVERLAP_RATIO)) + if estimate_tokens(previous[2]) <= budget: + current = [previous] + tokens = estimate_tokens(previous[2]) + current.append(segment) + tokens += segment_tokens + flush() + return chunks diff --git a/corpus/cli.py b/corpus/cli.py index 22f70c5..689b5f7 100644 --- a/corpus/cli.py +++ b/corpus/cli.py @@ -28,9 +28,10 @@ import os import sys from . import config as config_module -from . import freshness, pipeline, store as store_module +from . import freshness, guides, pipeline, store as store_module from .config import Config, ConfigError from .qdrant import Qdrant, QdrantError +from .sources import graph as graph_module def main(argv=None) -> int: @@ -49,6 +50,11 @@ def main(argv=None) -> int: "dim": args.dim, "db_path": args.db, "data_dir": args.data, + "offline": True if args.offline else None, + "platforms": tuple(part.strip() for part in args.platforms.split(",") if part.strip()) + if args.platforms + else None, + "graph_path": args.graph_path or None, } try: cfg = Config.from_env(**overrides) @@ -68,6 +74,9 @@ def main(argv=None) -> int: "hybrid": _cmd_hybrid, "measure": _cmd_measure, "info": _cmd_info, + "graph": _cmd_graph, + "scan": _cmd_scan, + "sources": _cmd_sources, }[args.command] return handler(cfg, args, log) except (pipeline.PipelineError, ConfigError) as exc: @@ -92,6 +101,9 @@ def _build_parser() -> argparse.ArgumentParser: parser.add_argument("--db", help="SQLite corpus database path") parser.add_argument("--data", help="corpus data directory") parser.add_argument("--quiet", action="store_true", help="progress on stderr only becomes silent") + parser.add_argument("--offline", action="store_true", help="read content/ only; do not harvest the estate") + parser.add_argument("--platforms", help="comma-separated platforms to harvest (default: all enabled)") + parser.add_argument("--graph-path", help="read graph.json from this path instead of Gitea") sub = parser.add_subparsers(dest="command") @@ -128,6 +140,23 @@ def _build_parser() -> argparse.ArgumentParser: measure.add_argument("--keep", action="store_true", help="keep the measurement collection") measure.add_argument("--json", action="store_true") + graph = sub.add_parser("graph", help="query graph nodes and edges by keyword or vector") + graph.add_argument("--query", required=True) + graph.add_argument("--area") + graph.add_argument("--limit", type=int, default=10) + graph.add_argument("--json", action="store_true") + + neighbours = sub.add_parser("graph-neighbours", help="edges attached to a graph node id") + neighbours.add_argument("--node", required=True) + neighbours.add_argument("--limit", type=int, default=50) + neighbours.add_argument("--json", action="store_true") + + scan = sub.add_parser("scan", help="search the finished pool for credential-shaped content") + scan.add_argument("--json", action="store_true") + + sources = sub.add_parser("sources", help="show the platform guides and the last harvest") + sources.add_argument("--json", action="store_true") + sub.add_parser("info", help="print the resolved configuration") return parser @@ -235,6 +264,88 @@ def _cmd_measure(cfg, args, log) -> int: return 0 +def _cmd_graph(cfg, args, log) -> int: + with store_module.Store(cfg.db_path) as store: + report = graph_module.query(cfg, store, args.query, area=args.area, limit=args.limit) + if args.json: + print(json.dumps(report, indent=2, sort_keys=True, default=str)) + return 0 + print(f"graph query {args.query!r} area={args.area!r}") + print("") + print(f"keyword ({len(report['keyword'])} row(s)):") + for row in report["keyword"]: + kind = row.get("kind") + print(f" [{kind}] {row.get('label')} -> {row.get('source_note')} {row.get('origin_url')}") + print("") + print(f"vector ({len(report['vector'])} row(s)):") + for row in report["vector"]: + print(f" {row['score']:.3f} [{row['kind']}] {row['label']} -> {row['source_note']}") + if report.get("vector_note"): + print(f" note: {report['vector_note']}") + return 0 if report["ok"] else 1 + + +def _cmd_graph_neighbours(cfg, args, log) -> int: + with store_module.Store(cfg.db_path) as store: + node = store.graph_node(args.node) + edges = store.graph_neighbours(args.node, limit=args.limit) + if args.json: + print(json.dumps({"node": node, "edges": edges}, indent=2, sort_keys=True, default=str)) + return 0 + if not node: + print(f"no graph node with id {args.node!r}") + return 1 + print(f"{node.get('label')} ({node.get('node_type')})") + print(f" note: {node.get('source_note')}") + print(f" origin: {node.get('origin_url')}") + for edge in edges: + print( + f" {edge.get('source_label')} --{edge.get('relation')}--> {edge.get('target_label')}" + f" [{edge.get('source_note')}]" + ) + return 0 + + +def _cmd_scan(cfg, args, log) -> int: + report = pipeline.scan_pool(cfg) + if args.json: + print(json.dumps(report, indent=2, sort_keys=True)) + else: + print( + f"scanned {report['chunks']} chunk(s): " + + ("clean" if report["ok"] else f"{len(report['hits'])} hit(s)") + ) + for hit in report["hits"]: + print(f" {hit}") + return 0 if report["ok"] else 1 + + +def _cmd_sources(cfg, args, log) -> int: + loaded = guides.load_guides(cfg.root) + public = guides.build_public_manifest(loaded) + last = {} + if os.path.exists(cfg.db_path): + with store_module.Store(cfg.db_path) as store: + last = store.get_meta("sources") or {} + if args.json: + print(json.dumps({"guides": public, "last_harvest": last}, indent=2, sort_keys=True)) + return 0 + for row in public["platforms"]: + state = "enabled" if row["enabled"] else "disabled" + print(f"{row['name']} ({state}) — {row['what']}") + print(f" base: {row['base']} fits: {', '.join(row['fits'])} read: {', '.join(row['read'])}") + print(f" include: {json.dumps(row['include'])}") + print(f" exclude: {json.dumps(row['exclude'])}") + seen = last.get(row["name"]) or {} + if seen: + print( + f" last harvest: {seen.get('documents', 0)} document(s), " + f"newest {seen.get('newest') or '—'}" + ) + print("") + return 0 + + def _cmd_info(cfg, args, log) -> int: print(cfg.describe()) qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout) @@ -248,6 +359,14 @@ def _cmd_info(cfg, args, log) -> int: with store_module.Store(cfg.db_path) as store: print(f"store: {store.chunk_count()} chunk(s) across {store.source_count()} source(s)") print(f"contract: {json.dumps(store.get_meta('collection'), sort_keys=True)}") + print(f"graph: {json.dumps(store.graph_counts(), sort_keys=True)}") + sources = store.get_meta("sources") or {} + for name in sorted(sources): + row = sources[name] + print( + f"source {name}: {row.get('documents', 0)} document(s), " + f"newest {row.get('newest') or '—'}" + ) else: print(f"store: {cfg.db_path} does not exist") return 0 diff --git a/corpus/config.py b/corpus/config.py index 163305a..ef1e06c 100644 --- a/corpus/config.py +++ b/corpus/config.py @@ -13,7 +13,7 @@ from dataclasses import dataclass # Bumped whenever chunk boundaries or payload shape change. A stored corpus # carrying a different version is stale by definition and gets rebuilt. -CHUNKER_VERSION = "blocks-v1" +CHUNKER_VERSION = "blocks-v2-code-v1" # Provider defaults. `local-hashing` is a deterministic, offline provider used # by the tests, by `make corpus-test`, and by any rebuild on a host with no @@ -29,6 +29,14 @@ DEFAULT_QDRANT_URL = "http://127.0.0.1:6333" DEFAULT_COLLECTION = "resume_corpus" DEFAULT_PROVIDER = "local-hashing" DEFAULT_OMNIROUTE_URL = "https://omniroute.lab.audasmedia.com.au" +DEFAULT_GITEA_URL = "https://gitea.lab.audasmedia.com.au" +DEFAULT_PLATFORMS = ("gitea", "obsidian", "graphify") +DEFAULT_GRAPH_COLLECTION = "resume_graph" + +# A source that has not changed in this window is reported by the audit. The +# estate is active; silence is evidence that ingestion stopped, not that the +# work stopped. +DEFAULT_SOURCE_SILENCE_DAYS = 30 # Freshness horizon: a document whose `last_verified` is older than this is # reported stale by the nightly audit. docs/phase-4-notes.md section 5. @@ -60,6 +68,13 @@ class Config: max_age_days: int warn_days: int apprise_url: str + gitea_url: str = DEFAULT_GITEA_URL + gitea_token: str = "" + platforms: tuple = DEFAULT_PLATFORMS + offline: bool = False + graph_path: str = "" + graph_collection: str = DEFAULT_GRAPH_COLLECTION + source_silence_days: int = DEFAULT_SOURCE_SILENCE_DAYS vectors_on_disk: bool = False # -- construction ----------------------------------------------------- @@ -102,6 +117,19 @@ class Config: ), warn_days=int(os.environ.get("RESUME_WARN_DAYS") or DEFAULT_WARN_DAYS), apprise_url=os.environ.get("RESUME_APPRISE_URL") or "", + gitea_url=(os.environ.get("GITEA_URL") or os.environ.get("RESUME_GITEA_URL") or DEFAULT_GITEA_URL).rstrip("/"), + gitea_token=os.environ.get("GITEA_TOKEN") or "", + platforms=tuple( + part.strip() + for part in (os.environ.get("RESUME_PLATFORMS") or ",".join(DEFAULT_PLATFORMS)).split(",") + if part.strip() + ), + offline=(os.environ.get("RESUME_OFFLINE") or "").lower() in ("1", "true", "yes"), + graph_path=os.environ.get("RESUME_GRAPH_PATH") or "", + graph_collection=os.environ.get("RESUME_GRAPH_COLLECTION") or DEFAULT_GRAPH_COLLECTION, + source_silence_days=int( + os.environ.get("RESUME_SOURCE_SILENCE_DAYS") or DEFAULT_SOURCE_SILENCE_DAYS + ), vectors_on_disk=(os.environ.get("RESUME_QDRANT_VECTORS_ON_DISK") or "").lower() in ("1", "true", "yes"), ) @@ -124,6 +152,21 @@ class Config: ) if not os.path.isdir(self.content_dir): raise ConfigError(f"content directory not found: {self.content_dir}") + unknown = [name for name in self.platforms if name not in ("gitea", "obsidian", "graphify")] + if unknown: + raise ConfigError( + f"RESUME_PLATFORMS contains unknown platform(s) {unknown}; " + "expected gitea, obsidian, graphify" + ) + if not self.offline and self.gitea_token == "": + # Only a hard error when a platform that needs Gitea is selected; the + # token itself is never printed or stored. + if set(self.platforms) & {"gitea", "obsidian", "graphify"}: + raise ConfigError( + "GITEA_TOKEN is not set. Harvesting needs read access to Gitea " + "(~/.config/environment.d/10-secrets.conf), or set RESUME_OFFLINE=1 " + "to rebuild only the curated tier." + ) # -- presentation ----------------------------------------------------- @@ -140,7 +183,10 @@ class Config: f"batch={self.batch_size}\n" f"api_url={self.api_url} api_key={key}\n" f"freshness=max_age_days:{self.max_age_days} warn_days:{self.warn_days}\n" - f"qdrant vectors_on_disk={self.vectors_on_disk}\n" + f"estate=platforms:{','.join(self.platforms)} offline:{self.offline} " + f"gitea:{self.gitea_url} token:{'set' if self.gitea_token else 'unset'} " + f"graph_path:{self.graph_path or 'from gitea'} silence_days:{self.source_silence_days}\n" + f"qdrant vectors_on_disk={self.vectors_on_disk} graph_collection={self.graph_collection}\n" f"chunker={CHUNKER_VERSION}" ) diff --git a/corpus/exclusions.py b/corpus/exclusions.py new file mode 100644 index 0000000..20f5b44 --- /dev/null +++ b/corpus/exclusions.py @@ -0,0 +1,256 @@ +"""What must never reach the pool. + +AGENTS.md §4.4 and `docs/phase-4-corpus.md` §6 make this a hard gate, not a +suggestion: node_modules, virtualenvs, git internals, `.trash`, `.obsidian`, the +SilverBullet auth file, credential-shaped files and binaries are refused. The +platform guide excludes are *additional*; they cannot widen this list. + +Two different outcomes, because they are different things: + +* **reject** — the file is not read at all. The reason is recorded and reported + so a run says *why* it skipped something. +* **image** — a media file is recorded as an image node reference (path only, + never bytes, never extracted text). Phase 4 §1. + +Credential-shaped *content* is a third thing: a file with a valid path can still +contain a private key or an API token. `scan_text` rejects the whole document +rather than redacting part of it, and the same scan runs over the finished pool +so "no secret in the pool" is verified by search, not asserted. +""" + +from __future__ import annotations + +import re + +from . import globs + +# Directory names that are never descended into. Checked per path component, so +# a file deep inside one is refused by the same rule as its parent. +EXCLUDED_DIRS = frozenset( + { + "node_modules", + ".venv", + "venv", + ".git", + ".trash", + ".obsidian", + ".pi", + "__pycache__", + ".mypy_cache", + ".pytest_cache", + ".ruff_cache", + "graphify-out", + "graphify-out-temp", + ".next", + ".cache", + "site-packages", + } +) + +# Basenames (and simple suffix patterns) that are credential-shaped by name. +CREDENTIAL_FILES = ( + ".silverbullet.auth.json", + ".env", + ".env.*", + "*.env", + "*.env.*", + ".netrc", + ".git-credentials", + ".npmrc", + ".pypirc", + "credentials", + "credentials.*", + "secrets.*", + "secret.*", + "auth.json", + "*.pem", + "*.key", + "*.p12", + "*.pfx", + "*.jks", + "*.keystore", + "*.kdbx", + "*.age", + "id_rsa", + "id_ed25519", + "id_ecdsa", + "*.lock", + "*.lockb", +) + +# Media is never text. It becomes an image node reference and nothing else. +IMAGE_EXTENSIONS = frozenset( + {"png", "jpg", "jpeg", "gif", "webp", "bmp", "tif", "tiff", "heic", "avif", "ico"} +) +VIDEO_EXTENSIONS = frozenset({"mp4", "mov", "mkv", "webm", "avi", "m4v", "mpg", "mpeg"}) +AUDIO_EXTENSIONS = frozenset({"mp3", "wav", "flac", "ogg", "m4a", "aac", "opus", "wma"}) +MEDIA_EXTENSIONS = IMAGE_EXTENSIONS | VIDEO_EXTENSIONS | AUDIO_EXTENSIONS + +# Everything else that is bytes, not prose. Rejected with a reason; not indexed +# and not recorded as media. +BINARY_EXTENSIONS = frozenset( + { + "pdf", "zip", "tar", "gz", "bz2", "xz", "7z", "rar", "iso", "dmg", + "exe", "dll", "so", "dylib", "o", "a", "class", "jar", "war", + "pyc", "pyo", "wasm", "ttf", "otf", "woff", "woff2", "eot", + "sqlite", "sqlite3", "db", "mdb", "bin", "dat", "pkl", "npy", "npz", + "stl", "step", "obj", "blend", "psd", "ai", "sketch", "fig", + } +) + +# Text-ish extensions the estate actually contains. A file with an unknown +# extension is admitted only when it decodes as UTF-8 and does not look binary. +TEXT_EXTENSIONS = frozenset( + { + "md", "markdown", "mdx", "txt", "rst", "adoc", "org", + "py", "go", "ts", "tsx", "js", "jsx", "mjs", "cjs", "rs", "rb", + "php", "kt", "kts", "java", "cs", "c", "h", "cc", "cpp", "hpp", + "ino", "lua", "pl", "r", "scala", "swift", "dart", "ex", "exs", + "sh", "bash", "zsh", "fish", "ps1", "bat", + "nix", "toml", "yml", "yaml", "ini", "conf", "cfg", "properties", + "json", "jsonc", "xml", "html", "htm", "css", "scss", "sass", "less", + "sql", "graphql", "proto", "tf", "hcl", "dockerfile", "makefile", + "gitignore", "dockerignore", "editorconfig", "service", "timer", + "csv", "tsv", "svg", "tex", "patch", "diff", "env.example", + "example", "sample", "template", "j2", "tmpl", + } +) + +_CREDENTIAL_PATTERNS = ( + ("private-key", re.compile(r"-----BEGIN (?:[A-Z]+ )?PRIVATE KEY-----")), + ("aws-access-key", re.compile(r"\bAKIA[0-9A-Z]{16}\b")), + ("github-token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b")), + ("slack-token", re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b")), + ("openai-key", re.compile(r"\bsk-[A-Za-z0-9]{20,}\b")), + ("anthropic-key", re.compile(r"\bsk-ant-[A-Za-z0-9_\-]{20,}\b")), + ("jwt", re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\b")), + ("age-key", re.compile(r"\bAGE-SECRET-KEY-1[0-9A-Z]{20,}\b")), + ( + "assigned-secret", + re.compile( + r"(?im)^\s*(?:export\s+)?[A-Z0-9_]*(?:SECRET|TOKEN|PASSWORD|PASSWD|API_KEY|" + r"ACCESS_KEY|PRIVATE_KEY)[A-Z0-9_]*\s*[:=]\s*[\"']?[^\s\"'#]{8,}" + ), + ), +) + +# A `.env.example`-style file may document the *name* of a variable. It may not +# carry a value. This filter keeps `GITEA_TOKEN=` and `TOKEN=changeme` out. +_PLACEHOLDERS = {"", "changeme", "change-me", "xxx", "todo", "placeholder", "your-token-here", "redacted"} + + +class Rejection: + """One file that was not admitted, and the rule that refused it.""" + + __slots__ = ("path", "platform", "reason", "detail") + + def __init__(self, path: str, platform: str, reason: str, detail: str = "") -> None: + self.path = path + self.platform = platform + self.reason = reason + self.detail = detail + + def to_dict(self) -> dict: + return { + "path": self.path, + "platform": self.platform, + "reason": self.reason, + "detail": self.detail, + } + + +def extension(path: str) -> str: + base = path.rsplit("/", 1)[-1].lower() + if "." not in base: + return base + return base.rsplit(".", 1)[-1] + + +def basename(path: str) -> str: + return path.rsplit("/", 1)[-1] + + +def classify(path: str, guide=None, size: int | None = None) -> tuple: + """Return `(kind, reason)` for a path. + + `kind` is one of `text`, `image`, `reject`. `reason` is empty for text and + names the rule for the other two. Guide excludes are applied here so a + caller cannot forget them. + """ + path = globs.normalise(path) + if not path: + return "reject", "empty path" + + parts = path.split("/") + for part in parts[:-1]: + if part.lower() in EXCLUDED_DIRS: + return "reject", f"excluded directory {part!r}" + + name = basename(path) + lowered = name.lower() + if lowered in EXCLUDED_DIRS: + return "reject", f"excluded directory {lowered!r}" + for pattern in CREDENTIAL_FILES: + if globs.matches(lowered, pattern) or globs.matches(name, pattern): + return "reject", "credential-shaped file name" + + if guide is not None: + exclude = getattr(guide, "exclude", {}) or {} + if globs.under_any(path, exclude.get("paths")): + return "reject", "excluded by platform guide" + if globs.matches_any(path, exclude.get("files") or ()): + return "reject", "excluded by platform guide" + + ext = extension(path) + if ext in MEDIA_EXTENSIONS: + return "image", f"media ({ext}) recorded as an image reference" + if ext in BINARY_EXTENSIONS: + return "reject", f"binary ({ext})" + + if size is not None: + limit = int(((getattr(guide, "limits", {}) or {}).get("max_file_kb") or 200) * 1024) + if size > limit: + return "reject", f"oversize ({size} bytes > {limit})" + + if guide is not None: + include = getattr(guide, "include", {}) or {} + include_paths = include.get("paths") or [] + if include_paths and not globs.under_any(path, include_paths): + return "reject", "not under an included path" + + if ext not in TEXT_EXTENSIONS and name.lower() not in TEXT_EXTENSIONS: + # Unknown extension: admitted only if the caller proves it is UTF-8 text. + return "unknown", "" + return "text", "" + + +def looks_binary(data: bytes) -> bool: + if b"\x00" in data[:4096]: + return True + try: + data.decode("utf-8") + except UnicodeDecodeError: + return True + return False + + +def scan_text(text: str) -> list: + """Return `[(kind, line_number, excerpt)]` for credential-shaped content. + + The excerpt is short and never includes the matched value: a rejection + report must not become the leak it is reporting. + """ + hits = [] + for number, line in enumerate(text.splitlines(), start=1): + for kind, pattern in _CREDENTIAL_PATTERNS: + match = pattern.search(line) + if not match: + continue + value = match.group(0) + if kind == "assigned-secret": + assigned = value.split("=", 1)[-1].split(":", 1)[-1].strip().strip("\"'") + if assigned.lower() in _PLACEHOLDERS: + continue + hits.append((kind, number, line.strip()[:60].split("=", 1)[0])) + break + return hits diff --git a/corpus/globs.py b/corpus/globs.py new file mode 100644 index 0000000..fa14a88 --- /dev/null +++ b/corpus/globs.py @@ -0,0 +1,89 @@ +"""Path globs, with the one extension the guides need: `**/` matches nothing. + +`fnmatch` is close but wrong for this job in two ways: it lets `*` cross a `/`, +and it has no `**`. Platform guides are read by a human and written as ordinary +gitignore-style globs (`**/node_modules/**`, `200 projects/**`), so this module +implements exactly that little language and nothing else. + +Matching is always against a normalised, repository-relative POSIX path. +""" + +from __future__ import annotations + +import re + +_CACHE: dict = {} + + +def normalise(path: str) -> str: + """Repository-relative, POSIX separators, no leading `./` or `/`.""" + path = str(path).replace("\\", "/").strip() + while path.startswith("./"): + path = path[2:] + return path.lstrip("/") + + +def compile_glob(pattern: str): + """Compile one glob to a regular expression. Cached.""" + cached = _CACHE.get(pattern) + if cached is not None: + return cached + + pattern = normalise(pattern) + out = ["^"] + index = 0 + while index < len(pattern): + char = pattern[index] + if char == "*": + if pattern[index : index + 2] == "**": + index += 2 + if pattern[index : index + 1] == "/": + # `**/` — any number of whole directories, including none. + out.append("(?:[^/]+/)*") + index += 1 + else: + out.append(".*") + continue + out.append("[^/]*") + elif char == "?": + out.append("[^/]") + elif char == "[": + close = pattern.find("]", index + 1) + if close == -1: + out.append(re.escape(char)) + else: + body = pattern[index + 1 : close] + if body.startswith("!"): + body = "^" + body[1:] + out.append("[" + body + "]") + index = close + else: + out.append(re.escape(char)) + index += 1 + out.append("$") + compiled = re.compile("".join(out)) + _CACHE[pattern] = compiled + return compiled + + +def matches(path: str, pattern: str) -> bool: + return bool(compile_glob(pattern).match(normalise(path))) + + +def matches_any(path: str, patterns) -> bool: + return any(matches(path, pattern) for pattern in (patterns or ())) + + +def under_any(path: str, patterns) -> bool: + """True when `path` is a pattern's directory itself or inside it. + + `200 projects/**` should also exclude the bare directory entry `200 projects` + that a tree listing returns, not only the files underneath it. + """ + path = normalise(path) + for pattern in patterns or (): + if matches(path, pattern): + return True + if pattern.endswith("/**") and path == normalise(pattern[:-3]): + return True + return False diff --git a/corpus/guides.py b/corpus/guides.py new file mode 100644 index 0000000..3dbe10f --- /dev/null +++ b/corpus/guides.py @@ -0,0 +1,227 @@ +"""Platform guides: one small file per platform, configuration and public copy. + +`platforms/.yml` says what a platform is, where it lives, what to include, +what to exclude, which resume area it feeds, and whether it is enabled. The same +wording is shown on the public site, so every value here is written to be read +by a visitor, not just by the pipeline. + +The schema is fixed and validated at load time. An unknown area, a missing +`what`, or a guide whose filename and `name` disagree is a configuration error, +not a default. A disabled guide is loaded and shown, but not harvested. + +There is exactly one YAML dialect in this repository and one parser for it +(`tools/validate_content.py`); this module imports that parser rather than +adding a second one. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + +from . import loader +from .payloads import AREAS # one area enum for guides and payloads + +# The platforms the pipeline knows how to harvest. `curated` is content/ and is +# always on: it is tier 1 and has no guide because it is not an external estate. +KINDS = ("gitea", "obsidian", "graphify") +REQUIRED = ("name", "enabled", "what", "fits", "read") +DEFAULT_LIMITS = { + "max_documents": 4000, + "max_file_kb": 200, + "max_files_per_repo": 40, + "max_commits": 20, + "max_tree_entries": 3000, + "max_nodes": 20000, + "max_edges": 40000, +} + + +class GuideError(RuntimeError): + """A platform guide is missing, malformed, or inconsistent.""" + + +@dataclass +class Guide: + name: str + enabled: bool + what: str + fits: list + read: list + base: str = "" + auth: str = "" + include: dict = field(default_factory=dict) + exclude: dict = field(default_factory=dict) + limits: dict = field(default_factory=dict) + public: bool = True + path: str = "" # repository-relative, for the public page and error messages + + # -- derived ---------------------------------------------------------- + + @property + def kind(self) -> str: + return self.name + + @property + def primary_area(self) -> str: + return self.fits[0] + + def limit(self, name: str) -> int: + return int(self.limits.get(name) or DEFAULT_LIMITS.get(name) or 0) + + def to_public_dict(self) -> dict: + """What the site shows. No auth variable values, only its name.""" + return { + "name": self.name, + "enabled": self.enabled, + "what": self.what, + "fits": list(self.fits), + "read": list(self.read), + "base": self.base, + "include": { + "repos": _as_list((self.include or {}).get("repos")), + "paths": _as_list((self.include or {}).get("paths")), + }, + "exclude": { + "repos": _as_list((self.exclude or {}).get("repos")), + "paths": _as_list((self.exclude or {}).get("paths")), + }, + "public": self.public, + "path": self.path, + } + + def to_dict(self) -> dict: + data = self.to_public_dict() + data["limits"] = dict(self.limits) + data["auth"] = self.auth # the variable name, never its value + return data + + +def load_guides(root: str, directory: str = None) -> dict: + """Load every guide under platforms/, keyed by name. Sorted, deterministic.""" + directory = directory or os.path.join(root, "platforms") + if not os.path.isdir(directory): + raise GuideError(f"platform guide directory not found: {directory}") + + validator = loader.load_validator(root) + guides = {} + for filename in sorted(os.listdir(directory)): + if not filename.endswith((".yml", ".yaml")): + continue + path = os.path.join(directory, filename) + rel = os.path.relpath(path, root).replace(os.sep, "/") + guide = _load_one(validator, path, rel, os.path.splitext(filename)[0]) + if guide.name in guides: + raise GuideError(f"{rel}: duplicate platform name {guide.name!r}") + guides[guide.name] = guide + + missing = sorted(set(KINDS) - set(guides)) + if missing: + raise GuideError( + f"no platform guide for {', '.join(missing)} under {directory}. " + "One small file per platform is required (docs/phase-4-corpus.md §2)." + ) + return guides + + +def _load_one(validator, path: str, rel: str, stem: str) -> Guide: + with open(path, "r", encoding="utf-8") as handle: + text = handle.read() + lines = [(number, line) for number, line in enumerate(text.splitlines(), start=1)] + try: + node = validator.parse_front_matter(lines) + except validator.YamlError as exc: + raise GuideError(f"{rel}:{exc.line}: invalid YAML: {exc.msg}") from exc + values = loader.to_python(node) + if not isinstance(values, dict): + raise GuideError(f"{rel}: a platform guide must be a mapping") + + for key in REQUIRED: + if values.get(key) in (None, "", []): + raise GuideError(f"{rel}: missing required key {key!r}") + + name = str(values["name"]).strip() + if name != stem: + raise GuideError(f"{rel}: name {name!r} does not match the filename {stem!r}") + if name not in KINDS: + raise GuideError(f"{rel}: unknown platform {name!r}; expected one of {list(KINDS)}") + + fits = [str(area) for area in values["fits"]] + unknown = [area for area in fits if area not in AREAS] + if unknown: + raise GuideError(f"{rel}: fits contains unknown area(s) {unknown}; allowed {list(AREAS)}") + + read = [str(item) for item in values["read"]] + if not read: + raise GuideError(f"{rel}: read must name at least one thing to harvest") + + include = _mapping(values.get("include"), f"{rel}: include") + exclude = _mapping(values.get("exclude"), f"{rel}: exclude") + limits = _mapping(values.get("limits"), f"{rel}: limits") + for key, value in limits.items(): + if isinstance(value, list): + # `graph_paths` is the one list-valued limit: the places a graph may + # be committed, tried in order. + continue + try: + limits[key] = int(value) + except (TypeError, ValueError): + raise GuideError(f"{rel}: limit {key!r} must be an integer, got {value!r}") from None + + return Guide( + name=name, + enabled=bool(values.get("enabled")), + what=str(values["what"]).strip(), + fits=fits, + read=read, + base=str(values.get("base") or "").strip(), + auth=str(values.get("auth") or "").strip(), + include=include, + exclude=exclude, + limits=limits, + public=bool(values.get("public", True)), + path=rel, + ) + + +def _mapping(value, where: str) -> dict: + if value is None: + return {} + if not isinstance(value, dict): + raise GuideError(f"{where} must be a mapping") + out = {} + for key, item in value.items(): + if isinstance(item, list): + out[key] = [str(entry) for entry in item] + else: + out[key] = item + return out + + +def _as_list(value) -> list: + """`repos: all` is a scalar; the public manifest still emits a list.""" + if value in (None, ""): + return [] + if isinstance(value, list): + return [str(item) for item in value] + return [str(value)] + + +def enabled_guides(guides: dict) -> list: + return [guide for _, guide in sorted(guides.items()) if guide.enabled] + + +def auth_value(guide: Guide, environ=None) -> str: + """Read a guide's token from the environment. Never from a file in the repo.""" + if not guide.auth: + return "" + environ = os.environ if environ is None else environ + return environ.get(guide.auth, "") + + +def build_public_manifest(guides: dict) -> dict: + """The guides as the public site shows them. Deterministic JSON shape.""" + return { + "version": 1, + "platforms": [guide.to_public_dict() for _, guide in sorted(guides.items())], + } diff --git a/corpus/loader.py b/corpus/loader.py index a8b6a65..8d958de 100644 --- a/corpus/loader.py +++ b/corpus/loader.py @@ -78,6 +78,16 @@ class Document: digest: str # sha256 of the file bytes: the incremental-mode key url: str # canonical page URL, matching internal/content.PageURL + # Provenance (docs/phase-4-corpus.md §3). Curated content is tier 1 and has + # no external origin; harvested documents set these in corpus/sources/. + platform: str = "curated" + tier: int = 1 + origin_url: str = "" + last_modified: str = "" + source_type: str = "" # empty means: derive from tags (payloads.source_type_for) + kind: str = "markdown" # "markdown" or "code", selects the chunker + repo_path: str = "" # origin-relative path; `path` above is the unique source key + # Filled in by the pipeline as it goes; not part of the schema. chunk_count: int = field(default=0) @@ -158,5 +168,13 @@ def _load_one(cfg, validator, path: str) -> Document: body=body, digest=digest_of(path), url=page_url(str(fields["area"]), str(fields["id"]), level), + platform="curated", + tier=1, + # A curated claim is traceable to the page it appears on. The URL is + # site-relative because the public origin is a deployment concern. + origin_url=page_url(str(fields["area"]), str(fields["id"]), level), + last_modified=f"{fields['last_verified']}T00:00:00Z", + kind="markdown", + repo_path=rel, ) return doc diff --git a/corpus/payloads.py b/corpus/payloads.py index 8fe1aca..d7b547f 100644 --- a/corpus/payloads.py +++ b/corpus/payloads.py @@ -16,8 +16,14 @@ import datetime AREAS = ("ai", "platform", "edge", "products", "history", "about") LEVELS = (1, 2, 3) -SOURCE_TYPES = ("doc", "code", "adr", "metric") +# docs/phase-4-corpus.md §3 fixes the harvested types; adr and metric were +# already used by curated content and stay valid. +SOURCE_TYPES = ( + "doc", "code", "commit", "graph_node", "graph_edge", "claim", "adr", "metric", +) CONFIDENCE = ("delivered", "running", "in-progress", "abandoned", "historic") +PLATFORMS = ("curated", "gitea", "obsidian", "graphify") +TIERS = (1, 2) # Fields every payload must carry (docs/phase-4-corpus.md §4.2), plus the ones # the index needs to answer "where did this come from?" without a second lookup. @@ -30,6 +36,13 @@ REQUIRED_FIELDS = ( "artefacts", "last_verified", "text", + # docs/phase-4-corpus.md §3: provenance is part of the payload, not an + # optional extra. A chunk that cannot say where it came from is not admitted. + "platform", + "tier", + "origin_url", + "path", + "last_modified", ) PROVENANCE_FIELDS = ( "id", @@ -51,11 +64,15 @@ class PayloadError(RuntimeError): def source_type_for(doc) -> str: - """Classify a document. Everything in content/ is a doc unless it says so. + """Classify a document. - The schema has no `source_type` field and the schema is frozen, so the - classification is derived from tags the schema does allow. + Harvested documents carry an explicit `source_type`. Curated content does + not: the schema is frozen and has no such field, so it is derived from the + tags the schema does allow. """ + explicit = str(getattr(doc, "source_type", "") or "").strip() + if explicit: + return explicit tags = {str(tag).lower() for tag in doc.tags} if "adr" in tags: return "adr" @@ -82,6 +99,12 @@ def build_payload(doc, chunk, cfg) -> dict: "artefacts": [str(a.get("url", "")) for a in doc.artefacts], "last_verified": doc.last_verified, "text": chunk.text, + # -- provenance (docs/phase-4-corpus.md §3) ------------------------ + "platform": doc.platform, + "tier": int(doc.tier), + "origin_url": doc.origin_url, + "path": getattr(doc, "repo_path", "") or doc.path, + "last_modified": doc.last_modified, # -- provenance ---------------------------------------------------- "id": None, # filled by the pipeline once the id is computed "source_path": doc.path, @@ -125,7 +148,17 @@ def validate_payload(payload: dict, where: str = "chunk") -> None: f"{where}: confidence {payload['confidence']!r} is not one of {list(CONFIDENCE)}" ) - for field in ("period", "title", "claim", "url", "source_path", "source_id", "embedding_model"): + if payload["platform"] not in PLATFORMS: + raise PayloadError( + f"{where}: platform {payload['platform']!r} is not one of {list(PLATFORMS)}" + ) + if payload["tier"] not in TIERS: + raise PayloadError(f"{where}: tier {payload['tier']!r} is not one of {list(TIERS)}") + + for field in ( + "period", "title", "claim", "url", "source_path", "source_id", + "embedding_model", "origin_url", "path", + ): value = payload[field] if not isinstance(value, str) or not value.strip(): raise PayloadError(f"{where}: field {field!r} must be a non-empty string") @@ -138,6 +171,7 @@ def validate_payload(payload: dict, where: str = "chunk") -> None: raise PayloadError(f"{where}: field 'artefacts' must be a list (empty is allowed)") _check_date(payload["last_verified"], where) + _check_timestamp(payload["last_modified"], where) if not isinstance(payload["embedding_dim"], int) or payload["embedding_dim"] <= 0: raise PayloadError(f"{where}: field 'embedding_dim' must be a positive integer") if not isinstance(payload["chunk_index"], int) or payload["chunk_index"] < 0: @@ -146,6 +180,19 @@ def validate_payload(payload: dict, where: str = "chunk") -> None: raise PayloadError(f"{where}: field 'content_hash' must be a sha256 hex digest") +def _check_timestamp(value: str, where: str) -> None: + """`last_modified` is an ISO-8601 instant, e.g. 2026-10-14T05:00:00Z.""" + text = str(value) + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + datetime.datetime.fromisoformat(text) + except ValueError as exc: + raise PayloadError( + f"{where}: field 'last_modified' must be an ISO-8601 timestamp, got {value!r}" + ) from exc + + def _check_date(value: str, where: str) -> None: try: datetime.date.fromisoformat(str(value)) diff --git a/corpus/pipeline.py b/corpus/pipeline.py index ae79f46..46e959c 100644 --- a/corpus/pipeline.py +++ b/corpus/pipeline.py @@ -1,20 +1,32 @@ -"""The pipeline: content/ -> validate -> chunk -> enrich -> embed -> upsert. +"""The ingestion pipeline: estate -> normalise -> chunk -> enrich -> embed -> upsert. One entry point, `run`, with two modes: -* **rebuild** — drop the collection, empty the store, embed everything. This is - the "one command rebuilds the corpus from an empty volume" path. +* **rebuild** — drop the collections, empty the store, embed everything. This is + the "one command rebuilds the pool from an empty volume" path. * **sync** — the default. Only documents whose content hash, embedding model, - embedding dimension or chunker version changed are re-chunked and - re-embedded. Everything else is left alone. + embedding dimension or chunker version changed are re-chunked and re-embedded. + Everything else is left alone, including the graph when graph.json is byte-for-byte + unchanged. -The order matters. Payloads are built and validated for *every* changed document -before a single vector is written, so an invalid payload stops the run with an -empty diff rather than leaving half an index behind. +The order matters. Every payload for every changed document is built and +validated *before* a single vector is written, so an invalid payload stops the +run with an empty diff rather than leaving half an index behind. Embedding is one +flat pass over the changed chunks, batched, so the provider call count is a +direct measure of how much work incremental mode actually did. Deterministic ids (`sha256(source_path + chunk_index + content_hash)`) mean re-running updates points instead of duplicating them, and the id is a valid UUID so Qdrant accepts it as a point id without a mapping table. + +Two Qdrant collections, one store: + +* `resume_corpus` — one point per chunk of a document, from every platform; +* `resume_graph` — one point per graph node and one per graph edge, joined to the + note it came from by `source_note`. + +Neither is ever read by the public site (acceptance test A1): the site serves +pre-rendered files and does not open a socket to Qdrant. """ from __future__ import annotations @@ -25,9 +37,15 @@ import time import uuid from dataclasses import asdict, dataclass, field -from . import chunker, embeddings, loader, payloads, store as store_module +from . import chunker, embeddings, exclusions, payloads, store as store_module from .config import CHUNKER_VERSION from .qdrant import Qdrant, QdrantError +from .sources import estate as estate_module +from .sources import graph as graph_module + +# How many texts one embed call carries. The provider may batch smaller +# internally (OmniRoute does); this bounds memory on the ingest host. +EMBED_BATCH = 128 class PipelineError(RuntimeError): @@ -61,12 +79,22 @@ class RunStats: points_sent: int = 0 collection_points: int = 0 store_chunks: int = 0 + graph_nodes: int = 0 + graph_edges: int = 0 + graph_points: int = 0 + graph_rebuilt: bool = False + rejections: int = 0 + images: int = 0 duration_s: float = 0.0 collection: str = "" + graph_collection: str = "" provider: str = "" model: str = "" dim: int = 0 chunker_version: str = CHUNKER_VERSION + platforms: dict = field(default_factory=dict) + rejection_summary: dict = field(default_factory=dict) + warnings: list = field(default_factory=list) notes: list = field(default_factory=list) def summary(self) -> str: @@ -76,10 +104,18 @@ class RunStats: f"chunks written={self.chunks_written} deleted={self.chunks_deleted}", f"embedding calls={self.embed_calls} texts={self.texts_embedded}", f"points sent={self.points_sent} collection points={self.collection_points}", + f"graph nodes={self.graph_nodes} edges={self.graph_edges} points={self.graph_points} " + f"rebuilt={self.graph_rebuilt}", + f"rejections={self.rejections} image references={self.images}", + f"platforms={json.dumps(self.platforms, sort_keys=True)}", f"provider={self.provider} model={self.model} dim={self.dim} chunker={self.chunker_version}", - f"collection={self.collection} duration={self.duration_s:.2f}s", + f"collection={self.collection} graph_collection={self.graph_collection} " + f"duration={self.duration_s:.2f}s", ] + if self.rejection_summary: + rows.append(f"rejected: {json.dumps(self.rejection_summary, sort_keys=True)}") rows.extend(self.notes) + rows.extend(f"warning: {warning}" for warning in self.warnings) return "\n".join(rows) def to_dict(self) -> dict: @@ -87,7 +123,7 @@ class RunStats: def run(cfg, mode: str = "sync", log=None) -> RunStats: - """Build or refresh the corpus. Returns what happened; raises on any error.""" + """Build or refresh the pool. Returns what happened; raises on any error.""" if mode not in ("sync", "rebuild"): raise PipelineError(f"unknown mode {mode!r}; expected 'sync' or 'rebuild'") started = time.time() @@ -95,15 +131,26 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats: stats = RunStats( mode=mode, collection=cfg.collection, + graph_collection=cfg.graph_collection, provider=cfg.provider, model=cfg.model, dim=cfg.dim, chunker_version=cfg.chunker_version, ) - documents = loader.load_documents(cfg) + # 1. Read the estate. Curated content is validated first (loader does it), + # so a broken content tree stops the run before Gitea is touched. + harvest = estate_module.harvest(cfg, log=log) + documents = harvest.documents stats.documents = len(documents) - log(f"validated {len(documents)} document(s)") + stats.rejections = len(harvest.rejections) + stats.images = len(harvest.images) + stats.rejection_summary = harvest.rejection_summary() + stats.warnings = list(harvest.warnings) + stats.platforms = { + name: {"documents": row.get("documents", 0)} for name, row in harvest.sources.items() + } + log(f"estate: {len(documents)} document(s), {stats.rejections} rejection(s), {stats.images} image reference(s)") provider = embeddings.build_provider(cfg) qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout) @@ -116,8 +163,7 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats: with store_module.Store(cfg.db_path) as store: contract = _contract(cfg) mode = _check_contract(store, qdrant, cfg, contract, mode) - - _ensure_collection(qdrant, cfg) + _ensure_collection(qdrant, cfg.collection, cfg) manifest = store.manifest() changed, removed = _classify(documents, manifest, cfg, mode) @@ -133,33 +179,46 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats: qdrant.delete_by_source(cfg.collection, path) log(f"removed {path}") + # 2. Build and validate every payload before any write. + prepared = [] # [(doc, payloads)] for doc in changed: - payloads_for_doc = _prepare(doc, cfg, manifest) - stats.chunks_written += len(payloads_for_doc) + prepared.append((doc, _prepare(doc, cfg))) + all_payloads = [payload for _, rows in prepared for payload in rows] - vectors = provider.embed([p["text"] for p in payloads_for_doc]) - if len(vectors) != len(payloads_for_doc): - raise PipelineError( - f"{doc.path}: provider returned {len(vectors)} vectors for " - f"{len(payloads_for_doc)} chunks" - ) + # 3. Embed in one flat, batched pass. This is the number the DoD uses to + # prove incremental mode: it is proportional to changed chunks only. + vectors = _embed_all(provider, [payload["text"] for payload in all_payloads]) + if len(vectors) != len(all_payloads): + raise PipelineError( + f"provider returned {len(vectors)} vectors for {len(all_payloads)} chunks" + ) + cursor = 0 + platform_changes = stats.platforms + for doc, rows in prepared: points = [] - for payload, vector in zip(payloads_for_doc, vectors): + for payload in rows: + vector = vectors[cursor] + cursor += 1 if len(vector) != cfg.dim: raise PipelineError( f"{doc.path}: vector dimension {len(vector)} != configured {cfg.dim}" ) points.append({"id": payload["id"], "vector": vector, "payload": payload}) - # Order: remove what this document used to have (chunk counts change), - # then write the new rows and points. A crash between the two leaves an - # index that reports as incomplete, never one that claims to be whole. + # Remove what this document used to have, then write the new rows and + # points. A crash between the two leaves an index that reports as + # incomplete, never one that claims to be whole. stats.chunks_deleted += store.delete_document(doc.path) qdrant.delete_by_source(cfg.collection, doc.path) stats.points_sent += qdrant.upsert(cfg.collection, points) - store.write_document(doc, payloads_for_doc) + store.write_document(doc, rows) + stats.chunks_written += len(rows) + bucket = platform_changes.setdefault(doc.platform, {"documents": 0}) + bucket["added" if doc.path in manifest else "updated"] = ( + bucket.get("added" if doc.path in manifest else "updated", 0) + 1 + ) if doc.path in manifest: stats.updated += 1 log(f"updated {doc.path} ({len(points)} chunk(s))") @@ -170,24 +229,47 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats: provider_stats = provider.stats() stats.embed_calls = provider_stats["calls"] stats.texts_embedded = provider_stats["texts"] + + # 4. Graph: its own collection, rebuilt only when graph.json changed. + graph_stats = _sync_graph(qdrant, store, cfg, harvest, contract, mode, provider, log) + stats.graph_nodes = graph_stats["nodes"] + stats.graph_edges = graph_stats["edges"] + stats.graph_points = graph_stats["points"] + stats.graph_rebuilt = graph_stats["rebuilt"] + stats.notes.extend(graph_stats["notes"]) + + # 5. Consistency and the credential scan. Both are checks, not hopes. stats.store_chunks = store.chunk_count() stats.collection_points = qdrant.point_count(cfg.collection) - if stats.collection_points != stats.store_chunks: raise PipelineError( "index is inconsistent after the run: " f"{stats.collection_points} point(s) in Qdrant but " f"{stats.store_chunks} chunk(s) in SQLite. Run `python3 -m corpus.cli verify`." ) + _scan_pool(store) + _scan_graph(store) store.set_meta("collection", contract) - store.set_meta("last_run", {**stats.to_dict(), "finished_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}) - _write_collection_file(cfg, store, stats) + store.set_meta("sources", harvest.sources) + store.set_meta("rejections", harvest.rejections[:200]) + store.set_meta("images", harvest.images[:500]) + store.set_meta( + "last_run", + { + **stats.to_dict(), + "finished_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + }, + ) + _write_collection_file(cfg, store, stats, harvest) stats.duration_s = round(time.time() - started, 3) return stats +# -- planning ------------------------------------------------------------- + + def _contract(cfg) -> dict: """What this collection *is*. Recorded so a change is a migration, not drift.""" return { @@ -200,13 +282,24 @@ def _contract(cfg) -> dict: def _check_contract(store, qdrant, cfg, contract: dict, mode: str) -> str: - stored = store.get_meta("collection") if mode == "rebuild": qdrant.drop_collection(cfg.collection) - store.reset() + qdrant.drop_collection(cfg.graph_collection) + if store.schema_stale: + store.rebuild_schema() + else: + store.reset() + store.set_meta("schema_version", store_module.SCHEMA_VERSION) return mode + + if store.schema_stale: + raise PipelineError( + "the corpus database was written by an older payload shape and has no " + "provenance columns. It is derived data: run `python3 -m corpus.cli rebuild`." + ) + + stored = store.get_meta("collection") if stored is None: - # No contract yet: either a fresh store, or one whose database was lost. if qdrant.collection_exists(cfg.collection) and qdrant.point_count(cfg.collection) > 0: raise PipelineError( f"collection {cfg.collection!r} holds points but this store has no " @@ -217,7 +310,7 @@ def _check_contract(store, qdrant, cfg, contract: dict, mode: str) -> str: if stored != contract: raise PipelineError( "the collection contract does not match this configuration — refusing to " - "mix vectors in one collection (docs/phase-4-corpus.md §4.3).\n" + "mix vectors in one collection (docs/phase-4-corpus.md §4).\n" f" stored: {json.dumps(stored, sort_keys=True)}\n" f" config: {json.dumps(contract, sort_keys=True)}\n" "A provider or chunker change is a governed migration: run " @@ -226,20 +319,18 @@ def _check_contract(store, qdrant, cfg, contract: dict, mode: str) -> str: return mode -def _ensure_collection(qdrant: Qdrant, cfg) -> None: - if not qdrant.collection_exists(cfg.collection): - qdrant.create_collection( - cfg.collection, cfg.dim, vectors_on_disk=cfg.vectors_on_disk - ) +def _ensure_collection(qdrant: Qdrant, name: str, cfg) -> None: + if not qdrant.collection_exists(name): + qdrant.create_collection(name, cfg.dim, vectors_on_disk=cfg.vectors_on_disk) return - info = qdrant.collection_info(cfg.collection) + info = qdrant.collection_info(name) vectors = ((info.get("config") or {}).get("params") or {}).get("vectors") size = None if isinstance(vectors, dict): size = vectors.get("size") if size is not None and int(size) != int(cfg.dim): raise PipelineError( - f"collection {cfg.collection!r} has vectors of size {size} but this run " + f"collection {name!r} has vectors of size {size} but this run " f"is configured for {cfg.dim}. That is a governed migration: rebuild." ) @@ -268,10 +359,13 @@ def _classify(documents: list, manifest: dict, cfg, mode: str): return changed, removed -def _prepare(doc, cfg, manifest: dict) -> list: +def _prepare(doc, cfg) -> list: """Chunk one document and build validated payloads. No network, no writes.""" try: - chunks = chunker.chunk_document(doc.body) + if getattr(doc, "kind", "markdown") == "code": + chunks = chunker.chunk_code(doc.body) + else: + chunks = chunker.chunk_document(doc.body) except chunker.ChunkError as exc: raise PipelineError(f"{doc.path}: {exc}") from exc if not chunks: @@ -290,14 +384,118 @@ def _prepare(doc, cfg, manifest: dict) -> list: return out -def _write_collection_file(cfg, store, stats: RunStats) -> None: - """Mirror the contract to data/corpus/collection.json for the operator.""" +def _embed_all(provider, texts: list) -> list: + out = [] + for start in range(0, len(texts), EMBED_BATCH): + out.extend(provider.embed(texts[start : start + EMBED_BATCH])) + return out + + +# -- graph ---------------------------------------------------------------- + + +def _sync_graph(qdrant, store, cfg, harvest, contract, mode, provider, log) -> dict: + graph = harvest.graph + stored = store.get_meta("graph") or {} + notes = [] + if graph is None: + if mode == "rebuild": + qdrant.drop_collection(cfg.graph_collection) + store.write_graph([], []) + store.set_meta("graph", {}) + notes.append("graph: no graph.json was available; the graph collection is empty") + return {"nodes": 0, "edges": 0, "points": 0, "rebuilt": True, "notes": notes} + counts = store.graph_counts() + notes.append( + f"graph: graph.json was not available this run; kept the previous graph " + f"({counts['nodes']} nodes, {counts['edges']} edges)" + ) + return {**counts, "points": counts["nodes"] + counts["edges"], "rebuilt": False, "notes": notes} + + graph_contract = { + "digest": graph.digest, + "source_path": graph.source_path, + "provider": contract["provider"], + "model": contract["model"], + "dim": contract["dim"], + } + rebuild = mode == "rebuild" or stored != graph_contract + if not rebuild: + counts = store.graph_counts() + return { + **counts, + "points": counts["nodes"] + counts["edges"], + "rebuilt": False, + "notes": [f"graph: unchanged ({graph.digest[:12]}), not re-embedded"], + } + + if not qdrant.collection_exists(cfg.graph_collection): + _ensure_collection(qdrant, cfg.graph_collection, cfg) + else: + qdrant.drop_collection(cfg.graph_collection) + qdrant.create_collection(cfg.graph_collection, cfg.dim, vectors_on_disk=cfg.vectors_on_disk) + + rows = [("node", row) for row in graph.nodes] + [("edge", row) for row in graph.edges] + payloads_for_graph = [graph_module.point_payload(row, kind) for kind, row in rows] + vectors = _embed_all(provider, [payload["text"] for payload in payloads_for_graph]) + points = [ + {"id": row["point_id"], "vector": vector, "payload": payload} + for (_, row), vector, payload in zip(rows, vectors, payloads_for_graph) + ] + sent = qdrant.upsert(cfg.graph_collection, points) + store.write_graph(graph.nodes, graph.edges) + store.set_meta("graph", graph_contract) + log(f"graph: rebuilt {len(graph.nodes)} node(s) and {len(graph.edges)} edge(s), {sent} point(s)") + return { + "nodes": len(graph.nodes), + "edges": len(graph.edges), + "points": sent, + "rebuilt": True, + "notes": notes, + } + + +# -- checks --------------------------------------------------------------- + + +def _scan_pool(store) -> None: + """Search the finished pool for credential-shaped content. Fail, do not warn.""" + hits = [] + for source_path, chunk_index, text in store.pool_texts(): + for kind, line, _excerpt in exclusions.scan_text(text): + hits.append((source_path, chunk_index, kind, line)) + if hits: + detail = ", ".join(f"{path}#chunk-{index} ({kind} line {line})" for path, index, kind, line in hits[:10]) + raise PipelineError( + f"credential-shaped content reached the pool in {len(hits)} place(s): {detail}. " + "The run is failed rather than published; fix the source or the exclusion rule." + ) + + +def _scan_graph(store) -> None: + rows = store.db.execute("SELECT id, text FROM graph_nodes UNION ALL SELECT id, text FROM graph_edges") + hits = [] + for row in rows: + for kind, line, _excerpt in exclusions.scan_text(row["text"]): + hits.append((row["id"], kind, line)) + if hits: + detail = ", ".join(f"{node} ({kind} line {line})" for node, kind, line in hits[:10]) + raise PipelineError(f"credential-shaped content reached the graph in {len(hits)} place(s): {detail}") + + +def _write_collection_file(cfg, store, stats: RunStats, harvest) -> None: + """Mirror the contract and the last run to data/corpus/collection.json.""" body = { "collection": cfg.collection, "contract": _contract(cfg), "qdrant_url": cfg.qdrant_url, "chunks": store.chunk_count(), "sources": store.source_count(), + "graph_collection": cfg.graph_collection, + "graph": store.graph_counts(), + "platforms": harvest.sources, + "rejections": harvest.rejection_summary(), + "images": len(harvest.images), "last_run": store.get_meta("last_run"), } with open(cfg.meta_path, "w", encoding="utf-8") as handle: @@ -309,10 +507,11 @@ def _write_collection_file(cfg, store, stats: RunStats) -> None: def verify(cfg, log=None) -> list: - """Return a list of problems. Empty means the index agrees with content/.""" + """Return a list of problems. Empty means the index agrees with the sources.""" log = log or (lambda _msg: None) problems = [] - documents = loader.load_documents(cfg) + harvest = estate_module.harvest(cfg) + documents = harvest.documents qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout) if not qdrant.health(): return [f"Qdrant is not answering at {cfg.qdrant_url}"] @@ -342,6 +541,7 @@ def verify(cfg, log=None) -> list: except payloads.PayloadError as exc: problems.append(str(exc)) + on_disk = {doc.path: doc for doc in documents} for doc in documents: payload_rows = by_source.get(doc.path, []) if not payload_rows: @@ -356,18 +556,48 @@ def verify(cfg, log=None) -> list: f"{doc.path}: {expected} chunk row(s) but {actual} vector(s) in Qdrant" ) - for source in sorted(set(by_source) - {doc.path for doc in documents}): - problems.append(f"{source}: indexed but no longer in content/ (run sync)") + for source in sorted(set(by_source) - set(on_disk)): + problems.append(f"{source}: indexed but no longer in content/ or the estate (run sync)") total_points = sum(points_by_source.values()) if total_points != store.chunk_count(): problems.append( f"{total_points} vector(s) in Qdrant but {store.chunk_count()} chunk(s) in SQLite" ) + + # Graph: every row must resolve to a note, and Qdrant must agree. + counts = store.graph_counts() + graph_meta = store.get_meta("graph") or {} + if harvest.graph is not None: + if not graph_meta: + problems.append("graph: graph.json is available but the graph was never indexed") + if qdrant.collection_exists(cfg.graph_collection): + graph_points = qdrant.point_count(cfg.graph_collection) + if graph_points != counts["nodes"] + counts["edges"]: + problems.append( + f"graph: {graph_points} point(s) in Qdrant but " + f"{counts['nodes'] + counts['edges']} row(s) in SQLite" + ) + elif counts["nodes"] or counts["edges"]: + problems.append(f"graph: collection {cfg.graph_collection!r} does not exist") + for row in store.db.execute("SELECT id, source_note, origin_url FROM graph_nodes LIMIT 5"): + if not row["source_note"] or not row["origin_url"]: + problems.append(f"graph node {row['id']}: no source note or origin") + missing_notes = [ + row["id"] + for row in store.db.execute( + "SELECT id FROM graph_nodes WHERE source_note = '' OR origin_url = '' LIMIT 10" + ) + ] + if missing_notes: + problems.append(f"graph: {len(missing_notes)} node(s) carry no source note") + log( f"verify: {store.chunk_count()} chunk(s) across {store.source_count()} source(s), " - f"{total_points} vector(s) in Qdrant" + f"{total_points} vector(s) in Qdrant, {counts['nodes']} node(s), {counts['edges']} edge(s)" ) + for warning in harvest.warnings: + problems.append(f"estate warning: {warning}") return problems @@ -391,6 +621,8 @@ def _hit(row: dict) -> dict: return { "score": round(row["score"], 4), "area": payload.get("area"), + "platform": payload.get("platform"), + "tier": payload.get("tier"), "confidence": payload.get("confidence"), "title": payload.get("title"), "url": payload.get("url"), @@ -400,7 +632,7 @@ def _hit(row: dict) -> dict: } -def hybrid_search(cfg, query: str, area: str = None, limit: int = 5) -> list: +def hybrid_search(cfg, query: str, area: str = None, limit: int = 5, platform: str = None, tier: int = None) -> list: """Vectors + FTS5 fused with reciprocal rank fusion (decision D10). Phase 5 owns the query graph; this is here because Phase 4 is the phase that @@ -410,11 +642,15 @@ def hybrid_search(cfg, query: str, area: str = None, limit: int = 5) -> list: qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout) vector = provider.embed([query])[0] ranked = {} - for rank, hit in enumerate(qdrant.search(cfg.collection, vector, limit=limit * 4, area=area)): + for rank, hit in enumerate( + qdrant.search(cfg.collection, vector, limit=limit * 4, area=area, platform=platform, tier=tier) + ): entry = ranked.setdefault(hit["id"], {"id": hit["id"], "payload": hit["payload"], "score": 0.0}) entry["score"] += 1.0 / (60 + rank) with store_module.Store(cfg.db_path) as store: - for rank, hit in enumerate(store.keyword_search(query, area=area, limit=limit * 4)): + for rank, hit in enumerate( + store.keyword_search(query, area=area, platform=platform, tier=tier, limit=limit * 4) + ): entry = ranked.setdefault(hit["id"], {"id": hit["id"], "payload": {}, "score": 0.0}) entry.setdefault("keyword", hit["snippet"]) entry["score"] += 1.0 / (60 + rank) @@ -424,3 +660,21 @@ def hybrid_search(cfg, query: str, area: str = None, limit: int = 5) -> list: row["title"] = payload.get("title") row["url"] = payload.get("url") return ordered + + +def scan_pool(cfg) -> dict: + """Run the credential scan without writing anything. Used by `make corpus-scan`.""" + with store_module.Store(cfg.db_path) as store: + hits = [] + texts = store.pool_texts() + for source_path, chunk_index, text in texts: + for kind, line, _excerpt in exclusions.scan_text(text): + hits.append( + {"source_path": source_path, "chunk": chunk_index, "kind": kind, "line": line} + ) + try: + _scan_graph(store) + except PipelineError as exc: + hits.append({"source_path": "graph", "kind": "credential-shaped content", "detail": str(exc)}) + chunks = len(texts) + return {"ok": not hits, "hits": hits, "chunks": chunks} diff --git a/corpus/qdrant.py b/corpus/qdrant.py index 31370eb..0259443 100644 --- a/corpus/qdrant.py +++ b/corpus/qdrant.py @@ -198,6 +198,8 @@ class Qdrant: limit: int = 5, area: str = None, confidence: str = None, + platform: str = None, + tier: int = None, with_payload: bool = True, ) -> list: """Nearest neighbours, optionally filtered *before* the search.""" @@ -206,7 +208,7 @@ class Qdrant: "limit": int(limit), "with_payload": bool(with_payload), } - filters = _filter_conditions(area, confidence) + filters = _filter_conditions(area, confidence, platform, tier) if filters: body["filter"] = {"must": filters} try: @@ -260,10 +262,16 @@ class QdrantNotFound(QdrantError): """A 404 from Qdrant: the collection or route does not exist.""" -def _filter_conditions(area: str = None, confidence: str = None) -> list: +def _filter_conditions( + area: str = None, confidence: str = None, platform: str = None, tier: int = None +) -> list: conditions = [] if area: conditions.append({"key": "area", "match": {"value": area}}) if confidence: conditions.append({"key": "confidence", "match": {"value": confidence}}) + if platform: + conditions.append({"key": "platform", "match": {"value": platform}}) + if tier is not None: + conditions.append({"key": "tier", "match": {"value": int(tier)}}) return conditions diff --git a/corpus/sources/estate.py b/corpus/sources/estate.py new file mode 100644 index 0000000..09da499 --- /dev/null +++ b/corpus/sources/estate.py @@ -0,0 +1,804 @@ +"""Harvest the estate into one set of documents. + +Four sources, one shape (`loader.Document`), one tier rule: + +| Platform | Tier | Reached by | +|---|---|---| +| `curated` (`content/`) | 1 | the local repository, already validated | +| `gitea` | 2 | the Gitea API | +| `obsidian` | 2 | the same Gitea API, one repository | +| `graphify` | 2 | `graph.json` committed to the vault repository | + +`content/` is loaded first and is never displaced: the query layer uses `tier` +to prefer a curated claim over harvested text that disagrees with it. + +Every refused file is recorded with the rule that refused it. The DoD asks for +this explicitly ("the run reports the rejection"), and it is the difference +between an exclusion list that works and one nobody can audit. +""" + +from __future__ import annotations + +import datetime +import hashlib +import json +import os +import re + +from .. import areas, exclusions, globs, guides, loader +from . import graph as graph_module +from .gitea import GiteaClient, GiteaError, GiteaNotFound, file_url, repo_url + +PLATFORM_ORDER = ("curated", "gitea", "obsidian", "graphify") + + +class EstateError(RuntimeError): + """The estate could not be read. The run stops rather than indexing half of it.""" + + +class Harvest: + def __init__(self) -> None: + self.documents: list = [] + self.rejections: list = [] + self.images: list = [] + self.warnings: list = [] + self.sources: dict = {} + self.graph = None # graph_module.GraphResult | None + self.offline = False + + # -- reporting -------------------------------------------------------- + + def add_rejection(self, rejection) -> None: + self.rejections.append(rejection.to_dict() if hasattr(rejection, "to_dict") else rejection) + + def rejection_summary(self) -> dict: + counts = {} + for row in self.rejections: + key = f"{row.get('platform')}/{row.get('reason')}" + counts[key] = counts.get(key, 0) + 1 + return dict(sorted(counts.items())) + + def to_dict(self) -> dict: + return { + "documents": len(self.documents), + "images": len(self.images), + "rejections": len(self.rejections), + "rejection_summary": self.rejection_summary(), + "warnings": list(self.warnings), + "sources": self.sources, + "graph": self.graph.counts() if self.graph else None, + "offline": self.offline, + } + + +def harvest(cfg, log=None) -> Harvest: + """Read every enabled platform. Curated content is always read.""" + log = log or (lambda _msg: None) + result = Harvest() + result.offline = bool(cfg.offline) + + curated = loader.load_documents(cfg) + result.documents.extend(curated) + result.sources["curated"] = _source_row( + "curated", enabled=True, documents=curated, extra={"path": "content/"} + ) + log(f"curated: {len(curated)} document(s) from content/") + + if cfg.offline: + # Not a warning: offline is a deliberate mode, not a failure. + log("offline mode: only content/ was read (RESUME_OFFLINE=1)") + return result + + all_guides = guides.load_guides(cfg.root) + selected = { + name: guide + for name, guide in all_guides.items() + if guide.enabled and name in cfg.platforms + } + missing = [name for name in cfg.platforms if name not in all_guides] + if missing: + result.warnings.append(f"platform(s) {missing} have no guide and were skipped") + + needs_gitea = {"gitea", "obsidian", "graphify"} & set(selected) + if not needs_gitea: + return result + + base = cfg.gitea_url or selected.get("gitea", all_guides.get("gitea")).base + client = GiteaClient(base, cfg.gitea_token, timeout=cfg.timeout) + try: + repos = client.repos() + except GiteaError as exc: + raise EstateError(f"cannot list Gitea repositories: {exc}") from exc + log(f"gitea: {len(repos)} repositories visible") + + vault_repo = _vault_repo(selected) + vault_tree = None + if vault_repo: + try: + vault_tree = client.tree(vault_repo, "main") + except GiteaNotFound: + try: + vault_tree = client.tree(vault_repo, "master") + except GiteaError as exc: + raise EstateError(f"cannot read {vault_repo}: {exc}") from exc + except GiteaError as exc: + raise EstateError(f"cannot read {vault_repo}: {exc}") from exc + + if "gitea" in selected: + _harvest_repos(result, client, selected["gitea"], repos, base, log) + if "obsidian" in selected: + _harvest_obsidian(result, client, selected["obsidian"], repos, vault_repo, vault_tree, cfg, base, log) + if "graphify" in selected: + _harvest_graph(result, client, selected["graphify"], repos, vault_repo, vault_tree, cfg, base, log) + + return result + + +# -- gitea repositories --------------------------------------------------- + + +def _harvest_repos(result: Harvest, client, guide, repos, base_url, log) -> None: + excluded = set(_as_list((guide.exclude or {}).get("repos"))) + include = (guide.include or {}).get("repos") or "all" + wanted = None if include == "all" or _as_list(include) == [] else set(_as_list(include)) + + documents = [] + seen_repos = set() + for repo in repos: + full_name = repo.get("full_name") or "" + if not full_name or full_name in excluded: + continue + if wanted is not None and full_name not in wanted: + continue + seen_repos.add(full_name) + try: + documents.extend(_harvest_one_repo(result, client, guide, repo, base_url)) + except GiteaError as exc: + result.warnings.append(f"gitea/{full_name}: {exc}") + continue + if len(documents) >= guide.limit("max_documents"): + result.warnings.append( + f"gitea: document limit {guide.limit('max_documents')} reached; " + "remaining repositories were skipped" + ) + break + + result.documents.extend(documents) + result.sources["gitea"] = _source_row( + "gitea", enabled=True, documents=documents, + extra={"repos": len(seen_repos), "base": base_url}, + ) + log(f"gitea: {len(documents)} document(s), {len(result.rejections)} rejection(s) so far") + + +def _harvest_one_repo(result: Harvest, client, guide, repo, base_url) -> list: + full_name = repo["full_name"] + branch = repo.get("default_branch") or "main" + updated = _timestamp(repo.get("updated_at")) + created = _timestamp(repo.get("created_at")) + area = areas.area_for_repo(repo, guide.fits) + topics = [str(topic) for topic in (repo.get("topics") or [])] + description = str(repo.get("description") or "").strip() + confidence = "abandoned" if repo.get("archived") else "historic" + + documents = [] + try: + entries = client.tree(full_name, branch) + except GiteaNotFound: + entries = [] + + blobs = [entry for entry in entries if entry.get("type") == "blob"] + admitted = [] + for entry in blobs: + path = globs.normalise(entry.get("path") or "") + kind, reason = exclusions.classify(path, guide, size=entry.get("size")) + if kind == "reject": + result.add_rejection(exclusions.Rejection(path, "gitea", reason, full_name)) + continue + if kind == "image": + result.images.append( + {"path": path, "platform": "gitea", "repo": full_name, + "origin_url": file_url(base_url, full_name, branch, path)} + ) + continue + admitted.append(entry) + + # The file tree is paths only. It is what lets a question like "does he have + # a Prometheus config anywhere?" find a repository whose README never says so. + tree_doc = _tree_document(guide, repo, area, updated, admitted, base_url) + if tree_doc: + documents.append(tree_doc) + + # README first: it is the document a reader would open. + readmes = [ + entry for entry in admitted + if _is_readme(entry.get("path") or "") and globs.normalise(entry.get("path") or "").count("/") <= 1 + ][:3] + used = set() + for entry in readmes: + doc = _file_document( + result, client, guide, repo, branch, area, updated, created, + confidence, description, topics, entry, base_url, + ) + if doc: + documents.append(doc) + used.add(globs.normalise(entry["path"])) + + # Selected source files: the evidence behind the README's claims. + candidates = [ + entry for entry in admitted + if globs.normalise(entry.get("path") or "") not in used + and not _is_readme(entry.get("path") or "") + and _source_rank(entry.get("path") or "") + ] + candidates.sort(key=lambda entry: (_depth(entry.get("path") or ""), globs.normalise(entry["path"]))) + for entry in candidates[: guide.limit("max_files_per_repo")]: + doc = _file_document( + result, client, guide, repo, branch, area, updated, created, + confidence, description, topics, entry, base_url, + ) + if doc: + documents.append(doc) + + # One document per repository for commits and one for languages. These are + # short, stable, and answer "what is being worked on?" without a filesystem. + commits = _commit_document(result, client, guide, repo, area, updated, created, confidence, base_url) + if commits: + documents.append(commits) + languages = _language_document(result, client, guide, repo, area, updated, created, confidence, base_url) + if languages: + documents.append(languages) + + if not any(not doc.source_id.endswith(("@tree", "@commits", "@languages")) for doc in documents): + # No README and no admitted source file: still admit a summary so the + # repository is discoverable and its silence is visible, not invisible. + summary = _summary_document(guide, repo, area, updated, created, confidence, description, topics, base_url) + if summary: + documents.insert(0, summary) + + return documents + + +def _file_document(result, client, guide, repo, branch, area, updated, created, confidence, description, topics, entry, base_url) -> object: + path = globs.normalise(entry.get("path") or "") + full_name = repo["full_name"] + try: + raw = client.raw(full_name, branch, path) + except GiteaError as exc: + result.warnings.append(f"gitea/{full_name}/{path}: {exc}") + return None + if exclusions.looks_binary(raw): + result.add_rejection(exclusions.Rejection(path, "gitea", "binary content", full_name)) + return None + text = raw.decode("utf-8", "replace") + hits = exclusions.scan_text(text) + if hits: + kinds = sorted({hit[0] for hit in hits}) + result.add_rejection( + exclusions.Rejection(path, "gitea", "credential-shaped content", full_name + ": " + ", ".join(kinds)) + ) + return None + + kind = "code" if exclusions.extension(path) not in ("md", "markdown", "mdx", "txt", "rst") else "markdown" + title = f"{full_name} — {path}" + body = text + if description or topics: + header = [f"Repository: {full_name}"] + if description: + header.append(description) + if topics: + header.append("Topics: " + ", ".join(topics)) + body = "\n\n".join(["\n".join(header), text]) + return _make_document( + key=f"gitea:{full_name}/{path}", + platform="gitea", + area=area, + title=title, + claim=_claim(text, title), + body=body, + digest=hashlib.sha256(raw).hexdigest(), + origin_url=file_url(base_url, full_name, branch, path), + path=path, + repo_path=path, + source_id=f"{full_name}/{path}", + source_type="code" if kind == "code" else "doc", + last_modified=updated, + confidence=confidence, + period=_period(created, updated), + tags=topics + ["gitea", area], + tech=[], + public=guide.public, + kind=kind, + ) + + +def _tree_document(guide, repo, area, updated, blobs, base_url) -> object: + if not blobs: + return None + full_name = repo["full_name"] + branch = repo.get("default_branch") or "main" + limit = guide.limit("max_tree_entries") + paths = [globs.normalise(entry.get("path") or "") for entry in blobs][:limit] + body = "\n".join( + [f"Repository: {full_name}", f"Branch: {branch}", f"Files: {len(blobs)}", "", "File tree:"] + + [f"- {path}" for path in paths] + ) + if len(blobs) > limit: + body += f"\n- … {len(blobs) - limit} more" + digest = hashlib.sha256(body.encode("utf-8")).hexdigest() + return _make_document( + key=f"gitea:{full_name}@tree", + platform="gitea", + area=area, + title=f"{full_name} — file tree", + claim=f"File tree of {full_name}, {len(blobs)} files.", + body=body, + digest=digest, + origin_url=repo_url(base_url, full_name), + path="", + repo_path="", + source_id=f"{full_name}@tree", + source_type="doc", + last_modified=updated, + confidence="abandoned" if repo.get("archived") else "historic", + period="", + tags=["gitea", "tree", area], + tech=[], + public=guide.public, + kind="markdown", + ) + + +def _commit_document(result, client, guide, repo, area, updated, created, confidence, base_url) -> object: + full_name = repo["full_name"] + branch = repo.get("default_branch") or "main" + try: + commits = client.commits(full_name, branch, limit=guide.limit("max_commits")) + except GiteaError: + return None + if not commits: + return None + lines = [f"Repository: {full_name}", f"Recent commits on {branch}:", ""] + newest = updated + for item in commits: + commit = item.get("commit") or {} + message = (commit.get("message") or "").strip().splitlines() + headline = message[0] if message else "(no message)" + date = _timestamp(((commit.get("author") or {}).get("date")) or ((commit.get("committer") or {}).get("date"))) + if date: + newest = max(newest, date) + lines.append(f"- {date[:10]} {headline}") + body = "\n".join(lines) + digest = hashlib.sha256(body.encode("utf-8")).hexdigest() + return _make_document( + key=f"gitea:{full_name}@commits", + platform="gitea", + area=area, + title=f"{full_name} — recent commits", + claim=f"Recent commit history for {full_name}.", + body=body, + digest=digest, + origin_url=f"{repo_url(base_url, full_name)}/commits/branch/{branch}", + path="", + repo_path="", + source_id=f"{full_name}@commits", + source_type="commit", + last_modified=newest, + confidence=confidence, + period=_period(created, newest), + tags=["gitea", "commits", area], + tech=[], + public=guide.public, + kind="markdown", + ) + + +def _language_document(result, client, guide, repo, area, updated, created, confidence, base_url) -> object: + full_name = repo["full_name"] + try: + languages = client.languages(full_name) + except GiteaError: + return None + if not languages: + return None + total = sum(int(value) for value in languages.values()) or 1 + rows = sorted(languages.items(), key=lambda item: int(item[1]), reverse=True) + lines = [f"Repository: {full_name}", "", "Language statistics:"] + for name, size in rows: + lines.append(f"- {name}: {int(size)} bytes ({100.0 * int(size) / total:.1f}%)") + body = "\n".join(lines) + digest = hashlib.sha256(body.encode("utf-8")).hexdigest() + return _make_document( + key=f"gitea:{full_name}@languages", + platform="gitea", + area=area, + title=f"{full_name} — languages", + claim=f"Language breakdown for {full_name}: " + ", ".join(name for name, _ in rows[:4]) + ".", + body=body, + digest=digest, + origin_url=repo_url(base_url, full_name), + path="", + repo_path="", + source_id=f"{full_name}@languages", + source_type="metric", + last_modified=updated, + confidence=confidence, + period=_period(created, updated), + tags=["gitea", "languages", area], + tech=[name for name, _ in rows[:6]], + public=guide.public, + kind="markdown", + ) + + +def _summary_document(guide, repo, area, updated, created, confidence, description, topics, base_url) -> object: + full_name = repo["full_name"] + body = "\n".join( + [f"Repository: {full_name}", f"Description: {description or '(none)'}", + f"Topics: {', '.join(topics) if topics else '(none)'}"] + ) + digest = hashlib.sha256(body.encode("utf-8")).hexdigest() + return _make_document( + key=f"gitea:{full_name}@summary", + platform="gitea", + area=area, + title=full_name, + claim=description or f"Repository {full_name}.", + body=body, + digest=digest, + origin_url=repo_url(base_url, full_name), + path="", + repo_path="", + source_id=f"{full_name}@summary", + source_type="doc", + last_modified=updated, + confidence=confidence, + period=_period(created, updated), + tags=topics + ["gitea", area], + tech=[], + public=guide.public, + kind="markdown", + ) + + +# -- obsidian vault ------------------------------------------------------- + + +def _harvest_obsidian(result, client, guide, repos, vault_repo, tree, cfg, base_url, log) -> None: + if not vault_repo or tree is None: + result.warnings.append("obsidian: the vault repository could not be listed") + result.sources["obsidian"] = _source_row("obsidian", enabled=True, documents=[], extra={"missing": True}) + return + repo = next((row for row in repos if row.get("full_name") == vault_repo), {}) + branch = repo.get("default_branch") or "main" + updated = _timestamp(repo.get("updated_at")) + created = _timestamp(repo.get("created_at")) + validator = loader.load_validator(cfg.root) + documents = [] + + blobs = [entry for entry in tree if entry.get("type") == "blob"] + for entry in blobs: + if len(documents) >= guide.limit("max_documents"): + result.warnings.append( + f"obsidian: document limit {guide.limit('max_documents')} reached" + ) + break + path = globs.normalise(entry.get("path") or "") + kind, reason = exclusions.classify(path, guide, size=entry.get("size")) + if kind == "reject": + # A path outside the include list is expected and is not a finding; + # only hard exclusions are reported, so the report stays readable. + if reason != "not under an included path": + result.add_rejection(exclusions.Rejection(path, "obsidian", reason)) + continue + if kind == "image": + result.images.append( + {"path": path, "platform": "obsidian", + "origin_url": file_url(base_url, vault_repo, branch, path)} + ) + continue + try: + raw = client.raw(vault_repo, branch, path) + except GiteaError as exc: + result.warnings.append(f"obsidian/{path}: {exc}") + continue + if exclusions.looks_binary(raw): + result.add_rejection(exclusions.Rejection(path, "obsidian", "binary content")) + continue + text = raw.decode("utf-8", "replace") + hits = exclusions.scan_text(text) + if hits: + result.add_rejection( + exclusions.Rejection( + path, "obsidian", "credential-shaped content", + ", ".join(sorted({hit[0] for hit in hits})), + ) + ) + continue + + fields, body = _split_note(text, validator) + title = str(fields.get("title") or _heading(body) or os.path.splitext(os.path.basename(path))[0]) + tag_values = fields.get("tags") or [] + if isinstance(tag_values, str): + tag_values = [tag_values] + tags = [str(tag) for tag in tag_values] + _folder_tags(path) + area = areas.area_for_note(path, guide.fits) + documents.append( + _make_document( + key=f"obsidian:{path}", + platform="obsidian", + area=area, + title=title, + claim=_claim(body, title), + body=body, + digest=hashlib.sha256(raw).hexdigest(), + origin_url=file_url(base_url, vault_repo, branch, path), + path=path, + repo_path=path, + source_id=path, + source_type="doc", + last_modified=updated, + confidence="historic", + period=_period(created, updated), + tags=tags + ["obsidian", area], + tech=[], + public=guide.public, + kind="markdown", + ) + ) + + result.documents.extend(documents) + result.sources["obsidian"] = _source_row( + "obsidian", enabled=True, documents=documents, + extra={"repo": vault_repo, "base": base_url}, + ) + log(f"obsidian: {len(documents)} note(s)") + + +# -- graphify ------------------------------------------------------------- + + +def _harvest_graph(result, client, guide, repos, vault_repo, tree, cfg, base_url, log) -> None: + if not vault_repo: + result.warnings.append("graphify: no vault repository configured in the guide") + result.sources["graphify"] = _source_row("graphify", enabled=True, documents=[], extra={"missing": True}) + return + repo = next((row for row in repos if row.get("full_name") == vault_repo), {}) + branch = repo.get("default_branch") or "main" + updated = _timestamp(repo.get("updated_at")) + vault_paths = { + globs.normalise(entry.get("path") or "") + for entry in (tree or []) + if entry.get("type") == "blob" + } + + raw = None + source_path = "" + if cfg.graph_path: + if not os.path.exists(cfg.graph_path): + raise EstateError(f"RESUME_GRAPH_PATH is set but {cfg.graph_path} does not exist") + with open(cfg.graph_path, "rb") as handle: + raw = handle.read() + source_path = cfg.graph_path + else: + for candidate in guide.limits.get("graph_paths") or ["graph.json"]: + try: + raw = client.raw(vault_repo, branch, candidate) + source_path = f"gitea:{vault_repo}/{candidate}" + break + except GiteaNotFound: + continue + except GiteaError as exc: + raise EstateError(f"cannot read graph.json from {vault_repo}: {exc}") from exc + + if raw is None: + result.warnings.append( + f"graphify: graph.json is not committed to {vault_repo} " + f"(looked in {guide.limits.get('graph_paths')}). The graph was not ingested; " + "the nightly audit will keep reporting this until it is committed." + ) + result.sources["graphify"] = _source_row( + "graphify", enabled=True, documents=[], extra={"missing": True, "repo": vault_repo} + ) + return + + graph = graph_module.parse( + raw, + platform="graphify", + base=base_url, + repo=vault_repo, + ref=branch, + last_modified=updated, + area=guide.primary_area, + vault_paths=vault_paths, + max_nodes=guide.limit("max_nodes"), + max_edges=guide.limit("max_edges"), + source_path=source_path, + ) + result.graph = graph + for rejection in graph.rejections: + result.add_rejection(rejection) + result.images.extend(graph.images) + result.sources["graphify"] = _source_row( + "graphify", enabled=True, documents=[], + extra={ + "repo": vault_repo, + "path": source_path, + "nodes": len(graph.nodes), + "edges": len(graph.edges), + "digest": graph.digest, + "built_at_commit": graph.built_at_commit, + }, + ) + if graph.warnings: + result.warnings.extend(f"graphify: {warning}" for warning in graph.warnings) + log( + f"graphify: {len(graph.nodes)} node(s), {len(graph.edges)} edge(s), " + f"{len(graph.images)} image node(s)" + ) + + +# -- shared helpers ------------------------------------------------------- + + +def _make_document(**values) -> loader.Document: + """One Document constructor, so every source produces the same shape.""" + defaults = dict( + metrics=[], artefacts=[], tech=[], tags=[], public=True, kind="markdown", + platform="gitea", tier=2, level=3, period="", source_type="doc", + ) + defaults.update(values) + title = defaults["title"] + origin = defaults["origin_url"] + path = defaults.get("path") or "" + return loader.Document( + path=defaults["key"], + source_id=defaults.get("source_id") or (path or defaults["key"]), + repo_path=defaults.get("repo_path") or path, + area=defaults["area"], + level=defaults.get("level", 3), + title=title, + claim=defaults.get("claim") or title, + confidence=defaults.get("confidence") or "historic", + period=defaults.get("period") or "—", + tech=defaults.get("tech") or [], + metrics=defaults.get("metrics") or [], + artefacts=defaults.get("artefacts") or [], + public=bool(defaults.get("public", True)), + last_verified=str(defaults["last_modified"])[:10], + tags=defaults.get("tags") or [], + body=defaults["body"], + digest=defaults["digest"], + url=origin, + platform=defaults["platform"], + tier=int(defaults.get("tier", 2)), + origin_url=origin, + last_modified=defaults["last_modified"], + source_type=defaults.get("source_type") or "doc", + kind=defaults.get("kind") or "markdown", + ) + + +def _split_note(text: str, validator) -> tuple: + """Tolerant front-matter split. A note without front-matter is still a note.""" + lines = text.splitlines() + if not lines or lines[0].strip() != "---": + return {}, text + end = None + for index in range(1, len(lines)): + if lines[index].strip() in ("---", "..."): + end = index + break + if end is None: + return {}, text + fields = {} + try: + node = validator.parse_front_matter( + [(number, line) for number, line in enumerate(lines[1:end], start=2)] + ) + parsed = loader.to_python(node) + if isinstance(parsed, dict): + fields = parsed + except validator.YamlError: + fields = {} + return fields, "\n".join(lines[end + 1 :]).strip() + + +def _source_row(name: str, enabled: bool, documents: list, extra: dict = None) -> dict: + stamps = sorted(doc.last_modified for doc in documents if doc.last_modified) + row = { + "platform": name, + "enabled": enabled, + "documents": len(documents), + "newest": stamps[-1] if stamps else "", + "oldest": stamps[0] if stamps else "", + "fetched_at": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + } + row.update(extra or {}) + return row + + +def _vault_repo(selected: dict) -> str: + for name in ("obsidian", "graphify"): + guide = selected.get(name) + if guide and guide.base and "/" in guide.base and not guide.base.startswith("http"): + return guide.base + return "" + + +def _as_list(value) -> list: + if value in (None, ""): + return [] + if isinstance(value, list): + return [str(item) for item in value] + return [str(value)] + + +def _timestamp(value) -> str: + """Normalise any RFC3339 timestamp to UTC `…Z`. Empty stays empty.""" + if not value: + return "1970-01-01T00:00:00Z" + text = str(value).strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + moment = datetime.datetime.fromisoformat(text) + except ValueError: + return "1970-01-01T00:00:00Z" + if moment.tzinfo is None: + moment = moment.replace(tzinfo=datetime.timezone.utc) + return moment.astimezone(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _period(created: str, updated: str) -> str: + start = (created or "")[:4] + end = (updated or "")[:4] + if not start or start == "1970": + return end or "—" + if start == end: + return start + return f"{start}–{end}" + + +def _claim(text: str, title: str) -> str: + for line in (text or "").splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith(("#", "|", "```", "---", ">")): + continue + cleaned = re.sub(r"[*_`\[\]]", "", stripped) + if len(cleaned) > 20: + return cleaned[:180] + return title[:180] + + +def _heading(text: str) -> str: + for line in (text or "").splitlines(): + if line.startswith("# "): + return line[2:].strip() + return "" + + +def _folder_tags(path: str) -> list: + parts = [part for part in path.split("/")[:-1] if part and not part[0].isdigit()] + return [part.lower() for part in parts[-2:]] + + +def _is_readme(path: str) -> bool: + return os.path.basename(path).lower().startswith("readme") + + +def _depth(path: str) -> int: + return globs.normalise(path).count("/") + + +def _source_rank(path: str) -> bool: + ext = exclusions.extension(path) + if ext not in exclusions.TEXT_EXTENSIONS: + return False + if ext in ("json", "csv", "tsv", "svg"): + return False + # Prefer files that carry decisions over generated or vendored files. + return not any( + part.lower() in ("vendor", "third_party", "examples", "testdata", "fixtures", "migrations") + for part in globs.normalise(path).split("/")[:-1] + ) diff --git a/corpus/sources/gitea.py b/corpus/sources/gitea.py new file mode 100644 index 0000000..ff97534 --- /dev/null +++ b/corpus/sources/gitea.py @@ -0,0 +1,165 @@ +"""Gitea API client — the only network dependency in the pipeline. + +`docs/phase-4-corpus.md` §1: Gitea is reached over HTTPS, and the vault and the +graph arrive through the same API, so there is no cross-machine copy and no SSH +from the runtime host. This client is standard library only (`urllib`), it never +prints or stores the token, and every call names only the path it requested. + +Read-only by construction: there is no method here that writes, creates, deletes +or changes anything in Gitea. The token only needs read scope. +""" + +from __future__ import annotations + +import http.client +import json +import time +import urllib.error +import urllib.parse +import urllib.request + + +class GiteaError(RuntimeError): + """A Gitea call failed. The message never contains the token.""" + + +class GiteaNotFound(GiteaError): + """The requested repository, branch or file does not exist.""" + + +class GiteaClient: + def __init__(self, base: str, token: str, timeout: float = 30.0, attempts: int = 3) -> None: + if not base: + raise GiteaError("a Gitea base URL is required (GITEA_URL or the guide's base)") + if not token: + raise GiteaError( + "a Gitea token is required. It is read from the environment variable " + "named by the platform guide (GITEA_TOKEN) and never from this repository." + ) + self.base = base.rstrip("/") + self._token = token + self.timeout = timeout + self.attempts = max(1, attempts) + self.calls = 0 + + # -- transport -------------------------------------------------------- + + def _request(self, path: str, expect_bytes: bool = False): + """One GET with retry/backoff. The only place that touches the network.""" + url = f"{self.base}/api/v1{path}" + last_error = None + for attempt in range(self.attempts): + request = urllib.request.Request( + url, + method="GET", + headers={ + "Authorization": f"token {self._token}", + "Accept": "application/vnd.gitea+json, application/json", + "User-Agent": "ai-resume-corpus/1", + }, + ) + try: + self.calls += 1 + with urllib.request.urlopen(request, timeout=self.timeout) as response: + raw = response.read() + if expect_bytes: + return raw + return json.loads(raw.decode("utf-8")) if raw.strip() else {} + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", "replace")[:200] + exc.close() + if exc.code == 404: + raise GiteaNotFound(f"GET {path}: not found") from None + if exc.code == 401: + raise GiteaError( + f"GET {path}: authentication failed. Check the token named by the " + "platform guide; it is not printed here." + ) from None + if exc.code == 403: + raise GiteaError(f"GET {path}: forbidden (the token may lack read scope)") from None + if exc.code != 429 and exc.code < 500: + raise GiteaError(f"GET {path}: HTTP {exc.code} {detail}") from None + last_error = f"HTTP {exc.code}" + except (urllib.error.URLError, TimeoutError, http.client.HTTPException, ConnectionError) as exc: + last_error = f"cannot reach {self.base}: {getattr(exc, 'reason', exc)}" + if attempt + 1 < self.attempts: + time.sleep(0.5 * (2 ** attempt)) + raise GiteaError(f"GET {path} failed after {self.attempts} attempts: {last_error}") + + # -- repositories ----------------------------------------------------- + + def repos(self, limit: int = 50) -> list: + """Every repository the token can read, sorted by full name.""" + out = [] + page = 1 + while True: + response = self._request(f"/repos/search?limit={limit}&page={page}") + rows = response.get("data") or [] + out.extend(rows) + if len(rows) < limit: + break + page += 1 + if page > 200: # a guard, not a limit anyone should reach + break + return sorted(out, key=lambda repo: repo.get("full_name", "")) + + def repo(self, full_name: str) -> dict: + return self._request(f"/repos/{_repo_path(full_name)}") + + # -- contents --------------------------------------------------------- + + def tree(self, full_name: str, ref: str) -> list: + """Recursive blob/tree listing. Returns every entry, following pages.""" + entries = [] + page = 1 + while True: + response = self._request( + f"/repos/{_repo_path(full_name)}/git/trees/{urllib.parse.quote(ref, safe='')}" + f"?recursive=true&per_page=1000&page={page}" + ) + rows = response.get("tree") or [] + entries.extend(rows) + if not rows or not response.get("truncated") or len(rows) < 1000: + break + page += 1 + if page > 50: + break + return entries + + def raw(self, full_name: str, ref: str, path: str) -> bytes: + quoted = urllib.parse.quote(path, safe="/") + return self._request( + f"/repos/{_repo_path(full_name)}/raw/{quoted}?ref={urllib.parse.quote(ref, safe='')}", + expect_bytes=True, + ) + + def commits(self, full_name: str, ref: str, limit: int = 20) -> list: + return self._request( + f"/repos/{_repo_path(full_name)}/commits?sha={urllib.parse.quote(ref, safe='')}" + f"&limit={int(limit)}&page=1" + ) + + def languages(self, full_name: str) -> dict: + return self._request(f"/repos/{_repo_path(full_name)}/languages") + + # -- presentation ----------------------------------------------------- + + def describe(self) -> dict: + return {"base": self.base, "token": "set", "calls": self.calls} + + +def file_url(base: str, full_name: str, ref: str, path: str) -> str: + """The URL a reader can open: the file at its branch, in Gitea.""" + return ( + f"{base.rstrip('/')}/{full_name}/src/branch/{urllib.parse.quote(ref, safe='')}/" + f"{urllib.parse.quote(path, safe='/')}" + ) + + +def repo_url(base: str, full_name: str) -> str: + return f"{base.rstrip('/')}/{full_name}" + + +def _repo_path(full_name: str) -> str: + owner, _, name = full_name.partition("/") + return f"{urllib.parse.quote(owner, safe='')}/{urllib.parse.quote(name, safe='')}" diff --git a/corpus/sources/graph.py b/corpus/sources/graph.py new file mode 100644 index 0000000..35bd320 --- /dev/null +++ b/corpus/sources/graph.py @@ -0,0 +1,335 @@ +"""Graphify output: nodes and edges, each joined back to the note it came from. + +`docs/phase-4-corpus.md` §4: *"Graph is not chunked. Nodes and edges go into +their own collection, joined to note chunks by `path`, so a graph answer can name +the note it came from."* + +The graph is not rebuilt here. It already exists and is rebuilt on the build host +by its own skill; this module consumes `graph.json`. + +Two rules from §3 drive the whole file: + +* **A node with no resolvable source note is refused, not defaulted.** Graphify + emits nodes for files outside the vault (`settings.json`, `.pi/…`). Those are + not traceable to a note, so they are rejected and counted. +* **Media never becomes text.** A media node becomes an image reference: path, + origin URL and note, and nothing else. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import uuid + +from .. import areas, exclusions, globs +from ..sources.gitea import file_url + + +class GraphError(RuntimeError): + """graph.json is not a graph this pipeline can admit.""" + + +class GraphResult: + def __init__(self, platform: str = "graphify"): + self.platform = platform + self.nodes: list = [] + self.edges: list = [] + self.images: list = [] + self.rejections: list = [] + self.warnings: list = [] + self.digest = "" + self.source_path = "" + self.built_at_commit = "" + + def counts(self) -> dict: + return { + "nodes": len(self.nodes), + "edges": len(self.edges), + "images": len(self.images), + "rejected": len(self.rejections), + } + + +def parse( + raw: bytes, + *, + platform: str, + base: str, + repo: str, + ref: str, + last_modified: str, + area: str, + vault_paths: set, + max_nodes: int = 20000, + max_edges: int = 40000, + source_path: str = "", +) -> GraphResult: + """Turn graph.json bytes into rows that carry their note. Never guesses.""" + result = GraphResult(platform) + result.digest = hashlib.sha256(raw).hexdigest() + result.source_path = source_path + try: + data = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise GraphError(f"graph.json is not valid UTF-8 JSON: {exc}") from exc + if not isinstance(data, dict): + raise GraphError("graph.json must be a JSON object") + result.built_at_commit = str(data.get("built_at_commit") or "") + + raw_nodes = data.get("nodes") or [] + raw_links = data.get("links") or data.get("edges") or [] + if not isinstance(raw_nodes, list) or not isinstance(raw_links, list): + raise GraphError("graph.json must contain `nodes` and `links` arrays") + + labels = {} + notes = {} + for item in raw_nodes: + if not isinstance(item, dict): + continue + node_id = str(item.get("id") or "").strip() + if not node_id: + continue + labels[node_id] = str(item.get("label") or node_id) + notes[node_id] = _note_for(item, vault_paths) + + for item in raw_nodes: + if len(result.nodes) >= max_nodes: + result.warnings.append(f"graph node limit {max_nodes} reached; remaining nodes skipped") + break + if not isinstance(item, dict): + continue + node_id = str(item.get("id") or "").strip() + if not node_id: + result.rejections.append( + _rejection("graph.json", platform, "graph node has no id") + ) + continue + source_file = globs.normalise(item.get("source_file") or "") + note = notes.get(node_id) + if not note: + result.rejections.append( + _rejection( + source_file or f"node:{node_id}", platform, + "no resolvable source note", "graph node does not point at a file in the vault", + ) + ) + continue + if globs.normalise(source_file) != note: + # A node can be derived from one note while naming another file; the + # note is the traceable origin either way. + pass + if exclusions.classify(note, None)[0] == "image" or exclusions.extension(note) in exclusions.MEDIA_EXTENSIONS: + result.images.append( + { + "path": note, + "label": str(item.get("label") or node_id), + "origin_url": file_url(base, repo, ref, note), + "platform": platform, + "note": note, + } + ) + continue + + node_area = areas.area_for(note, (area,), fallback=area) + label = str(item.get("label") or node_id) + content_hash = _hash( + {"kind": "node", "id": node_id, "label": label, "note": note, + "community": item.get("community"), "type": item.get("node_type")} + ) + text = ( + f"{label}. Knowledge graph node" + + (f" in community {item.get('community')}" if item.get("community") is not None else "") + + f", from the note {note}." + ) + result.nodes.append( + { + "id": node_id, + "point_id": _point_id("node", node_id, content_hash), + "graph_id": node_id, + "label": label, + "norm_label": str(item.get("norm_label") or _norm(label)), + "node_type": str(item.get("node_type") or item.get("file_type") or "entity"), + "community": item.get("community"), + "source_note": note, + "origin_url": file_url(base, repo, ref, note), + "area": node_area, + "tier": 2, + "platform": platform, + "last_modified": last_modified, + "content_hash": content_hash, + "text": text, + } + ) + + admitted_nodes = {row["graph_id"]: row for row in result.nodes} + for item in raw_links: + if len(result.edges) >= max_edges: + result.warnings.append(f"graph edge limit {max_edges} reached; remaining edges skipped") + break + if not isinstance(item, dict): + continue + source = str(item.get("source") or "").strip() + target = str(item.get("target") or "").strip() + if not source or not target: + result.rejections.append(_rejection("graph.json", platform, "graph edge has no endpoints")) + continue + note = notes.get(source) or _note_for(item, vault_paths) + if not note: + result.rejections.append( + _rejection( + f"edge:{source}->{target}", platform, + "no resolvable source note", "graph edge does not point at a file in the vault", + ) + ) + continue + source_label = labels.get(source, source) + target_label = labels.get(target, target) + relation = str(item.get("relation") or item.get("rel") or "related_to") + area_name = areas.area_for(note, (area,), fallback=area) + content_hash = _hash( + {"kind": "edge", "source": source, "target": target, + "relation": relation, "note": note} + ) + text = f"{source_label} {relation} {target_label}, from the note {note}." + result.edges.append( + { + "id": _edge_id(source, relation, target, content_hash), + "point_id": _point_id("edge", f"{source}>{relation}>{target}", content_hash), + "source_id": source, + "target_id": target, + "relation": relation, + "confidence": str(item.get("confidence") or "INFERRED"), + "confidence_score": item.get("confidence_score"), + "source_note": note, + "origin_url": file_url(base, repo, ref, note), + "area": area_name, + "tier": 2, + "platform": platform, + "last_modified": last_modified, + "content_hash": content_hash, + "text": text, + } + ) + + # Edges whose endpoints were not admitted are still traceable and are kept; + # an answer can name the relation and its note even when the node itself had + # no origin. `admitted_node_ids` is exposed for verify, which reports the gap. + result.admitted_node_ids = set(admitted_nodes) + result.orphan_edges = [ + edge for edge in result.edges + if edge["source_id"] not in admitted_nodes or edge["target_id"] not in admitted_nodes + ] + return result + + +def point_payload(row: dict, kind: str) -> dict: + """One Qdrant point payload for a node or edge. Same provenance as the row.""" + payload = { + "platform": row["platform"], + "tier": row["tier"], + "source_type": "graph_node" if kind == "node" else "graph_edge", + "area": row["area"], + "origin_url": row["origin_url"], + "path": row["source_note"], + "source_note": row["source_note"], + "title": row.get("label") or row.get("relation") or row["source_note"], + "confidence": row.get("confidence") or "INFERRED", + "last_modified": row["last_modified"], + "content_hash": row["content_hash"], + "text": row["text"], + "kind": kind, + "graph_id": row.get("graph_id") or row.get("id"), + } + if kind == "node": + payload["community"] = row.get("community") + payload["node_type"] = row.get("node_type") + else: + payload["relation"] = row.get("relation") + payload["source_id"] = row.get("source_id") + payload["target_id"] = row.get("target_id") + payload["confidence_score"] = row.get("confidence_score") + return payload + + +def query(cfg, store, text: str, area: str = None, limit: int = 10) -> dict: + """Graph query: keyword hits from SQLite, vector hits from Qdrant when up. + + Returning both is deliberate: the keyword side always works, and the vector + side proves the graph collection is live. Each row carries `source_note`. + """ + from ..qdrant import Qdrant, QdrantError + + report = {"query": text, "area": area, "keyword": [], "vector": [], "ok": True} + report["keyword"] = store.graph_search(text, area=area, limit=limit) + if not report["keyword"]: + report["ok"] = False + report["reason"] = "no graph rows matched the query" + qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout) + if not qdrant.health() or not qdrant.collection_exists(cfg.graph_collection): + report["vector"] = [] + report["vector_note"] = f"graph collection {cfg.graph_collection} is not available" + return report + from .. import embeddings + + provider = embeddings.build_provider(cfg) + try: + vector = provider.embed([text])[0] + hits = qdrant.search(cfg.graph_collection, vector, limit=limit, area=area) + except (QdrantError, Exception) as exc: # noqa: BLE001 - a graph query must not crash a shell + report["vector_note"] = f"graph vector search unavailable: {exc}" + return report + report["vector"] = [ + { + "score": round(hit.get("score", 0.0), 4), + "kind": (hit.get("payload") or {}).get("kind"), + "label": (hit.get("payload") or {}).get("title"), + "source_note": (hit.get("payload") or {}).get("source_note"), + "origin_url": (hit.get("payload") or {}).get("origin_url"), + } + for hit in hits + ] + return report + + +# -- helpers ------------------------------------------------------------- + + +def _note_for(item: dict, vault_paths: set) -> str: + source = globs.normalise(item.get("source_file") or "") + if not source: + return "" + if source in vault_paths: + return source + # Graphify sometimes prefixes `./` or uses an absolute scan root. A suffix + # match is only accepted when it is unique, so a note is never guessed. + candidates = [path for path in vault_paths if path.endswith("/" + source)] + if len(candidates) == 1: + return candidates[0] + return "" + + +def _rejection(path: str, platform: str, reason: str, detail: str = ""): + rejection = exclusions.Rejection(path, platform, reason, detail) + return rejection.to_dict() + + +def _hash(value) -> str: + body = json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(body).hexdigest() + + +def _point_id(kind: str, key: str, content_hash: str) -> str: + digest = hashlib.sha256(f"graph\x00{kind}\x00{key}\x00{content_hash}".encode("utf-8")).hexdigest() + return str(uuid.UUID(hex=digest[:32])) + + +def _edge_id(source: str, relation: str, target: str, content_hash: str) -> str: + return hashlib.sha256( + f"edge\x00{source}\x00{relation}\x00{target}\x00{content_hash}".encode("utf-8") + ).hexdigest()[:32] + + +def _norm(label: str) -> str: + return re.sub(r"[^0-9a-z]+", "_", (label or "").lower()).strip("_") diff --git a/corpus/store.py b/corpus/store.py index c163792..84df118 100644 --- a/corpus/store.py +++ b/corpus/store.py @@ -27,7 +27,7 @@ import sqlite3 import time import unicodedata -SCHEMA_VERSION = "1" +SCHEMA_VERSION = "2" SCHEMA = """ CREATE TABLE IF NOT EXISTS chunks ( @@ -49,6 +49,11 @@ CREATE TABLE IF NOT EXISTS chunks ( embedding_dim INTEGER NOT NULL, chunker_version TEXT NOT NULL, chunk_tokens INTEGER NOT NULL, + platform TEXT NOT NULL, + tier INTEGER NOT NULL, + origin_url TEXT NOT NULL, + path TEXT NOT NULL, + last_modified TEXT NOT NULL, public INTEGER NOT NULL, artefacts TEXT NOT NULL, tech TEXT NOT NULL, @@ -66,6 +71,8 @@ CREATE VIRTUAL TABLE IF NOT EXISTS chunk_fts USING fts5( text, area UNINDEXED, confidence UNINDEXED, + platform UNINDEXED, + tier UNINDEXED, tokenize = 'porter unicode61 remove_diacritics 2' ); @@ -84,6 +91,55 @@ CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); + +-- The knowledge graph is not chunked (docs/phase-4-corpus.md §4). Nodes and +-- edges live in their own tables and their own Qdrant collection, joined to note +-- chunks by source_note. Every row carries the note it came from. +CREATE TABLE IF NOT EXISTS graph_nodes ( + id TEXT PRIMARY KEY, + label TEXT NOT NULL, + norm_label TEXT NOT NULL, + node_type TEXT NOT NULL, + community INTEGER, + source_note TEXT NOT NULL, + origin_url TEXT NOT NULL, + area TEXT NOT NULL, + tier INTEGER NOT NULL, + platform TEXT NOT NULL, + last_modified TEXT NOT NULL, + content_hash TEXT NOT NULL, + text TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS graph_nodes_note ON graph_nodes(source_note); + +CREATE TABLE IF NOT EXISTS graph_edges ( + id TEXT PRIMARY KEY, + source_id TEXT NOT NULL, + target_id TEXT NOT NULL, + relation TEXT NOT NULL, + confidence TEXT NOT NULL, + confidence_score REAL, + source_note TEXT NOT NULL, + origin_url TEXT NOT NULL, + area TEXT NOT NULL, + tier INTEGER NOT NULL, + platform TEXT NOT NULL, + last_modified TEXT NOT NULL, + content_hash TEXT NOT NULL, + text TEXT NOT NULL +); +CREATE INDEX IF NOT EXISTS graph_edges_source ON graph_edges(source_id); +CREATE INDEX IF NOT EXISTS graph_edges_target ON graph_edges(target_id); +CREATE INDEX IF NOT EXISTS graph_edges_note ON graph_edges(source_note); + +CREATE VIRTUAL TABLE IF NOT EXISTS graph_fts USING fts5( + id UNINDEXED, + kind UNINDEXED, + label, + text, + area UNINDEXED, + tokenize = 'porter unicode61 remove_diacritics 2' +); """ # Column order used by every INSERT in this module. @@ -91,6 +147,7 @@ _CHUNK_COLUMNS = ( "id", "source_path", "source_id", "chunk_index", "area", "level", "source_type", "confidence", "period", "last_verified", "title", "claim", "url", "content_hash", "embedding_model", "embedding_dim", "chunker_version", "chunk_tokens", + "platform", "tier", "origin_url", "path", "last_modified", "public", "artefacts", "tech", "tags", "text", ) @@ -110,7 +167,13 @@ class Store: self.db.execute("PRAGMA journal_mode = WAL") self.db.execute("PRAGMA synchronous = NORMAL") self.db.executescript(SCHEMA) - self.set_meta("schema_version", SCHEMA_VERSION) + stored = self.get_meta("schema_version") + # A database from an older payload shape has no provenance columns. It is + # derived data, so the fix is a rebuild, but it must never be silently + # mixed with the new shape: `rebuild_schema` is called by a rebuild only. + self.schema_stale = stored not in (None, SCHEMA_VERSION) + if not self.schema_stale: + self.set_meta("schema_version", SCHEMA_VERSION) # -- lifecycle -------------------------------------------------------- @@ -125,10 +188,22 @@ class Store: def reset(self) -> None: """Empty every table. Used by a from-scratch rebuild.""" - for table in ("chunk_fts", "chunks", "manifest", "meta"): + for table in ("chunk_fts", "chunks", "manifest", "meta", "graph_fts", "graph_nodes", "graph_edges"): self.db.execute(f"DELETE FROM {table}") self.db.commit() + def rebuild_schema(self) -> None: + """Drop and recreate every table. Only a rebuild may call this.""" + for table in ( + "chunk_fts", "chunks", "manifest", "meta", + "graph_fts", "graph_nodes", "graph_edges", + ): + self.db.execute(f"DROP TABLE IF EXISTS {table}") + self.db.commit() + self.db.executescript(SCHEMA) + self.schema_stale = False + self.set_meta("schema_version", SCHEMA_VERSION) + # -- writing ---------------------------------------------------------- def write_document(self, doc, payloads: list) -> None: @@ -161,8 +236,9 @@ class Store: if value ) self.db.execute( - "INSERT INTO chunk_fts (rowid, id, source_path, title, terms, text, area, confidence) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + "INSERT INTO chunk_fts " + "(rowid, id, source_path, title, terms, text, area, confidence, platform, tier) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ( cursor.lastrowid, payload["id"], @@ -172,6 +248,8 @@ class Store: payload["text"], payload["area"], payload["confidence"], + payload["platform"], + payload["tier"], ), ) self.db.execute( @@ -237,7 +315,10 @@ class Store: out.append(payload) return out - def keyword_search(self, query: str, area: str = None, confidence: str = None, limit: int = 10) -> list: + def keyword_search( + self, query: str, area: str = None, confidence: str = None, + platform: str = None, tier: int = None, limit: int = 10, + ) -> list: """BM25 over chunk text, optionally filtered before ranking.""" match = match_expression(query) if not match: @@ -256,11 +337,110 @@ class Store: if confidence: sql.append("AND c.confidence = ?") args.append(confidence) + if platform: + sql.append("AND c.platform = ?") + args.append(platform) + if tier is not None: + sql.append("AND c.tier = ?") + args.append(int(tier)) sql.append("ORDER BY bm25(chunk_fts) LIMIT ?") args.append(int(limit)) rows = self.db.execute(" ".join(sql), args) return [dict(row) for row in rows] + def pool_texts(self) -> list: + """Every chunk text in the pool. The credential scan reads this.""" + return [ + (row["source_path"], row["chunk_index"], row["text"]) + for row in self.db.execute("SELECT source_path, chunk_index, text FROM chunks") + ] + + # -- graph ------------------------------------------------------------ + + def write_graph(self, nodes: list, edges: list) -> None: + """Replace the whole graph. Nodes and edges are rebuilt together.""" + self.db.execute("DELETE FROM graph_fts") + self.db.execute("DELETE FROM graph_nodes") + self.db.execute("DELETE FROM graph_edges") + for node in nodes: + self.db.execute( + "INSERT INTO graph_nodes (id, label, norm_label, node_type, community, " + "source_note, origin_url, area, tier, platform, last_modified, content_hash, text) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + node["id"], node["label"], node.get("norm_label") or _norm(node["label"]), + node.get("node_type") or "entity", node.get("community"), + node["source_note"], node["origin_url"], node["area"], node["tier"], + node["platform"], node["last_modified"], node["content_hash"], node["text"], + ), + ) + self.db.execute( + "INSERT INTO graph_fts (id, kind, label, text, area) VALUES (?, 'node', ?, ?, ?)", + (node["id"], node["label"], node["text"], node["area"]), + ) + for edge in edges: + self.db.execute( + "INSERT INTO graph_edges (id, source_id, target_id, relation, confidence, " + "confidence_score, source_note, origin_url, area, tier, platform, " + "last_modified, content_hash, text) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ( + edge["id"], edge["source_id"], edge["target_id"], edge["relation"], + edge["confidence"], edge.get("confidence_score"), edge["source_note"], + edge["origin_url"], edge["area"], edge["tier"], edge["platform"], + edge["last_modified"], edge["content_hash"], edge["text"], + ), + ) + self.db.execute( + "INSERT INTO graph_fts (id, kind, label, text, area) VALUES (?, 'edge', ?, ?, ?)", + (edge["id"], edge["relation"], edge["text"], edge["area"]), + ) + self.db.commit() + + def graph_counts(self) -> dict: + return { + "nodes": int(self.db.execute("SELECT count(*) FROM graph_nodes").fetchone()[0]), + "edges": int(self.db.execute("SELECT count(*) FROM graph_edges").fetchone()[0]), + } + + def graph_node(self, node_id: str) -> dict: + row = self.db.execute("SELECT * FROM graph_nodes WHERE id = ?", (node_id,)).fetchone() + return dict(row) if row else {} + + def graph_neighbours(self, node_id: str, limit: int = 50) -> list: + rows = self.db.execute( + "SELECT e.*, sn.label AS source_label, tn.label AS target_label " + "FROM graph_edges e " + "LEFT JOIN graph_nodes sn ON sn.id = e.source_id " + "LEFT JOIN graph_nodes tn ON tn.id = e.target_id " + "WHERE e.source_id = ? OR e.target_id = ? LIMIT ?", + (node_id, node_id, int(limit)), + ) + return [dict(row) for row in rows] + + def graph_search(self, query: str, area: str = None, limit: int = 10) -> list: + """Keyword search over node labels and edge relations, with provenance.""" + match = match_expression(query) + if not match: + return [] + sql = [ + "SELECT f.id, f.kind, f.label, f.text, f.area,", + " COALESCE(n.source_note, e.source_note) AS source_note,", + " COALESCE(n.origin_url, e.origin_url) AS origin_url,", + " n.community, e.relation, e.source_id, e.target_id", + "FROM graph_fts f", + "LEFT JOIN graph_nodes n ON f.kind = 'node' AND n.id = f.id", + "LEFT JOIN graph_edges e ON f.kind = 'edge' AND e.id = f.id", + "WHERE graph_fts MATCH ?", + ] + args = [match] + if area: + sql.append("AND f.area = ?") + args.append(area) + sql.append("ORDER BY bm25(graph_fts) LIMIT ?") + args.append(int(limit)) + return [dict(row) for row in self.db.execute(" ".join(sql), args)] + # -- meta ------------------------------------------------------------- def set_meta(self, key: str, value) -> None: @@ -280,6 +460,10 @@ class Store: return {row["key"]: json.loads(row["value"]) for row in self.db.execute("SELECT key, value FROM meta")} +def _norm(label: str) -> str: + return re.sub(r"[^0-9a-z]+", "_", (label or "").lower()).strip("_") + + def match_expression(query: str) -> str: """Free text -> a safe FTS5 expression: bare words, each a prefix, all required. diff --git a/corpus/tests/support.py b/corpus/tests/support.py index 8fb41f6..d0110cd 100644 --- a/corpus/tests/support.py +++ b/corpus/tests/support.py @@ -119,6 +119,8 @@ def make_config(root: str, **overrides) -> Config: dim=32, qdrant_url="fake://qdrant", collection="test_corpus", + graph_collection="test_graph", + offline=True, ) values.update(overrides) return Config.from_env(root=root, **values) diff --git a/platforms/gitea.yml b/platforms/gitea.yml new file mode 100644 index 0000000..b11a983 --- /dev/null +++ b/platforms/gitea.yml @@ -0,0 +1,46 @@ +# Platform guide — Gitea. +# +# This file is both configuration and public copy. The site shows it so a +# visitor can see exactly what the assistant reads. Keep the wording honest. +name: gitea +enabled: true +what: >- + Self-hosted Git. The source of truth for every current project: README files, + file trees, selected source files, recent commit messages and language + statistics. This is where the work itself lives. +fits: [ai, platform, edge, products] +base: https://gitea.lab.audasmedia.com.au +auth: GITEA_TOKEN +include: + repos: all + # No path include means: every text file the repository rules admit. + paths: [] +exclude: + repos: + # The vault is read by the obsidian guide, so its notes are not harvested + # twice under a different platform. + - sam/obsidian-vault + paths: + - "**/node_modules/**" + - "**/.venv/**" + - "**/venv/**" + - "**/.git/**" + - "**/.trash/**" + - "**/.obsidian/**" + - "**/.pi/**" + - "**/__pycache__/**" + - "**/graphify-out/**" + - "**/graphify-out-temp/**" + - "**/vendor/**" + - "**/third_party/**" + - "**/*.min.js" + - "**/*.min.css" + - "**/*.lock" +read: [readme, tree, selected_source, commits, languages] +public: true +limits: + max_documents: 4000 + max_file_kb: 200 + max_files_per_repo: 40 + max_commits: 20 + max_tree_entries: 3000 diff --git a/platforms/graphify.yml b/platforms/graphify.yml new file mode 100644 index 0000000..6544a52 --- /dev/null +++ b/platforms/graphify.yml @@ -0,0 +1,23 @@ +# Platform guide — Graphify knowledge graph. +# +# This file is both configuration and public copy. +name: graphify +enabled: true +what: >- + A knowledge graph over the notes. It answers "what connects X to Y" rather + than "what mentions X": nodes are the things in the estate, edges are the + relations between them, and every one points back at the note it came from. +fits: [ai] +base: sam/obsidian-vault +read: [graph_json] +public: true +limits: + max_documents: 4000 + # graph.json is the only file this platform reads. These are the places it may + # be committed; the first match wins. + graph_paths: + - graphify-out/graph.json + - graph.json + - Graph/graph.json + max_nodes: 20000 + max_edges: 40000 diff --git a/platforms/obsidian.yml b/platforms/obsidian.yml new file mode 100644 index 0000000..f055516 --- /dev/null +++ b/platforms/obsidian.yml @@ -0,0 +1,44 @@ +# Platform guide — Obsidian vault (sam/obsidian-vault, read through Gitea). +# +# This file is both configuration and public copy. +# +# Privacy note: only project, area and index notes are read. The personal and +# family folders (000 daily, 400 Personal Family, 500 archive) and the templates +# are deliberately out of scope. They are not work evidence and they contain +# third-party personal detail. See docs/content-redaction.md. +name: obsidian +enabled: true +what: >- + A personal knowledge base. High-level notes on every system, in Sam's own + words: what a system is for, what changed, and what went wrong. Project and + area notes only; personal, family and daily notes are not read. +fits: [about, ai, platform, edge, products] +base: sam/obsidian-vault +include: + paths: + - "010 inbox/**" + - "100 Table of Contents/**" + - "200 projects/**" + - "300 areas/**" +exclude: + paths: + - ".obsidian/**" + - ".trash/**" + - ".venv/**" + - "templates/**" + - "000 daily/**" + - "400 Personal Family/**" + - "500 archive/**" + - "**/*.excalidraw" + - "**/*.png" + - "**/*.jpg" + - "**/*.jpeg" + - "**/*.gif" + - "**/*.webp" + - "**/*.pdf" + - "**/*Bedroom Lights*" +read: [notes] +public: true +limits: + max_documents: 4000 + max_file_kb: 200