8 Commits

Author SHA1 Message Date
Sam Rolfe
447a0523fb deploy: publish on the LAN address so Caddy on .35 can reach it
127.0.0.1:8096 is unreachable from .35, which is where Caddy runs. Every other
service on .13 is published this way (8095, 8092, 3909, 4200) and proxied by the
same Caddy. Port is LAN-only; Caddy remains the sole public listener.
2026-10-26 22:38:54 +11:00
Paperclip Builder
0b910d216e ingestion: group oversize and binary rejections in the report 2026-10-26 22:36:34 +11:00
Paperclip Builder
720feddfcc ingestion: strip repeated front-matter blocks from notes 2026-10-26 22:33:57 +11:00
Paperclip Builder
4896e18a41 ingestion: strip front-matter from harvested markdown 2026-10-26 22:32:14 +11:00
Paperclip Builder
da674b5555 ingestion: skip empty files instead of failing the run 2026-10-26 22:30:36 +11:00
Paperclip Builder
4e52015fd2 ingestion: report per-platform added/updated correctly; qdrant compose subnet 2026-10-26 22:28:57 +11:00
Paperclip Builder
e3f54f58c1 Phase 4 ingestion: exclusion/provenance/incremental/graph tests 2026-10-26 22:28:21 +11:00
Paperclip Builder
1e46ec009a WIP Phase 4 ingestion: platform guides, exclusion gate, gitea/vault/graph harvest, payload provenance 2026-10-26 22:26:45 +11:00
25 changed files with 3507 additions and 84 deletions

View File

@@ -23,7 +23,8 @@ BIN := bin
DIST := dist
.PHONY: build dist generate dev test validate validate-self-test fmt vet clean help \
corpus corpus-sync corpus-rebuild corpus-audit corpus-verify corpus-demo corpus-test
corpus corpus-sync corpus-rebuild corpus-offline corpus-audit corpus-verify \
corpus-demo corpus-test corpus-graph corpus-scan corpus-sources
help:
@echo "make build templ generate, compile bin/, no dist"
@@ -36,11 +37,15 @@ help:
@echo "make clean remove bin/ and dist/"
@echo ""
@echo "corpus (Phase 4, python3 -m corpus.cli):"
@echo "make corpus rebuild the corpus from content/ (drops the collection)"
@echo "make corpus rebuild the pool from the estate (drops both collections)"
@echo "make corpus-offline rebuild content/ only, no Gitea (recovery and CI)"
@echo "make corpus-sync incremental update; only changed documents are re-embedded"
@echo "make corpus-audit freshness + drift audit (exit 1 on findings)"
@echo "make corpus-verify check the index against content/ and Qdrant"
@echo "make corpus-demo same query with and without an area filter"
@echo "make corpus-graph Q= query graph nodes and edges (with their source note)"
@echo "make corpus-scan search the pool for credential-shaped content"
@echo "make corpus-sources show the platform guides and the last harvest"
@echo "make corpus-test corpus unit tests (no container, no network)"
build: generate
@@ -87,6 +92,9 @@ corpus: corpus-rebuild
corpus-rebuild:
@$(CORPUS) rebuild
corpus-offline:
@$(CORPUS) --offline rebuild
corpus-sync:
@$(CORPUS) sync
@@ -99,6 +107,16 @@ corpus-verify:
corpus-demo:
@$(CORPUS) demo-filter --query "distributed audio endpoints" --area edge
corpus-graph:
@test -n "$(Q)" || { echo "usage: make corpus-graph Q='photo pipeline'"; exit 2; }
@$(CORPUS) graph --query "$(Q)"
corpus-scan:
@$(CORPUS) scan
corpus-sources:
@$(CORPUS) sources
corpus-test:
@$(PYTHON) -m unittest discover -s corpus/tests -t .

94
corpus/areas.py Normal file
View File

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

View File

@@ -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

View File

@@ -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,10 @@ def main(argv=None) -> int:
"hybrid": _cmd_hybrid,
"measure": _cmd_measure,
"info": _cmd_info,
"graph": _cmd_graph,
"graph-neighbours": _cmd_graph_neighbours,
"scan": _cmd_scan,
"sources": _cmd_sources,
}[args.command]
return handler(cfg, args, log)
except (pipeline.PipelineError, ConfigError) as exc:
@@ -92,6 +102,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 +141,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 +265,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 +360,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

View File

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

256
corpus/exclusions.py Normal file
View File

@@ -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

View File

@@ -30,15 +30,74 @@ from . import loader, store as store_module
from .config import CHUNKER_VERSION
from .pipeline import _contract
from .qdrant import Qdrant
from .sources import estate as estate_module
def audit(cfg, today: datetime.date = None, log=None) -> dict:
"""Return the audit report as a plain dict (JSON-serialisable)."""
log = log or (lambda _msg: None)
today = today or datetime.date.today()
documents = loader.load_documents(cfg)
findings = {"stale": [], "due_soon": [], "index_drift": [], "contract_drift": []}
# The audit reads the same estate the pipeline reads, so "the source has
# stopped changing" is measured against the source, not against the index.
harvest = None
try:
harvest = estate_module.harvest(cfg, log=log)
documents = harvest.documents
except estate_module.EstateError as exc:
documents = loader.load_documents(cfg)
estate_error = str(exc)
else:
estate_error = ""
findings = {
"stale": [],
"due_soon": [],
"index_drift": [],
"contract_drift": [],
"stale_source": [],
"source_missing": [],
"source_unavailable": [],
}
if estate_error:
findings["source_unavailable"].append({"platform": "estate", "problem": estate_error})
if harvest is not None:
for name, row in sorted((harvest.sources or {}).items()):
if not row.get("enabled"):
continue
# Curated content is measured by last_verified above; source silence
# is about the harvested estate, which should change on its own.
if name == "curated":
continue
if row.get("missing") or not row.get("documents"):
findings["source_missing"].append(
{
"platform": name,
"problem": "no documents were harvested",
"repo": row.get("repo") or row.get("base") or "",
}
)
continue
newest = row.get("newest") or ""
try:
changed = datetime.date.fromisoformat(newest[:10])
except ValueError:
findings["source_missing"].append(
{"platform": name, "problem": f"no usable last_modified ({newest!r})"}
)
continue
age = (today - changed).days
if age > cfg.source_silence_days:
findings["stale_source"].append(
{
"platform": name,
"repo": row.get("repo") or row.get("base") or "",
"newest": newest,
"age_days": age,
"problem": f"the source has not changed in {age} days",
}
)
for doc in documents:
try:
@@ -90,6 +149,7 @@ def audit(cfg, today: datetime.date = None, log=None) -> dict:
{"source_path": source, "problem": "indexed but no longer in content/"}
)
graph_counts = store.graph_counts()
stored_contract = store.get_meta("collection")
if stored_contract != _contract(cfg):
findings["contract_drift"].append(
@@ -120,23 +180,36 @@ def audit(cfg, today: datetime.date = None, log=None) -> dict:
"sources": store.source_count(),
"collection": cfg.collection,
"collection_points": points,
"graph_collection": cfg.graph_collection,
"graph_nodes": graph_counts["nodes"],
"graph_edges": graph_counts["edges"],
"provider": cfg.provider,
"model": cfg.model,
"dim": cfg.dim,
"chunker_version": CHUNKER_VERSION,
},
"findings": findings,
"sources": (harvest.sources if harvest is not None else {}),
"warnings": (harvest.warnings if harvest is not None else []),
"summary": {
"stale": len(findings["stale"]),
"due_soon": len(findings["due_soon"]),
"index_drift": len(findings["index_drift"]),
"contract_drift": len(findings["contract_drift"]),
"stale_source": len(findings["stale_source"]),
"source_missing": len(findings["source_missing"]),
"source_unavailable": len(findings["source_unavailable"]),
},
# `due_soon` is a warning, not a failure: the document is still
# inside its verification horizon. Stale, drift and contract
# problems are failures.
"ok": not (
findings["stale"] or findings["index_drift"] or findings["contract_drift"]
findings["stale"]
or findings["index_drift"]
or findings["contract_drift"]
or findings["stale_source"]
or findings["source_missing"]
or findings["source_unavailable"]
),
}
log(f"audit: {json.dumps(report['summary'], sort_keys=True)}")
@@ -147,9 +220,12 @@ def render_text(report: dict) -> str:
"""Human-readable audit, for a terminal or an Apprise body."""
lines = []
corpus = report["corpus"]
graph = ""
if corpus.get("graph_nodes") is not None:
graph = f", graph {corpus['graph_nodes']} node(s) / {corpus['graph_edges']} edge(s)"
lines.append(
f"corpus: {corpus['documents']} document(s), {corpus['chunks']} chunk(s), "
f"{corpus['collection_points']} vector(s) in {corpus['collection']}"
f"{corpus['collection_points']} vector(s) in {corpus['collection']}{graph}"
)
lines.append(
f"horizon: last_verified within {report['horizon']['max_age_days']} days "
@@ -159,10 +235,13 @@ def render_text(report: dict) -> str:
for kind, label in (
("stale", "STALE — past the verification horizon"),
("due_soon", "DUE SOON — verify before it goes stale"),
("index_drift", "INDEX DRIFT — the index does not match content/"),
("index_drift", "INDEX DRIFT — the index does not match the sources"),
("contract_drift", "CONTRACT DRIFT — rebuild required"),
("stale_source", "STALE SOURCE — the estate has stopped changing"),
("source_missing", "SOURCE MISSING — an enabled platform produced nothing"),
("source_unavailable", "SOURCE UNAVAILABLE — the estate could not be read"),
):
rows = findings[kind]
rows = findings.get(kind) or []
if not rows:
continue
lines.append("")
@@ -177,7 +256,10 @@ def render_text(report: dict) -> str:
)
if report["ok"]:
lines.append("")
lines.append("OK: nothing stale, no drift." + (" (see warnings above)" if findings["due_soon"] else ""))
lines.append(
"OK: nothing stale, no drift, every source current."
+ (" (see warnings above)" if findings["due_soon"] else "")
)
return "\n".join(lines)

