WIP Phase 4 ingestion: platform guides, exclusion gate, gitea/vault/graph harvest, payload provenance

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])

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

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

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}"
         )
 

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

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

corpus/guides.py

@@ -0,0 +1,227 @@
+"""Platform guides: one small file per platform, configuration and public copy.
+
+`platforms/<name>.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())],
+    }

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

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
+    Harvested documents carry an explicit `source_type`. Curated content does
-    classification is derived from tags the schema does allow.
+    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))

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
+* **rebuild** — drop the collections, empty the store, embed everything. This is
-  the "one command rebuilds the corpus from an empty volume" path.
+  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
+  embedding dimension or chunker version changed are re-chunked and re-embedded.
-  re-embedded. Everything else is left alone.
+  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
+The order matters. Every payload for every changed document is built and
-before a single vector is written, so an invalid payload stops the run with an
+validated *before* a single vector is written, so an invalid payload stops the
-empty diff rather than leaving half an index behind.
+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.collection, cfg)
-        _ensure_collection(qdrant, 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)
+            prepared.append((doc, _prepare(doc, cfg)))
-            stats.chunks_written += len(payloads_for_doc)
+        all_payloads = [payload for _, rows in prepared for payload in rows]
 
-            vectors = provider.embed([p["text"] for p in payloads_for_doc])
+        # 3. Embed in one flat, batched pass. This is the number the DoD uses to
-            if len(vectors) != len(payloads_for_doc):
+        #    prove incremental mode: it is proportional to changed chunks only.
-                raise PipelineError(
+        vectors = _embed_all(provider, [payload["text"] for payload in all_payloads])
-                    f"{doc.path}: provider returned {len(vectors)} vectors for "
+        if len(vectors) != len(all_payloads):
-                    f"{len(payloads_for_doc)} chunks"
+            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),
+            # Remove what this document used to have, then write the new rows and
-            # then write the new rows and points. A crash between the two leaves an
+            # points. A crash between the two leaves an index that reports as
-            # index that reports as incomplete, never one that claims to be whole.
+            # 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())})
+        store.set_meta("sources", harvest.sources)
-        _write_collection_file(cfg, store, stats)
+        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:
+def _ensure_collection(qdrant: Qdrant, name: str, cfg) -> None:
-    if not qdrant.collection_exists(cfg.collection):
+    if not qdrant.collection_exists(name):
-        qdrant.create_collection(
+        qdrant.create_collection(name, cfg.dim, vectors_on_disk=cfg.vectors_on_disk)
-            cfg.collection, 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:
+def _embed_all(provider, texts: list) -> list:
-    """Mirror the contract to data/corpus/collection.json for the operator."""
+    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}):
+        for source in sorted(set(by_source) - set(on_disk)):
-            problems.append(f"{source}: indexed but no longer in content/ (run sync)")
+            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}

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

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]
+    )

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='')}"

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("_")

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) "
+                "INSERT INTO chunk_fts "
-                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                "(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.
 

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)

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

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

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