89
corpus/globs.py Normal file
View File

@@ -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

227
corpus/guides.py Normal file
View File

@@ -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())],
}

View File

@@ -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

View File

@@ -16,8 +16,14 @@ import datetime
AREAS = ("ai", "platform", "edge", "products", "history", "about")
LEVELS = (1, 2, 3)
SOURCE_TYPES = ("doc", "code", "adr", "metric")
# docs/phase-4-corpus.md §3 fixes the harvested types; adr and metric were
# already used by curated content and stay valid.
SOURCE_TYPES = (
"doc", "code", "commit", "graph_node", "graph_edge", "claim", "adr", "metric",
)
CONFIDENCE = ("delivered", "running", "in-progress", "abandoned", "historic")
PLATFORMS = ("curated", "gitea", "obsidian", "graphify")
TIERS = (1, 2)
# Fields every payload must carry (docs/phase-4-corpus.md §4.2), plus the ones
# the index needs to answer "where did this come from?" without a second lookup.
@@ -30,6 +36,13 @@ REQUIRED_FIELDS = (
"artefacts",
"last_verified",
"text",
# docs/phase-4-corpus.md §3: provenance is part of the payload, not an
# optional extra. A chunk that cannot say where it came from is not admitted.
"platform",
"tier",
"origin_url",
"path",
"last_modified",
)
PROVENANCE_FIELDS = (
"id",
@@ -51,11 +64,15 @@ class PayloadError(RuntimeError):
def source_type_for(doc) -> str:
"""Classify a document. Everything in content/ is a doc unless it says so.
"""Classify a document.
The schema has no `source_type` field and the schema is frozen, so the
classification is derived from tags the schema does allow.
Harvested documents carry an explicit `source_type`. Curated content does
not: the schema is frozen and has no such field, so it is derived from the
tags the schema does allow.
"""
explicit = str(getattr(doc, "source_type", "") or "").strip()
if explicit:
return explicit
tags = {str(tag).lower() for tag in doc.tags}
if "adr" in tags:
return "adr"
@@ -82,6 +99,12 @@ def build_payload(doc, chunk, cfg) -> dict:
"artefacts": [str(a.get("url", "")) for a in doc.artefacts],
"last_verified": doc.last_verified,
"text": chunk.text,
# -- provenance (docs/phase-4-corpus.md §3) ------------------------
"platform": doc.platform,
"tier": int(doc.tier),
"origin_url": doc.origin_url,
"path": getattr(doc, "repo_path", "") or doc.path,
"last_modified": doc.last_modified,
# -- provenance ----------------------------------------------------
"id": None, # filled by the pipeline once the id is computed
"source_path": doc.path,
@@ -125,7 +148,17 @@ def validate_payload(payload: dict, where: str = "chunk") -> None:
f"{where}: confidence {payload['confidence']!r} is not one of {list(CONFIDENCE)}"
)
for field in ("period", "title", "claim", "url", "source_path", "source_id", "embedding_model"):
if payload["platform"] not in PLATFORMS:
raise PayloadError(
f"{where}: platform {payload['platform']!r} is not one of {list(PLATFORMS)}"
)
if payload["tier"] not in TIERS:
raise PayloadError(f"{where}: tier {payload['tier']!r} is not one of {list(TIERS)}")
for field in (
"period", "title", "claim", "url", "source_path", "source_id",
"embedding_model", "origin_url", "path",
):
value = payload[field]
if not isinstance(value, str) or not value.strip():
raise PayloadError(f"{where}: field {field!r} must be a non-empty string")
@@ -138,6 +171,7 @@ def validate_payload(payload: dict, where: str = "chunk") -> None:
raise PayloadError(f"{where}: field 'artefacts' must be a list (empty is allowed)")
_check_date(payload["last_verified"], where)
_check_timestamp(payload["last_modified"], where)
if not isinstance(payload["embedding_dim"], int) or payload["embedding_dim"] <= 0:
raise PayloadError(f"{where}: field 'embedding_dim' must be a positive integer")
if not isinstance(payload["chunk_index"], int) or payload["chunk_index"] < 0:
@@ -146,6 +180,19 @@ def validate_payload(payload: dict, where: str = "chunk") -> None:
raise PayloadError(f"{where}: field 'content_hash' must be a sha256 hex digest")
def _check_timestamp(value: str, where: str) -> None:
"""`last_modified` is an ISO-8601 instant, e.g. 2026-10-14T05:00:00Z."""
text = str(value)
if text.endswith("Z"):
text = text[:-1] + "+00:00"
try:
datetime.datetime.fromisoformat(text)
except ValueError as exc:
raise PayloadError(
f"{where}: field 'last_modified' must be an ISO-8601 timestamp, got {value!r}"
) from exc
def _check_date(value: str, where: str) -> None:
try:
datetime.date.fromisoformat(str(value))

View File

@@ -1,20 +1,32 @@
"""The pipeline: content/ -> validate -> chunk -> enrich -> embed -> upsert.
"""The ingestion pipeline: estate -> normalise -> chunk -> enrich -> embed -> upsert.
One entry point, `run`, with two modes:
* **rebuild** — drop the collection, empty the store, embed everything. This is
the "one command rebuilds the corpus from an empty volume" path.
* **rebuild** — drop the collections, empty the store, embed everything. This is
the "one command rebuilds the pool from an empty volume" path.
* **sync** — the default. Only documents whose content hash, embedding model,
embedding dimension or chunker version changed are re-chunked and
re-embedded. Everything else is left alone.
embedding dimension or chunker version changed are re-chunked and re-embedded.
Everything else is left alone, including the graph when graph.json is byte-for-byte
unchanged.
The order matters. Payloads are built and validated for *every* changed document
before a single vector is written, so an invalid payload stops the run with an
empty diff rather than leaving half an index behind.
The order matters. Every payload for every changed document is built and
validated *before* a single vector is written, so an invalid payload stops the
run with an empty diff rather than leaving half an index behind. Embedding is one
flat pass over the changed chunks, batched, so the provider call count is a
direct measure of how much work incremental mode actually did.
Deterministic ids (`sha256(source_path + chunk_index + content_hash)`) mean
re-running updates points instead of duplicating them, and the id is a valid
UUID so Qdrant accepts it as a point id without a mapping table.
Two Qdrant collections, one store:
* `resume_corpus` — one point per chunk of a document, from every platform;
* `resume_graph` — one point per graph node and one per graph edge, joined to the
note it came from by `source_note`.
Neither is ever read by the public site (acceptance test A1): the site serves
pre-rendered files and does not open a socket to Qdrant.
"""
from __future__ import annotations
@@ -25,9 +37,15 @@ import time
import uuid
from dataclasses import asdict, dataclass, field
from . import chunker, embeddings, loader, payloads, store as store_module
from . import chunker, embeddings, exclusions, payloads, store as store_module
from .config import CHUNKER_VERSION
from .qdrant import Qdrant, QdrantError
from .sources import estate as estate_module
from .sources import graph as graph_module
# How many texts one embed call carries. The provider may batch smaller
# internally (OmniRoute does); this bounds memory on the ingest host.
EMBED_BATCH = 128
class PipelineError(RuntimeError):
@@ -61,12 +79,22 @@ class RunStats:
points_sent: int = 0
collection_points: int = 0
store_chunks: int = 0
graph_nodes: int = 0
graph_edges: int = 0
graph_points: int = 0
graph_rebuilt: bool = False
rejections: int = 0
images: int = 0
duration_s: float = 0.0
collection: str = ""
graph_collection: str = ""
provider: str = ""
model: str = ""
dim: int = 0
chunker_version: str = CHUNKER_VERSION
platforms: dict = field(default_factory=dict)
rejection_summary: dict = field(default_factory=dict)
warnings: list = field(default_factory=list)
notes: list = field(default_factory=list)
def summary(self) -> str:
@@ -76,10 +104,18 @@ class RunStats:
f"chunks written={self.chunks_written} deleted={self.chunks_deleted}",
f"embedding calls={self.embed_calls} texts={self.texts_embedded}",
f"points sent={self.points_sent} collection points={self.collection_points}",
f"graph nodes={self.graph_nodes} edges={self.graph_edges} points={self.graph_points} "
f"rebuilt={self.graph_rebuilt}",
f"rejections={self.rejections} image references={self.images}",
f"platforms={json.dumps(self.platforms, sort_keys=True)}",
f"provider={self.provider} model={self.model} dim={self.dim} chunker={self.chunker_version}",
f"collection={self.collection} duration={self.duration_s:.2f}s",
f"collection={self.collection} graph_collection={self.graph_collection} "
f"duration={self.duration_s:.2f}s",
]
if self.rejection_summary:
rows.append(f"rejected: {json.dumps(self.rejection_summary, sort_keys=True)}")
rows.extend(self.notes)
rows.extend(f"warning: {warning}" for warning in self.warnings)
return "\n".join(rows)
def to_dict(self) -> dict:
@@ -87,7 +123,7 @@ class RunStats:
def run(cfg, mode: str = "sync", log=None) -> RunStats:
"""Build or refresh the corpus. Returns what happened; raises on any error."""
"""Build or refresh the pool. Returns what happened; raises on any error."""
if mode not in ("sync", "rebuild"):
raise PipelineError(f"unknown mode {mode!r}; expected 'sync' or 'rebuild'")
started = time.time()
@@ -95,15 +131,26 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
stats = RunStats(
mode=mode,
collection=cfg.collection,
graph_collection=cfg.graph_collection,
provider=cfg.provider,
model=cfg.model,
dim=cfg.dim,
chunker_version=cfg.chunker_version,
)
documents = loader.load_documents(cfg)
# 1. Read the estate. Curated content is validated first (loader does it),
# so a broken content tree stops the run before Gitea is touched.
harvest = estate_module.harvest(cfg, log=log)
documents = harvest.documents
stats.documents = len(documents)
log(f"validated {len(documents)} document(s)")
stats.rejections = len(harvest.rejections)
stats.images = len(harvest.images)
stats.rejection_summary = harvest.rejection_summary()
stats.warnings = list(harvest.warnings)
stats.platforms = {
name: {"documents": row.get("documents", 0)} for name, row in harvest.sources.items()
}
log(f"estate: {len(documents)} document(s), {stats.rejections} rejection(s), {stats.images} image reference(s)")
provider = embeddings.build_provider(cfg)
qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
@@ -116,8 +163,7 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
with store_module.Store(cfg.db_path) as store:
contract = _contract(cfg)
mode = _check_contract(store, qdrant, cfg, contract, mode)
_ensure_collection(qdrant, cfg)
_ensure_collection(qdrant, cfg.collection, cfg)
manifest = store.manifest()
changed, removed = _classify(documents, manifest, cfg, mode)
@@ -133,33 +179,45 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
qdrant.delete_by_source(cfg.collection, path)
log(f"removed {path}")
# 2. Build and validate every payload before any write.
prepared = [] # [(doc, payloads)]
for doc in changed:
payloads_for_doc = _prepare(doc, cfg, manifest)
stats.chunks_written += len(payloads_for_doc)
prepared.append((doc, _prepare(doc, cfg)))
all_payloads = [payload for _, rows in prepared for payload in rows]
vectors = provider.embed([p["text"] for p in payloads_for_doc])
if len(vectors) != len(payloads_for_doc):
raise PipelineError(
f"{doc.path}: provider returned {len(vectors)} vectors for "
f"{len(payloads_for_doc)} chunks"
)
# 3. Embed in one flat, batched pass. This is the number the DoD uses to
# prove incremental mode: it is proportional to changed chunks only.
vectors = _embed_all(provider, [payload["text"] for payload in all_payloads])
if len(vectors) != len(all_payloads):
raise PipelineError(
f"provider returned {len(vectors)} vectors for {len(all_payloads)} chunks"
)
cursor = 0
platform_changes = stats.platforms
for doc, rows in prepared:
points = []
for payload, vector in zip(payloads_for_doc, vectors):
for payload in rows:
vector = vectors[cursor]
cursor += 1
if len(vector) != cfg.dim:
raise PipelineError(
f"{doc.path}: vector dimension {len(vector)} != configured {cfg.dim}"
)
points.append({"id": payload["id"], "vector": vector, "payload": payload})
# Order: remove what this document used to have (chunk counts change),
# then write the new rows and points. A crash between the two leaves an
# index that reports as incomplete, never one that claims to be whole.
# Remove what this document used to have, then write the new rows and
# points. A crash between the two leaves an index that reports as
# incomplete, never one that claims to be whole.
stats.chunks_deleted += store.delete_document(doc.path)
qdrant.delete_by_source(cfg.collection, doc.path)
stats.points_sent += qdrant.upsert(cfg.collection, points)
store.write_document(doc, payloads_for_doc)
store.write_document(doc, rows)
stats.chunks_written += len(rows)
bucket = platform_changes.setdefault(doc.platform, {"documents": 0})
key = "updated" if doc.path in manifest else "added"
bucket[key] = bucket.get(key, 0) + 1
if doc.path in manifest:
stats.updated += 1
log(f"updated {doc.path} ({len(points)} chunk(s))")
@@ -170,24 +228,47 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
provider_stats = provider.stats()
stats.embed_calls = provider_stats["calls"]
stats.texts_embedded = provider_stats["texts"]
# 4. Graph: its own collection, rebuilt only when graph.json changed.
graph_stats = _sync_graph(qdrant, store, cfg, harvest, contract, mode, provider, log)
stats.graph_nodes = graph_stats["nodes"]
stats.graph_edges = graph_stats["edges"]
stats.graph_points = graph_stats["points"]
stats.graph_rebuilt = graph_stats["rebuilt"]
stats.notes.extend(graph_stats["notes"])
# 5. Consistency and the credential scan. Both are checks, not hopes.
stats.store_chunks = store.chunk_count()
stats.collection_points = qdrant.point_count(cfg.collection)
if stats.collection_points != stats.store_chunks:
raise PipelineError(
"index is inconsistent after the run: "
f"{stats.collection_points} point(s) in Qdrant but "
f"{stats.store_chunks} chunk(s) in SQLite. Run `python3 -m corpus.cli verify`."
)
_scan_pool(store)
_scan_graph(store)
store.set_meta("collection", contract)
store.set_meta("last_run", {**stats.to_dict(), "finished_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())})
_write_collection_file(cfg, store, stats)
store.set_meta("sources", harvest.sources)
store.set_meta("rejections", harvest.rejections[:200])
store.set_meta("images", harvest.images[:500])
store.set_meta(
"last_run",
{
**stats.to_dict(),
"finished_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
},
)
_write_collection_file(cfg, store, stats, harvest)
stats.duration_s = round(time.time() - started, 3)
return stats
# -- planning -------------------------------------------------------------
def _contract(cfg) -> dict:
"""What this collection *is*. Recorded so a change is a migration, not drift."""
return {
@@ -200,13 +281,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 +309,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 +318,18 @@ def _check_contract(store, qdrant, cfg, contract: dict, mode: str) -> str:
return mode
def _ensure_collection(qdrant: Qdrant, cfg) -> None:
if not qdrant.collection_exists(cfg.collection):
qdrant.create_collection(
cfg.collection, cfg.dim, vectors_on_disk=cfg.vectors_on_disk
)
def _ensure_collection(qdrant: Qdrant, name: str, cfg) -> None:
if not qdrant.collection_exists(name):
qdrant.create_collection(name, cfg.dim, vectors_on_disk=cfg.vectors_on_disk)
return
info = qdrant.collection_info(cfg.collection)
info = qdrant.collection_info(name)
vectors = ((info.get("config") or {}).get("params") or {}).get("vectors")
size = None
if isinstance(vectors, dict):
size = vectors.get("size")
if size is not None and int(size) != int(cfg.dim):
raise PipelineError(
f"collection {cfg.collection!r} has vectors of size {size} but this run "
f"collection {name!r} has vectors of size {size} but this run "
f"is configured for {cfg.dim}. That is a governed migration: rebuild."
)
@@ -268,10 +358,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 +383,118 @@ def _prepare(doc, cfg, manifest: dict) -> list:
return out
def _write_collection_file(cfg, store, stats: RunStats) -> None:
"""Mirror the contract to data/corpus/collection.json for the operator."""
def _embed_all(provider, texts: list) -> list:
out = []
for start in range(0, len(texts), EMBED_BATCH):
out.extend(provider.embed(texts[start : start + EMBED_BATCH]))
return out
# -- graph ----------------------------------------------------------------
def _sync_graph(qdrant, store, cfg, harvest, contract, mode, provider, log) -> dict:
graph = harvest.graph
stored = store.get_meta("graph") or {}
notes = []
if graph is None:
if mode == "rebuild":
qdrant.drop_collection(cfg.graph_collection)
store.write_graph([], [])
store.set_meta("graph", {})
notes.append("graph: no graph.json was available; the graph collection is empty")
return {"nodes": 0, "edges": 0, "points": 0, "rebuilt": True, "notes": notes}
counts = store.graph_counts()
notes.append(
f"graph: graph.json was not available this run; kept the previous graph "
f"({counts['nodes']} nodes, {counts['edges']} edges)"
)
return {**counts, "points": counts["nodes"] + counts["edges"], "rebuilt": False, "notes": notes}
graph_contract = {
"digest": graph.digest,
"source_path": graph.source_path,
"provider": contract["provider"],
"model": contract["model"],
"dim": contract["dim"],
}
rebuild = mode == "rebuild" or stored != graph_contract
if not rebuild:
counts = store.graph_counts()
return {
**counts,
"points": counts["nodes"] + counts["edges"],
"rebuilt": False,
"notes": [f"graph: unchanged ({graph.digest[:12]}), not re-embedded"],
}
if not qdrant.collection_exists(cfg.graph_collection):
_ensure_collection(qdrant, cfg.graph_collection, cfg)
else:
qdrant.drop_collection(cfg.graph_collection)
qdrant.create_collection(cfg.graph_collection, cfg.dim, vectors_on_disk=cfg.vectors_on_disk)
rows = [("node", row) for row in graph.nodes] + [("edge", row) for row in graph.edges]
payloads_for_graph = [graph_module.point_payload(row, kind) for kind, row in rows]
vectors = _embed_all(provider, [payload["text"] for payload in payloads_for_graph])
points = [
{"id": row["point_id"], "vector": vector, "payload": payload}
for (_, row), vector, payload in zip(rows, vectors, payloads_for_graph)
]
sent = qdrant.upsert(cfg.graph_collection, points)
store.write_graph(graph.nodes, graph.edges)
store.set_meta("graph", graph_contract)
log(f"graph: rebuilt {len(graph.nodes)} node(s) and {len(graph.edges)} edge(s), {sent} point(s)")
return {
"nodes": len(graph.nodes),
"edges": len(graph.edges),
"points": sent,
"rebuilt": True,
"notes": notes,
}
# -- checks ---------------------------------------------------------------
def _scan_pool(store) -> None:
"""Search the finished pool for credential-shaped content. Fail, do not warn."""
hits = []
for source_path, chunk_index, text in store.pool_texts():
for kind, line, _excerpt in exclusions.scan_text(text):
hits.append((source_path, chunk_index, kind, line))
if hits:
detail = ", ".join(f"{path}#chunk-{index} ({kind} line {line})" for path, index, kind, line in hits[:10])
raise PipelineError(
f"credential-shaped content reached the pool in {len(hits)} place(s): {detail}. "
"The run is failed rather than published; fix the source or the exclusion rule."
)
def _scan_graph(store) -> None:
rows = store.db.execute("SELECT id, text FROM graph_nodes UNION ALL SELECT id, text FROM graph_edges")
hits = []
for row in rows:
for kind, line, _excerpt in exclusions.scan_text(row["text"]):
hits.append((row["id"], kind, line))
if hits:
detail = ", ".join(f"{node} ({kind} line {line})" for node, kind, line in hits[:10])
raise PipelineError(f"credential-shaped content reached the graph in {len(hits)} place(s): {detail}")
def _write_collection_file(cfg, store, stats: RunStats, harvest) -> None:
"""Mirror the contract and the last run to data/corpus/collection.json."""
body = {
"collection": cfg.collection,
"contract": _contract(cfg),
"qdrant_url": cfg.qdrant_url,
"chunks": store.chunk_count(),
"sources": store.source_count(),
"graph_collection": cfg.graph_collection,
"graph": store.graph_counts(),
"platforms": harvest.sources,
"rejections": harvest.rejection_summary(),
"images": len(harvest.images),
"last_run": store.get_meta("last_run"),
}
with open(cfg.meta_path, "w", encoding="utf-8") as handle:
@@ -309,10 +506,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 +540,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 +555,48 @@ def verify(cfg, log=None) -> list:
f"{doc.path}: {expected} chunk row(s) but {actual} vector(s) in Qdrant"
)
for source in sorted(set(by_source) - {doc.path for doc in documents}):
problems.append(f"{source}: indexed but no longer in content/ (run sync)")
for source in sorted(set(by_source) - set(on_disk)):
problems.append(f"{source}: indexed but no longer in content/ or the estate (run sync)")
total_points = sum(points_by_source.values())
if total_points != store.chunk_count():
problems.append(
f"{total_points} vector(s) in Qdrant but {store.chunk_count()} chunk(s) in SQLite"
)
# Graph: every row must resolve to a note, and Qdrant must agree.
counts = store.graph_counts()
graph_meta = store.get_meta("graph") or {}
if harvest.graph is not None:
if not graph_meta:
problems.append("graph: graph.json is available but the graph was never indexed")
if qdrant.collection_exists(cfg.graph_collection):
graph_points = qdrant.point_count(cfg.graph_collection)
if graph_points != counts["nodes"] + counts["edges"]:
problems.append(
f"graph: {graph_points} point(s) in Qdrant but "
f"{counts['nodes'] + counts['edges']} row(s) in SQLite"
)
elif counts["nodes"] or counts["edges"]:
problems.append(f"graph: collection {cfg.graph_collection!r} does not exist")
for row in store.db.execute("SELECT id, source_note, origin_url FROM graph_nodes LIMIT 5"):
if not row["source_note"] or not row["origin_url"]:
problems.append(f"graph node {row['id']}: no source note or origin")
missing_notes = [
row["id"]
for row in store.db.execute(
"SELECT id FROM graph_nodes WHERE source_note = '' OR origin_url = '' LIMIT 10"
)
]
if missing_notes:
problems.append(f"graph: {len(missing_notes)} node(s) carry no source note")
log(
f"verify: {store.chunk_count()} chunk(s) across {store.source_count()} source(s), "
f"{total_points} vector(s) in Qdrant"
f"{total_points} vector(s) in Qdrant, {counts['nodes']} node(s), {counts['edges']} edge(s)"
)
for warning in harvest.warnings:
problems.append(f"estate warning: {warning}")
return problems
@@ -391,6 +620,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 +631,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 +641,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 +659,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}

View File

@@ -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

848
corpus/sources/estate.py Normal file
View File

@@ -0,0 +1,848 @@
"""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:
reason = str(row.get("reason") or "unknown")
# Keep the summary readable: 200 distinct oversize byte counts are
# one finding, not two hundred.
if reason.startswith("oversize"):
reason = "oversize"
elif reason.startswith("binary"):
reason = "binary"
key = f"{row.get('platform')}/{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")
if not text.strip():
result.add_rejection(exclusions.Rejection(path, "gitea", "empty file", full_name))
return None
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 = _strip_front_matter(text) if kind == "markdown" else text
if not body.strip():
result.add_rejection(exclusions.Rejection(path, "gitea", "front-matter only", full_name))
return None
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")
if not text.strip():
result.add_rejection(exclusions.Rejection(path, "obsidian", "empty file"))
continue
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)
if not body.strip():
result.add_rejection(exclusions.Rejection(path, "obsidian", "empty note"))
continue
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 _strip_front_matter(text: str) -> str:
"""Drop leading YAML blocks. The Note path parses them; files only need them gone.
Some notes carry two blocks in a row (a stale export, or a block followed by
a horizontal rule that closes), so this repeats. It never strips a body that
does not have a closing delimiter.
"""
body = text
for _ in range(3):
lines = body.splitlines()
if not lines or lines[0].strip() != "---":
return body
closing = None
for index in range(1, len(lines)):
if lines[index].strip() in ("---", "..."):
closing = index
break
if closing is None:
return body
body = "\n".join(lines[closing + 1 :]).strip()
return body
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 = {}
# Strip any further leading block as well, so a duplicate front-matter block
# never reaches the chunker.
return fields, _strip_front_matter("\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]
)

165
corpus/sources/gitea.py Normal file
View File

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

335
corpus/sources/graph.py Normal file
View File

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

View File

@@ -27,7 +27,7 @@ import sqlite3
import time
import unicodedata
SCHEMA_VERSION = "1"
SCHEMA_VERSION = "2"
SCHEMA = """
CREATE TABLE IF NOT EXISTS chunks (
@@ -49,6 +49,11 @@ CREATE TABLE IF NOT EXISTS chunks (
embedding_dim INTEGER NOT NULL,
chunker_version TEXT NOT NULL,
chunk_tokens INTEGER NOT NULL,
platform TEXT NOT NULL,
tier INTEGER NOT NULL,
origin_url TEXT NOT NULL,
path TEXT NOT NULL,
last_modified TEXT NOT NULL,
public INTEGER NOT NULL,
artefacts TEXT NOT NULL,
tech TEXT NOT NULL,
@@ -66,6 +71,8 @@ CREATE VIRTUAL TABLE IF NOT EXISTS chunk_fts USING fts5(
text,
area UNINDEXED,
confidence UNINDEXED,
platform UNINDEXED,
tier UNINDEXED,
tokenize = 'porter unicode61 remove_diacritics 2'
);
@@ -84,6 +91,55 @@ CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
-- The knowledge graph is not chunked (docs/phase-4-corpus.md §4). Nodes and
-- edges live in their own tables and their own Qdrant collection, joined to note
-- chunks by source_note. Every row carries the note it came from.
CREATE TABLE IF NOT EXISTS graph_nodes (
id TEXT PRIMARY KEY,
label TEXT NOT NULL,
norm_label TEXT NOT NULL,
node_type TEXT NOT NULL,
community INTEGER,
source_note TEXT NOT NULL,
origin_url TEXT NOT NULL,
area TEXT NOT NULL,
tier INTEGER NOT NULL,
platform TEXT NOT NULL,
last_modified TEXT NOT NULL,
content_hash TEXT NOT NULL,
text TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS graph_nodes_note ON graph_nodes(source_note);
CREATE TABLE IF NOT EXISTS graph_edges (
id TEXT PRIMARY KEY,
source_id TEXT NOT NULL,
target_id TEXT NOT NULL,
relation TEXT NOT NULL,
confidence TEXT NOT NULL,
confidence_score REAL,
source_note TEXT NOT NULL,
origin_url TEXT NOT NULL,
area TEXT NOT NULL,
tier INTEGER NOT NULL,
platform TEXT NOT NULL,
last_modified TEXT NOT NULL,
content_hash TEXT NOT NULL,
text TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS graph_edges_source ON graph_edges(source_id);
CREATE INDEX IF NOT EXISTS graph_edges_target ON graph_edges(target_id);
CREATE INDEX IF NOT EXISTS graph_edges_note ON graph_edges(source_note);
CREATE VIRTUAL TABLE IF NOT EXISTS graph_fts USING fts5(
id UNINDEXED,
kind UNINDEXED,
label,
text,
area UNINDEXED,
tokenize = 'porter unicode61 remove_diacritics 2'
);
"""
# Column order used by every INSERT in this module.
@@ -91,6 +147,7 @@ _CHUNK_COLUMNS = (
"id", "source_path", "source_id", "chunk_index", "area", "level", "source_type",
"confidence", "period", "last_verified", "title", "claim", "url", "content_hash",
"embedding_model", "embedding_dim", "chunker_version", "chunk_tokens",
"platform", "tier", "origin_url", "path", "last_modified",
"public", "artefacts", "tech", "tags", "text",
)
@@ -110,7 +167,13 @@ class Store:
self.db.execute("PRAGMA journal_mode = WAL")
self.db.execute("PRAGMA synchronous = NORMAL")
self.db.executescript(SCHEMA)
self.set_meta("schema_version", SCHEMA_VERSION)
stored = self.get_meta("schema_version")
# A database from an older payload shape has no provenance columns. It is
# derived data, so the fix is a rebuild, but it must never be silently
# mixed with the new shape: `rebuild_schema` is called by a rebuild only.
self.schema_stale = stored not in (None, SCHEMA_VERSION)
if not self.schema_stale:
self.set_meta("schema_version", SCHEMA_VERSION)
# -- lifecycle --------------------------------------------------------
@@ -125,10 +188,22 @@ class Store:
def reset(self) -> None:
"""Empty every table. Used by a from-scratch rebuild."""
for table in ("chunk_fts", "chunks", "manifest", "meta"):
for table in ("chunk_fts", "chunks", "manifest", "meta", "graph_fts", "graph_nodes", "graph_edges"):
self.db.execute(f"DELETE FROM {table}")
self.db.commit()
def rebuild_schema(self) -> None:
"""Drop and recreate every table. Only a rebuild may call this."""
for table in (
"chunk_fts", "chunks", "manifest", "meta",
"graph_fts", "graph_nodes", "graph_edges",
):
self.db.execute(f"DROP TABLE IF EXISTS {table}")
self.db.commit()
self.db.executescript(SCHEMA)
self.schema_stale = False
self.set_meta("schema_version", SCHEMA_VERSION)
# -- writing ----------------------------------------------------------
def write_document(self, doc, payloads: list) -> None:
@@ -161,8 +236,9 @@ class Store:
if value
)
self.db.execute(
"INSERT INTO chunk_fts (rowid, id, source_path, title, terms, text, area, confidence) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
"INSERT INTO chunk_fts "
"(rowid, id, source_path, title, terms, text, area, confidence, platform, tier) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
cursor.lastrowid,
payload["id"],
@@ -172,6 +248,8 @@ class Store:
payload["text"],
payload["area"],
payload["confidence"],
payload["platform"],
payload["tier"],
),
)
self.db.execute(
@@ -237,7 +315,10 @@ class Store:
out.append(payload)
return out
def keyword_search(self, query: str, area: str = None, confidence: str = None, limit: int = 10) -> list:
def keyword_search(
self, query: str, area: str = None, confidence: str = None,
platform: str = None, tier: int = None, limit: int = 10,
) -> list:
"""BM25 over chunk text, optionally filtered before ranking."""
match = match_expression(query)
if not match:
@@ -256,11 +337,110 @@ class Store:
if confidence:
sql.append("AND c.confidence = ?")
args.append(confidence)
if platform:
sql.append("AND c.platform = ?")
args.append(platform)
if tier is not None:
sql.append("AND c.tier = ?")
args.append(int(tier))
sql.append("ORDER BY bm25(chunk_fts) LIMIT ?")
args.append(int(limit))
rows = self.db.execute(" ".join(sql), args)
return [dict(row) for row in rows]
def pool_texts(self) -> list:
"""Every chunk text in the pool. The credential scan reads this."""
return [
(row["source_path"], row["chunk_index"], row["text"])
for row in self.db.execute("SELECT source_path, chunk_index, text FROM chunks")
]
# -- graph ------------------------------------------------------------
def write_graph(self, nodes: list, edges: list) -> None:
"""Replace the whole graph. Nodes and edges are rebuilt together."""
self.db.execute("DELETE FROM graph_fts")
self.db.execute("DELETE FROM graph_nodes")
self.db.execute("DELETE FROM graph_edges")
for node in nodes:
self.db.execute(
"INSERT INTO graph_nodes (id, label, norm_label, node_type, community, "
"source_note, origin_url, area, tier, platform, last_modified, content_hash, text) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
node["id"], node["label"], node.get("norm_label") or _norm(node["label"]),
node.get("node_type") or "entity", node.get("community"),
node["source_note"], node["origin_url"], node["area"], node["tier"],
node["platform"], node["last_modified"], node["content_hash"], node["text"],
),
)
self.db.execute(
"INSERT INTO graph_fts (id, kind, label, text, area) VALUES (?, 'node', ?, ?, ?)",
(node["id"], node["label"], node["text"], node["area"]),
)
for edge in edges:
self.db.execute(
"INSERT INTO graph_edges (id, source_id, target_id, relation, confidence, "
"confidence_score, source_note, origin_url, area, tier, platform, "
"last_modified, content_hash, text) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
edge["id"], edge["source_id"], edge["target_id"], edge["relation"],
edge["confidence"], edge.get("confidence_score"), edge["source_note"],
edge["origin_url"], edge["area"], edge["tier"], edge["platform"],
edge["last_modified"], edge["content_hash"], edge["text"],
),
)
self.db.execute(
"INSERT INTO graph_fts (id, kind, label, text, area) VALUES (?, 'edge', ?, ?, ?)",
(edge["id"], edge["relation"], edge["text"], edge["area"]),
)
self.db.commit()
def graph_counts(self) -> dict:
return {
"nodes": int(self.db.execute("SELECT count(*) FROM graph_nodes").fetchone()[0]),
"edges": int(self.db.execute("SELECT count(*) FROM graph_edges").fetchone()[0]),
}
def graph_node(self, node_id: str) -> dict:
row = self.db.execute("SELECT * FROM graph_nodes WHERE id = ?", (node_id,)).fetchone()
return dict(row) if row else {}
def graph_neighbours(self, node_id: str, limit: int = 50) -> list:
rows = self.db.execute(
"SELECT e.*, sn.label AS source_label, tn.label AS target_label "
"FROM graph_edges e "
"LEFT JOIN graph_nodes sn ON sn.id = e.source_id "
"LEFT JOIN graph_nodes tn ON tn.id = e.target_id "
"WHERE e.source_id = ? OR e.target_id = ? LIMIT ?",
(node_id, node_id, int(limit)),
)
return [dict(row) for row in rows]
def graph_search(self, query: str, area: str = None, limit: int = 10) -> list:
"""Keyword search over node labels and edge relations, with provenance."""
match = match_expression(query)
if not match:
return []
sql = [
"SELECT f.id, f.kind, f.label, f.text, f.area,",
" COALESCE(n.source_note, e.source_note) AS source_note,",
" COALESCE(n.origin_url, e.origin_url) AS origin_url,",
" n.community, e.relation, e.source_id, e.target_id",
"FROM graph_fts f",
"LEFT JOIN graph_nodes n ON f.kind = 'node' AND n.id = f.id",
"LEFT JOIN graph_edges e ON f.kind = 'edge' AND e.id = f.id",
"WHERE graph_fts MATCH ?",
]
args = [match]
if area:
sql.append("AND f.area = ?")
args.append(area)
sql.append("ORDER BY bm25(graph_fts) LIMIT ?")
args.append(int(limit))
return [dict(row) for row in self.db.execute(" ".join(sql), args)]
# -- meta -------------------------------------------------------------
def set_meta(self, key: str, value) -> None:
@@ -280,6 +460,10 @@ class Store:
return {row["key"]: json.loads(row["value"]) for row in self.db.execute("SELECT key, value FROM meta")}
def _norm(label: str) -> str:
return re.sub(r"[^0-9a-z]+", "_", (label or "").lower()).strip("_")
def match_expression(query: str) -> str:
"""Free text -> a safe FTS5 expression: bare words, each a prefix, all required.

View File

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

View File

@@ -0,0 +1,346 @@
"""Ingestion from the estate: exclusions, provenance, incremental work, graph.
These tests never touch the network. `FakeGitea` implements the four read calls
the harvester uses; `FakeQdrant` implements the collection routes the pipeline
uses. The real estate is exercised separately (docs/phase-4-notes.md), so a
failure here is a logic failure, not a connectivity one.
"""
from __future__ import annotations
import hashlib
import json
import os
import unittest
from unittest import mock
from corpus import exclusions, guides, loader, pipeline, store as store_module
from corpus.sources import estate, graph as graph_module
from corpus.tests import support
def gitea_guide(**overrides) -> guides.Guide:
values = dict(
name="gitea", enabled=True, what="Self-hosted Git.", fits=["ai", "platform", "edge", "products"],
read=["readme", "tree", "selected_source", "commits", "languages"],
base="https://gitea.example", auth="GITEA_TOKEN",
include={"repos": "all", "paths": []},
exclude={"repos": [], "paths": []},
limits={"max_file_kb": 200, "max_files_per_repo": 10, "max_commits": 5, "max_tree_entries": 100},
)
values.update(overrides)
return guides.Guide(**values)
class FakeGitea:
def __init__(self, repos=None, trees=None, files=None, commits=None, languages=None):
self.repos_list = repos or []
self.trees = trees or {}
self.files = files or {}
self.commit_list = commits or []
self.language_map = languages or {}
self.calls = []
def repos(self):
return list(self.repos_list)
def tree(self, full_name, ref):
self.calls.append(("tree", full_name, ref))
return list(self.trees.get(full_name, []))
def raw(self, full_name, ref, path):
self.calls.append(("raw", full_name, ref, path))
try:
return self.files[(full_name, path)]
except KeyError:
from corpus.sources.gitea import GiteaNotFound
raise GiteaNotFound(f"{full_name}/{path}") from None
def commits(self, full_name, ref, limit=20):
return list(self.commit_list)
def languages(self, full_name):
return dict(self.language_map)
def repo(full_name="sam/photo-pipeline", **overrides):
values = dict(
full_name=full_name, name=full_name.split("/")[-1], description="Photo ingestion pipeline",
topics=["photo", "python"], default_branch="main", archived=False,
created_at="2024-01-01T00:00:00Z", updated_at="2026-10-14T05:00:00Z",
html_url=f"https://gitea.example/{full_name}",
)
values.update(overrides)
return values
def blob(path, size=None):
return {"type": "blob", "path": path, "size": len(path) if size is None else size}
class ExclusionTest(unittest.TestCase):
"""The DoD probe: a file in an excluded path never appears in the pool."""
def setUp(self):
self.guide = gitea_guide()
self.result = estate.Harvest()
tree = [
blob("README.md"),
blob("node_modules/left-pad/probe.md"),
blob(".venv/lib/probe.py"),
blob(".git/config"),
blob(".trash/old.md"),
blob("docs/.silverbullet.auth.json"),
blob("src/main.py"),
blob("assets/photo.png"),
]
files = {
("sam/photo-pipeline", "README.md"): b"# Photo pipeline\n\nDeduplicates by hash.\n",
("sam/photo-pipeline", "node_modules/left-pad/probe.md"): b"# PROBE\n",
("sam/photo-pipeline", ".venv/lib/probe.py"): b"PROBE = 1\n",
("sam/photo-pipeline", ".git/config"): b"[core]\n",
("sam/photo-pipeline", ".trash/old.md"): b"# old\n",
("sam/photo-pipeline", "docs/.silverbullet.auth.json"): b'{"secret_key": "x"}\n',
("sam/photo-pipeline", "src/main.py"): b"def dedup():\n return True\n",
}
self.client = FakeGitea(
repos=[repo()],
trees={"sam/photo-pipeline": tree},
files=files,
commits=[{"commit": {"message": "dedup by hash", "author": {"date": "2026-10-14T05:00:00Z"}}}],
languages={"Python": 900, "Shell": 100},
)
self.documents = estate._harvest_one_repo(
self.result, self.client, self.guide, repo(), "https://gitea.example"
)
def test_probe_in_excluded_paths_never_appears(self):
paths = {doc.repo_path for doc in self.documents}
for probe in (
"node_modules/left-pad/probe.md",
".venv/lib/probe.py",
".git/config",
".trash/old.md",
"docs/.silverbullet.auth.json",
):
self.assertNotIn(probe, paths)
self.assertIn("README.md", paths)
self.assertIn("src/main.py", paths)
def test_rejections_are_reported_with_a_reason(self):
reasons = {(row["path"], row["reason"]) for row in self.result.rejections}
self.assertIn(("node_modules/left-pad/probe.md", "excluded directory 'node_modules'"), reasons)
self.assertIn(("docs/.silverbullet.auth.json", "credential-shaped file name"), reasons)
self.assertEqual([image["path"] for image in self.result.images], ["assets/photo.png"])
def test_media_is_an_image_reference_not_text(self):
self.assertEqual([image["path"] for image in self.result.images], ["assets/photo.png"])
for doc in self.documents:
self.assertNotIn("assets/photo.png", doc.body)
def test_every_document_carries_provenance(self):
for doc in self.documents:
self.assertEqual(doc.platform, "gitea")
self.assertEqual(doc.tier, 2)
self.assertTrue(doc.origin_url.startswith("https://gitea.example/sam/photo-pipeline"))
self.assertTrue(doc.last_modified)
self.assertIn(doc.area, self.guide.fits)
class CredentialContentTest(unittest.TestCase):
def test_a_valid_path_with_a_private_key_is_rejected(self):
result = estate.Harvest()
guide = gitea_guide()
client = FakeGitea(
repos=[repo()],
trees={"sam/photo-pipeline": [blob("README.md"), blob("src/leak.py")]},
files={
("sam/photo-pipeline", "README.md"): b"# Photo pipeline\n",
("sam/photo-pipeline", "src/leak.py"): (
("KEY = '" + "-----BEGIN " + "PRIVATE KEY-----'\n").encode("utf-8")
),
},
commits=[],
languages={"Python": 10},
)
estate._harvest_one_repo(result, client, guide, repo(), "https://gitea.example")
reasons = {row["reason"]: row for row in result.rejections}
self.assertIn("credential-shaped content", reasons)
self.assertEqual(reasons["credential-shaped content"]["path"], "src/leak.py")
class PayloadProvenanceTest(unittest.TestCase):
def test_missing_provenance_field_fails_the_payload(self):
root = support.temp_repo()
self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True))
docs = loader.load_documents(support.make_config(root))
chunk = __import__("corpus.chunker", fromlist=["chunk_document"]).chunk_document(docs[0].body)[0]
payload = __import__("corpus.payloads", fromlist=["build_payload"]).build_payload(
docs[0], chunk, support.make_config(root)
)
payload["id"] = pipeline.chunk_id(docs[0].path, 0, docs[0].digest)
for field in ("platform", "tier", "origin_url", "path", "last_modified"):
broken = dict(payload)
broken.pop(field)
with self.assertRaises(Exception):
__import__("corpus.payloads", fromlist=["validate_payload"]).validate_payload(broken)
GRAPH = {
"built_at_commit": "abc123",
"nodes": [
{"id": "photo_ingest", "label": "Photo ingest", "source_file": "300 areas/Photo.md",
"community": 7, "node_type": "entity"},
{"id": "settings", "label": "settings.json", "source_file": "settings.json",
"community": 118, "node_type": "code"},
{"id": "image_node", "label": "diagram", "source_file": "300 areas/diagram.png",
"community": 7, "node_type": "image"},
],
"links": [
{"source": "photo_ingest", "target": "settings", "relation": "configured_by",
"source_file": "300 areas/Photo.md", "confidence": "EXTRACTED", "confidence_score": 0.9},
{"source": "settings", "target": "photo_ingest", "relation": "configures",
"source_file": "settings.json", "confidence": "INFERRED"},
],
}
class GraphTest(unittest.TestCase):
def parse(self):
return graph_module.parse(
json.dumps(GRAPH).encode("utf-8"),
platform="graphify", base="https://gitea.example", repo="sam/obsidian-vault",
ref="main", last_modified="2026-10-14T05:00:00Z", area="ai",
vault_paths={"300 areas/Photo.md", "300 areas/diagram.png"},
)
def test_nodes_and_edges_carry_their_source_note(self):
result = self.parse()
self.assertEqual(len(result.nodes), 1)
self.assertEqual(result.nodes[0]["id"], "photo_ingest")
self.assertEqual(result.nodes[0]["source_note"], "300 areas/Photo.md")
self.assertIn("/sam/obsidian-vault/src/branch/main/300%20areas/Photo.md", result.nodes[0]["origin_url"])
self.assertTrue(result.edges)
for edge in result.edges:
self.assertTrue(edge["source_note"])
def test_a_node_without_a_resolvable_note_is_refused(self):
result = self.parse()
rejected = {row["path"] for row in result.rejections}
self.assertIn("settings.json", rejected)
self.assertNotIn("settings", {row["id"] for row in result.nodes})
def test_media_becomes_an_image_reference(self):
result = self.parse()
self.assertEqual([image["path"] for image in result.images], ["300 areas/diagram.png"])
self.assertNotIn("image_node", {row["id"] for row in result.nodes})
def test_graph_query_returns_the_note(self):
root = support.temp_repo()
self.addCleanup(lambda: __import__("shutil").rmtree(root, ignore_errors=True))
cfg = support.make_config(root)
harvest = estate.Harvest()
harvest.documents = loader.load_documents(cfg)
harvest.graph = self.parse()
harvest.sources["graphify"] = {"platform": "graphify", "enabled": True, "documents": 0}
fake = support.FakeQdrant(cfg.dim)
with mock.patch("corpus.sources.estate.harvest", return_value=harvest):
with support.fake_qdrant(cfg, fake):
pipeline.run(cfg, mode="rebuild")
with store_module.Store(cfg.db_path) as store:
rows = store.graph_search("photo")
self.assertTrue(rows)
self.assertTrue(all(row["source_note"] for row in rows))
self.assertEqual(store.graph_node("photo_ingest")["source_note"], "300 areas/Photo.md")
self.assertTrue(store.graph_neighbours("photo_ingest"))
class IncrementalEstateTest(unittest.TestCase):
def setUp(self):
self.root = support.temp_repo()
self.addCleanup(lambda: __import__("shutil").rmtree(self.root, ignore_errors=True))
self.cfg = support.make_config(self.root)
self.fake = support.FakeQdrant(self.cfg.dim)
self.harvest = estate.Harvest()
self.harvest.documents = loader.load_documents(self.cfg)
self.harvest.documents.append(self.harvested("a" * 64, "First version of the repository."))
self.harvest.sources["gitea"] = {"platform": "gitea", "enabled": True, "documents": 1}
def harvested(self, digest, body):
return loader.Document(
path="gitea:sam/photo-pipeline/README.md", source_id="sam/photo-pipeline/README.md",
area="platform", level=3, title="sam/photo-pipeline — README.md",
claim=body, confidence="historic", period="2024–2026", tech=[], metrics=[], artefacts=[],
public=True, last_verified="2026-10-14", tags=["gitea"], body=body, digest=digest,
url="https://gitea.example/sam/photo-pipeline/src/branch/main/README.md",
platform="gitea", tier=2, origin_url="https://gitea.example/sam/photo-pipeline/src/branch/main/README.md",
last_modified="2026-10-14T05:00:00Z", source_type="doc", kind="markdown",
repo_path="README.md",
)
def run_pipeline(self, mode="sync"):
with mock.patch("corpus.sources.estate.harvest", return_value=self.harvest):
with support.fake_qdrant(self.cfg, self.fake):
return pipeline.run(self.cfg, mode=mode)
def test_rebuild_and_second_run_do_not_duplicate(self):
first = self.run_pipeline("rebuild")
points = len(self.fake.collections[self.cfg.collection])
self.assertEqual(first.platforms["gitea"]["documents"], 1)
second = self.run_pipeline("sync")
self.assertEqual(len(self.fake.collections[self.cfg.collection]), points)
self.assertEqual(second.added, 0)
self.assertEqual(second.updated, 0)
self.assertEqual(second.texts_embedded, 0)
def test_only_the_changed_harvested_document_is_reembedded(self):
self.run_pipeline("rebuild")
self.harvest.documents[-1] = self.harvested("b" * 64, "Second version, one line changed.")
stats = self.run_pipeline("sync")
self.assertEqual(stats.updated, 1)
expected = len(__import__("corpus.chunker", fromlist=["chunk_document"]).chunk_document(
"Second version, one line changed."
))
self.assertEqual(stats.texts_embedded, expected)
self.assertEqual(stats.embed_calls, 1)
self.assertEqual(stats.unchanged, 3)
def test_pool_scan_finds_a_planted_secret(self):
self.harvest.documents[-1] = self.harvested(
"c" * 64,
"AWS_ACCESS_KEY_ID=" + "AKIA" + "ABCDEFGHIJKLMNOP" + "\nThis line is a planted credential.",
)
with self.assertRaises(pipeline.PipelineError) as caught:
self.run_pipeline("rebuild")
self.assertIn("credential-shaped content", str(caught.exception))
class CodeChunkerTest(unittest.TestCase):
def test_a_function_is_never_cut(self):
from corpus import chunker
body = "\n\n".join(
f"def function_{index}():\n value = {{'index': {index}}}\n return value"
for index in range(120)
)
chunks = chunker.chunk_code(body)
self.assertGreater(len(chunks), 1)
for chunk in chunks:
depth = 0
for line in chunk.text.splitlines():
depth = chunker._code_depth(line, depth)
self.assertEqual(depth, 0, f"chunk {chunk.index} ends inside a block")
self.assertIn("def function_", chunk.text)
def test_oversize_single_function_is_flagged_not_cut(self):
from corpus import chunker
body = "def huge():\n" + "\n".join(f" x{index} = {index}" for index in range(2000))
chunks = chunker.chunk_code(body)
self.assertEqual(len(chunks), 1)
self.assertTrue(chunks[0].oversize)
if __name__ == "__main__":
unittest.main()

View File

@@ -30,11 +30,13 @@ services:
restart: unless-stopped
mem_limit: 120m
memswap_limit: 120m
# The app listens on 0.0.0.0 inside the container namespace; the host-side
# binding is loopback, so nothing is reachable from the LAN except through
# docker-proxy on 127.0.0.1:8096 — which is what Caddy proxies to.
# Published on this host's LAN address, not loopback. Caddy lives on .35 and
# proxies to 192.168.20.13:8096, exactly as it does for offsite (:8095),
# photo-filter (:8092), garage-ui (:3909) and prefect (:4200). The port is
# reachable inside the house only; Caddy on .35 remains the sole public
# listener, which is the intent of AGENTS.md section 4.
ports:
- "127.0.0.1:8096:8096"
- "192.168.20.13:8096:8096"
read_only: true
security_opt:
- no-new-privileges:true

View File

@@ -51,5 +51,16 @@ services:
retries: 3
start_period: 10s
# The host's docker daemon has every default address pool allocated by other
# projects, so an implicit compose network cannot be created here. Pin a /28 of
# RFC1918 space that no other bridge uses; it is separate from the site's
# 10.89.0.0/28 so stopping one project cannot disturb the other.
networks:
default:
ipam:
config:
- subnet: 10.89.0.16/28
gateway: 10.89.0.17
volumes:
qdrant_storage:

View File

@@ -48,13 +48,49 @@ from corpus.config import Config # noqa: E402
@_flow(name="corpus-sync", log_prints=True)
def sync_corpus() -> dict:
"""Incremental refresh. Runs on every push that touches content/."""
"""Incremental refresh of the whole estate. Runs on a push or on demand."""
logger = _get_run_logger()
cfg = Config.from_env()
stats = pipeline.run(cfg, mode="sync", log=logger.info)
return stats.to_dict()
@_flow(name="corpus-scheduled-ingest", log_prints=True)
def scheduled_ingest(alert: bool = True) -> dict:
"""Nightly: harvest, report what changed, then audit and alert on staleness.
This is the scheduled run the DoD asks for: it reports documents added,
updated and removed per platform, and it raises an Apprise alert when a
source is missing, unavailable or has stopped changing.
"""
logger = _get_run_logger()
cfg = Config.from_env()
stats = pipeline.run(cfg, mode="sync", log=logger.info)
logger.info(stats.summary())
report = freshness.audit(cfg, log=logger.info)
logger.info(freshness.render_text(report))
if alert and not report["ok"]:
try:
freshness.alert(report, cfg.apprise_url)
except Exception as exc: # noqa: BLE001 - the alert must not fail the run
logger.warning(f"Apprise alert failed: {exc}")
return {
"changed": {
"documents": stats.documents,
"added": stats.added,
"updated": stats.updated,
"removed": stats.removed,
"unchanged": stats.unchanged,
"embedding_calls": stats.embed_calls,
"texts_embedded": stats.texts_embedded,
"platforms": stats.platforms,
"rejections": stats.rejection_summary,
"graph_rebuilt": stats.graph_rebuilt,
},
"audit": report,
}
@_flow(name="corpus-rebuild", log_prints=True)
def rebuild_corpus() -> dict:
"""From-scratch rebuild. The recovery path: deleting the volume is always safe."""
@@ -85,5 +121,7 @@ if __name__ == "__main__": # pragma: no cover - convenience for a manual run
print(freshness.render_text(freshness_audit(alert=False)))
elif command == "rebuild":
print(rebuild_corpus())
elif command == "scheduled":
print(scheduled_ingest(alert=False))
else:
print(sync_corpus())

46
platforms/gitea.yml Normal file
View File

@@ -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

23
platforms/graphify.yml Normal file
View File

@@ -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

44
platforms/obsidian.yml Normal file
View File

@@ -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