Research validator
Checks sources, prose, execution, links, rendering, and generated artifacts.
#!/usr/bin/env python3
"""Validate research artifacts, source metadata, links, and executed notebooks."""
import ast
import hashlib
import html
import json
import re
import sys
from collections import Counter
from html.parser import HTMLParser
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Dict, Iterable, List, Mapping, Sequence, Tuple
from urllib.parse import unquote, urljoin, urlparse
ROOT = Path(__file__).resolve().parents[1]
SOURCE_REGISTRY = ROOT / "research" / "sources.json"
FIELD_NOTEBOOK = ROOT / "research" / "field_notebook"
FIELD_NOTEBOOK_CSS = ROOT / "assets" / "field_notebook.css"
PREVIEW_DIRECTORY = ROOT / "previews"
PUBLICATION_INDEX = PREVIEW_DIRECTORY / "index.html"
SITE_MANIFEST = PREVIEW_DIRECTORY / "site_manifest.json"
RETIRED_MONOLITH = PREVIEW_DIRECTORY / "00_complete_rag_handbook.html"
BUILD_PROVENANCE = ROOT / "previews" / "build_provenance.json"
SYSTEM_DESIGNS = ROOT / "research" / "system_designs.json"
BOOK_PDF = ROOT / "output" / "pdf" / "the-evidence-path-readers-edition.pdf"
PUBLIC_SITE_ORIGIN = "https://rag.babushkai.com"
REQUIRED_RESEARCH_NAMES = (
"README.md",
"field_map.md",
"chronology.md",
"chronological_index.md",
"corpus_and_indexing.md",
"retrieval_and_ranking.md",
"context_and_generation.md",
"training_and_optimization.md",
"structured_and_multimodal_rag.md",
"agents_memory_and_temporal.md",
"frontier_2024_2026.md",
"evaluation_and_risks.md",
"security_privacy_and_governance.md",
"production_systems.md",
"mathematical_primer.md",
"decision_guide.md",
"glossary.md",
"coverage_matrix.md",
)
REQUIRED_RESEARCH = tuple(ROOT / "research" / name for name in REQUIRED_RESEARCH_NAMES)
REQUIRED_FIELD_NOTEBOOK_NAMES = (
"00_prologue.md",
"01_foundations_and_retrieval.md",
"02_generation_and_grounding.md",
"03_agents_memory_security.md",
"04_evaluation_production.md",
"05_epilogue.md",
)
REQUIRED_FIELD_NOTEBOOK = tuple(
FIELD_NOTEBOOK / name for name in REQUIRED_FIELD_NOTEBOOK_NAMES
)
REQUIRED_NOTEBOOKS = (
ROOT / "notebooks" / "00_complete_rag_handbook.ipynb",
ROOT / "notebooks" / "01_rag_evolution.ipynb",
ROOT / "notebooks" / "02_advanced_rag.ipynb",
ROOT / "notebooks" / "03_evaluation_and_failure_analysis.ipynb",
ROOT / "notebooks" / "04_corpus_chunking_and_indexes.ipynb",
ROOT / "notebooks" / "05_training_query_fusion_and_reranking.ipynb",
ROOT / "notebooks" / "06_structured_multimodal_and_graph_rag.ipynb",
ROOT / "notebooks" / "07_agents_memory_temporal_and_security.ipynb",
ROOT / "notebooks" / "08_production_evaluation_and_cost.ipynb",
)
FOCUSED_NOTEBOOK_NAMES = (
"01_rag_evolution.ipynb",
"04_corpus_chunking_and_indexes.ipynb",
"02_advanced_rag.ipynb",
"05_training_query_fusion_and_reranking.ipynb",
"06_structured_multimodal_and_graph_rag.ipynb",
"07_agents_memory_temporal_and_security.ipynb",
"03_evaluation_and_failure_analysis.ipynb",
"08_production_evaluation_and_cost.ipynb",
)
READER_ROUTE_SLUGS = (
"prologue",
"search",
"grounding",
"agents-memory-security",
"evaluation-production",
"epilogue",
)
RESEARCH_PAGE_ORDER = (
"README.md",
"field_map.md",
"mathematical_primer.md",
"chronology.md",
"frontier_2024_2026.md",
"corpus_and_indexing.md",
"retrieval_and_ranking.md",
"context_and_generation.md",
"training_and_optimization.md",
"structured_and_multimodal_rag.md",
"agents_memory_and_temporal.md",
"security_privacy_and_governance.md",
"evaluation_and_risks.md",
"production_systems.md",
"decision_guide.md",
"chronological_index.md",
"glossary.md",
"coverage_matrix.md",
)
REQUIRED_SITE_SCRIPTS = (
ROOT / "scripts" / "build_chronological_index.py",
ROOT / "scripts" / "build_curriculum_notebooks.py",
ROOT / "scripts" / "execute_notebooks.py",
ROOT / "scripts" / "render_field_notebook_preview.py",
ROOT / "scripts" / "python_source_renderer.py",
ROOT / "scripts" / "serve_notebook_site.py",
ROOT / "scripts" / "validate_research.py",
)
REQUIRED_MODULES = (
"agentic.py",
"chunking.py",
"context.py",
"evaluation.py",
"indexes.py",
"ingestion.py",
"memory.py",
"operations.py",
"pipeline.py",
"rerankers.py",
"retrievers.py",
"security.py",
"selection.py",
"structured.py",
"temporal.py",
"training.py",
)
LINK_PATTERN = re.compile(r"\[[^\]]*\]\(([^)]+)\)")
WORD_PATTERN = re.compile(r"\b[\w'-]+\b", re.UNICODE)
# Budgets apply to generated HTML, not source artifacts. They keep the landing
# page instant and prevent notebooks or highlighted Python from silently turning
# back into one multi-megabyte document.
ROOT_PAGE_BUDGET = 16 * 1024
COLLECTION_PAGE_BUDGET = 32 * 1024
SYSTEM_INDEX_BUDGET = 12 * 1024
SYSTEM_PAGE_BUDGET = 24 * 1024
READER_PAGE_BUDGET = 64 * 1024
RESEARCH_PAGE_BUDGET = 128 * 1024
NOTEBOOK_PAGE_BUDGET = 128 * 1024
PYTHON_PAGE_BUDGET = 512 * 1024
PUBLICATION_BUDGET = 4 * 1024 * 1024
class ValidationError(RuntimeError):
pass
class PreviewHTMLAudit(HTMLParser):
"""Collect structural, identity, and navigation evidence from the preview."""
VOID_ELEMENTS = {
"area",
"base",
"br",
"col",
"embed",
"hr",
"img",
"input",
"link",
"meta",
"param",
"source",
"track",
"wbr",
}
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.identifiers: List[str] = []
self.hrefs: List[str] = []
self.elements: List[Tuple[str, Dict[str, object]]] = []
self.stack: List[str] = []
self.errors: List[str] = []
self.workbench_notes = 0
self.workbench_notes_with_output = 0
self.binding_notes = 0
self.binding_notes_with_source_and_output = 0
self._component_frames: List[Dict[str, object]] = []
@staticmethod
def _classes(attributes: Sequence[Tuple[str, str]]) -> set:
return set(str(dict(attributes).get("class") or "").split())
def _track_component_start(
self,
attributes: Sequence[Tuple[str, str]],
) -> None:
classes = self._classes(attributes)
if "workbench-note" in classes:
self.workbench_notes += 1
self._component_frames.append(
{"kind": "workbench", "depth": len(self.stack), "output": False}
)
if "binding-note" in classes:
self.binding_notes += 1
self._component_frames.append(
{
"kind": "binding",
"depth": len(self.stack),
"source": False,
"output": False,
}
)
for frame in self._component_frames:
if "output-slip" in classes:
frame["output"] = True
if frame["kind"] == "binding" and "source-fold" in classes:
frame["source"] = True
def _finish_components(self) -> None:
while (
self._component_frames
and int(self._component_frames[-1]["depth"]) > len(self.stack)
):
frame = self._component_frames.pop()
if frame["kind"] == "workbench" and frame["output"]:
self.workbench_notes_with_output += 1
if (
frame["kind"] == "binding"
and frame["source"]
and frame["output"]
):
self.binding_notes_with_source_and_output += 1
def _collect_attributes(
self,
tag: str,
attributes: Sequence[Tuple[str, str]],
) -> None:
values = dict(attributes)
self.elements.append((tag, values))
if values.get("id"):
self.identifiers.append(values["id"])
if values.get("href"):
self.hrefs.append(values["href"])
def handle_starttag(
self,
tag: str,
attributes: Sequence[Tuple[str, str]],
) -> None:
self._collect_attributes(tag, attributes)
if tag == "p" and self.stack and self.stack[-1] in {"ul", "ol"}:
self.errors.append(f"paragraph is a direct child of {self.stack[-1]}")
if tag not in self.VOID_ELEMENTS:
self.stack.append(tag)
self._track_component_start(attributes)
def handle_startendtag(
self,
tag: str,
attributes: Sequence[Tuple[str, str]],
) -> None:
self._collect_attributes(tag, attributes)
self._track_component_start(attributes)
self._finish_components()
def handle_endtag(self, tag: str) -> None:
if not self.stack:
self.errors.append(f"unmatched closing tag: {tag}")
return
if self.stack[-1] != tag:
self.errors.append(
f"closing tag {tag} encountered inside {self.stack[-1]}"
)
if tag in self.stack:
while self.stack and self.stack[-1] != tag:
self.stack.pop()
else:
return
self.stack.pop()
self._finish_components()
def elements_with_class(audit: PreviewHTMLAudit, class_name: str) -> int:
"""Count parsed elements carrying one exact CSS class token."""
return sum(
class_name in str(attributes.get("class", "")).split()
for _, attributes in audit.elements
)
def require(condition: bool, message: str) -> None:
if not condition:
raise ValidationError(message)
def validate_sources() -> Tuple[int, Counter, Counter]:
payload = json.loads(SOURCE_REGISTRY.read_text(encoding="utf-8"))
require(payload.get("evidence_cutoff") == "2026-08-09", "unexpected evidence cutoff")
statuses = set(payload.get("status_vocabulary", ()))
require(statuses == {"peer-reviewed", "preprint", "industry-report", "benchmark-program"}, "invalid status vocabulary")
sources = payload.get("sources")
require(isinstance(sources, list) and len(sources) >= 200, "source registry is too small")
required = {"id", "first_public", "title", "venue", "status", "primary_url", "topics"}
identifiers = set()
urls = set()
counts: Counter = Counter()
topic_counts: Counter = Counter()
for index, source in enumerate(sources):
require(isinstance(source, dict), f"source {index} is not an object")
missing = required - set(source)
require(not missing, f"source {index} missing fields: {sorted(missing)}")
identifier = source["id"]
require(isinstance(identifier, str) and re.fullmatch(r"[a-z0-9][a-z0-9-]*", identifier) is not None, f"invalid source id: {identifier}")
require(identifier not in identifiers, f"duplicate source id: {identifier}")
identifiers.add(identifier)
status = source["status"]
require(status in statuses, f"invalid source status for {identifier}: {status}")
counts[status] += 1
url = source["primary_url"]
parsed = urlparse(url)
require(parsed.scheme == "https" and parsed.netloc, f"invalid primary URL for {identifier}")
require(url not in urls, f"duplicate primary URL: {url}")
urls.add(url)
first_public = str(source["first_public"])
require(re.fullmatch(r"\d{4}(?:-\d{2}(?:-\d{2})?)?", first_public) is not None, f"invalid first-public date for {identifier}: {first_public}")
year = int(first_public[:4])
require(1970 <= year <= 2026, f"out-of-range first-public year for {identifier}: {year}")
require(isinstance(source["topics"], list) and source["topics"], f"missing topics for {identifier}")
for topic in source["topics"]:
require(isinstance(topic, str) and re.fullmatch(r"[a-z0-9][a-z0-9-]*", topic) is not None, f"invalid topic for {identifier}: {topic}")
topic_counts[topic] += 1
required_topics = {
"document-parsing",
"chunking",
"sparse-retrieval",
"learned-sparse",
"dense-retrieval",
"late-interaction",
"ann",
"reranking",
"query-transformation",
"fusion",
"context-selection",
"generation",
"citations",
"agentic-rag",
"memory",
"freshness",
"graph-rag",
"multimodal",
"evaluation",
"security",
"privacy",
"systems",
"serving",
}
require(not (required_topics - set(topic_counts)), f"source registry missing topic families: {sorted(required_topics - set(topic_counts))}")
require(counts["peer-reviewed"] >= 170, "too few peer-reviewed primary sources")
return len(sources), counts, topic_counts
def markdown_files() -> Iterable[Path]:
yield ROOT / "README.md"
yield from REQUIRED_RESEARCH
yield from REQUIRED_FIELD_NOTEBOOK
def validate_markdown() -> Tuple[int, int]:
total_words = 0
external_links = set()
for path in markdown_files():
require(path.is_file(), f"missing Markdown artifact: {path.relative_to(ROOT)}")
text = path.read_text(encoding="utf-8")
require("TODO" not in text and "TBD" not in text, f"placeholder found in {path.name}")
total_words += len(WORD_PATTERN.findall(text))
for raw_target in LINK_PATTERN.findall(text):
target = raw_target.strip().strip("<>")
if target.startswith(("https://", "http://")):
external_links.add(target)
continue
if target.startswith(("#", "mailto:")):
continue
relative = target.split("#", 1)[0]
if not relative:
continue
resolved = (path.parent / relative).resolve()
require(
resolved.exists(),
f"broken internal link in {path.relative_to(ROOT)}: {target}",
)
require(total_words >= 60000, f"research narrative unexpectedly short: {total_words} words")
require(len(external_links) >= 190, f"too few external primary-source links: {len(external_links)}")
return total_words, len(external_links)
def validate_research_structure() -> Tuple[int, int, int]:
"""Verify the field map, chronological index, glossary, and coverage contract."""
registry = json.loads(SOURCE_REGISTRY.read_text(encoding="utf-8"))
chronology = (ROOT / "research" / "chronological_index.md").read_text(encoding="utf-8")
for source in registry["sources"]:
require(
chronology.count(source["primary_url"]) == 1,
f"chronological index does not contain exactly one URL for {source['id']}",
)
chronology_rows = sum(
bool(re.match(r"^\| \d+ \|", line)) for line in chronology.splitlines()
)
require(chronology_rows == len(registry["sources"]), "chronological source-row mismatch")
coverage = (ROOT / "research" / "coverage_matrix.md").read_text(encoding="utf-8")
coverage_rows = sum(line.startswith("|") for line in coverage.splitlines())
require(coverage_rows >= 160, f"coverage matrix is unexpectedly small: {coverage_rows} rows")
required_surfaces = (
"Corpus acquisition, parsing, and governance",
"Retrieval units and chunking",
"Indexes and first-stage retrieval",
"Query understanding, fusion, and ranking",
"Context and generation",
"Learning and optimization",
"Structured and multimodal RAG",
"Adaptive search, agents, memory, and time",
"Grounding, evaluation, and statistics",
"Security, privacy, and governance",
"Production systems and economics",
)
for surface in required_surfaces:
require(surface in coverage, f"coverage matrix missing surface: {surface}")
glossary = (ROOT / "research" / "glossary.md").read_text(encoding="utf-8")
glossary_terms = len(re.findall(r"^\*\*[^*]+\*\*\s+—", glossary, re.MULTILINE))
require(glossary_terms >= 120, f"glossary is unexpectedly small: {glossary_terms} terms")
return chronology_rows, coverage_rows, glossary_terms
def validate_field_notebook() -> Tuple[int, int, int]:
"""Verify that the expressive narrative remains substantial and portable."""
require(FIELD_NOTEBOOK_CSS.is_file(), "missing field-notebook visual language")
css = FIELD_NOTEBOOK_CSS.read_text(encoding="utf-8")
required_classes = (
".field-cover",
".folio-opener",
".experiment-opener",
".margin-note",
".field-question",
".observation",
".experiment",
".sketch",
".insert-legend",
".evidence-leaf",
".bench-insert",
".binding-note",
".workbench-note",
".output-slip",
)
for class_name in required_classes:
require(class_name in css, f"field-notebook CSS missing {class_name}")
require(
re.search(r"@media\s*\(\s*max-width\s*:", css) is not None,
"field-notebook CSS has no mobile layout",
)
require("@media print" in css, "field-notebook CSS has no print layout")
words = 0
margin_notes = 0
experiments = 0
for path in REQUIRED_FIELD_NOTEBOOK:
require(path.is_file(), f"missing narrative folio: {path.name}")
value = path.read_text(encoding="utf-8")
words += len(WORD_PATTERN.findall(value))
margin_notes += value.count('class="margin-note"')
experiments += value.count('class="experiment"')
require(
"field-question" in value or "question:" in value,
f"narrative folio has no opening question: {path.name}",
)
require(words >= 14000, f"field-notebook narrative is unexpectedly short: {words} words")
require(margin_notes >= 12, f"too few handwritten margin notes: {margin_notes}")
require(experiments >= 6, f"too few field experiments: {experiments}")
return words, margin_notes, experiments
def validate_notebooks() -> Tuple[int, int, int, int]:
code_cells = 0
markdown_cells = 0
output_cells = 0
markdown_words = 0
for path in REQUIRED_NOTEBOOKS:
require(path.is_file(), f"missing notebook: {path.relative_to(ROOT)}")
notebook = json.loads(path.read_text(encoding="utf-8"))
require(notebook.get("nbformat") == 4, f"wrong nbformat: {path.name}")
require(notebook.get("metadata", {}).get("kernelspec", {}).get("name") == "python3", f"missing Python kernelspec: {path.name}")
expected_presentation = (
"expressive-field-notebook"
if path == REQUIRED_NOTEBOOKS[0]
else "plain-field-notebook"
)
require(
notebook.get("metadata", {}).get("rag_evolution", {}).get("presentation")
== expected_presentation,
f"missing field-notebook presentation metadata: {path.name}",
)
first_cell = notebook.get("cells", [{}])[0]
first_text = "".join(first_cell.get("source", []))
require("<style>" in first_text, f"portable CSS is not embedded: {path.name}")
require("field-cover" in first_text, f"notebook cover is missing: {path.name}")
for cell_index, cell in enumerate(notebook.get("cells", []), start=1):
cell_type = cell.get("cell_type")
if cell_type == "markdown":
markdown_cells += 1
source = cell.get("source", "")
source_text = "".join(source) if isinstance(source, list) else str(source)
markdown_words += len(WORD_PATTERN.findall(source_text))
continue
if cell_type != "code":
continue
code_cells += 1
require(cell.get("execution_count") is not None, f"unexecuted code cell {path.name}:{cell_index}")
outputs = cell.get("outputs", [])
require(not any(output.get("output_type") == "error" for output in outputs), f"error output in {path.name}:{cell_index}")
output_text = "\n".join(
"".join(output.get("text", []))
if isinstance(output.get("text", ""), list)
else str(output.get("text", ""))
for output in outputs
)
require(
"/Users/" not in output_text and "\\Users\\" not in output_text,
f"machine-specific path in output {path.name}:{cell_index}",
)
if outputs:
output_cells += 1
require(code_cells >= 75, f"not enough executable notebook material: {code_cells} cells")
require(markdown_cells >= 90, f"not enough notebook explanation: {markdown_cells} cells")
require(output_cells >= 75, f"not enough saved notebook outputs: {output_cells} cells")
require(markdown_words >= 70000, f"notebook curriculum is unexpectedly shallow: {markdown_words} words")
complete_path = ROOT / "notebooks" / "00_complete_rag_handbook.ipynb"
complete = complete_path.read_text(encoding="utf-8")
complete_payload = json.loads(complete)
complete_source = "\n".join(
"".join(cell.get("source", []))
for cell in complete_payload.get("cells", [])
if cell.get("cell_type") == "markdown"
)
complete_cells = complete_payload.get("cells", [])
reader_source = "\n".join(
"".join(cell.get("source", []))
for cell in complete_cells
if cell.get("cell_type") == "markdown"
and "binding-note" not in cell.get("metadata", {}).get("tags", [])
)
for name, source_path in zip(REQUIRED_RESEARCH_NAMES, REQUIRED_RESEARCH):
require(f"research/{name}" in complete_source, f"complete handbook notebook missing {name}")
digest = hashlib.sha256(source_path.read_bytes()).hexdigest()
require(
f'data-source-sha256="{digest}"' in complete_source,
f"complete handbook notebook has stale content for {name}",
)
for name, source_path in zip(REQUIRED_FIELD_NOTEBOOK_NAMES, REQUIRED_FIELD_NOTEBOOK):
require(
f"research/field_notebook/{name}" in complete_source,
f"complete handbook notebook missing narrative folio {name}",
)
digest = hashlib.sha256(source_path.read_bytes()).hexdigest()
require(
f'data-source-sha256="{digest}"' in complete_source,
f"complete handbook notebook has stale narrative folio {name}",
)
folio_count = sum(
"field-notebook-folio" in cell.get("metadata", {}).get("tags", [])
for cell in complete_cells
)
evidence_cells = [
cell
for cell in complete_cells
if "evidence-insert" in cell.get("metadata", {}).get("tags", [])
]
experiment_openers = [
cell
for cell in complete_cells
if "experiment-opener" in cell.get("metadata", {}).get("tags", [])
]
binding_cells = [
cell
for cell in complete_cells
if "binding-note" in cell.get("metadata", {}).get("tags", [])
]
complete_code_cells = [
cell for cell in complete_cells if cell.get("cell_type") == "code"
]
complete_output_cells = [cell for cell in complete_code_cells if cell.get("outputs")]
bench_notebooks = {
str(cell.get("metadata", {}).get("source_notebook"))
for cell in complete_cells
if "bench-insert" in cell.get("metadata", {}).get("tags", [])
and cell.get("metadata", {}).get("source_notebook")
}
evidence_paths = {
str(cell.get("metadata", {}).get("source_path")) for cell in evidence_cells
}
binding_paths = {
str(cell.get("metadata", {}).get("source_path")) for cell in binding_cells
}
expected_evidence_paths = {
path.relative_to(ROOT).as_posix() for path in REQUIRED_RESEARCH
}
expected_bench_notebooks = {path.name for path in REQUIRED_NOTEBOOKS[1:]}
expected_binding_paths = {
path.relative_to(ROOT).as_posix() for path in REQUIRED_SITE_SCRIPTS
}
longest_evidence_run = 0
current_evidence_run = 0
for cell in complete_cells:
if "evidence-insert" in cell.get("metadata", {}).get("tags", []):
current_evidence_run += 1
longest_evidence_run = max(longest_evidence_run, current_evidence_run)
else:
current_evidence_run = 0
metadata = complete_payload.get("metadata", {}).get("rag_evolution", {})
require(metadata.get("binding") == "single-manuscript", "complete handbook is not marked as one manuscript")
require(folio_count == 6, f"complete handbook has {folio_count} folio openers")
require(len(evidence_cells) == 18, f"complete handbook has {len(evidence_cells)} evidence leaves")
require(evidence_paths == expected_evidence_paths, "complete handbook evidence leaves are incomplete")
require(len(experiment_openers) == 8, f"complete handbook has {len(experiment_openers)} experiment openers")
require(bench_notebooks == expected_bench_notebooks, "complete handbook bench inserts are incomplete")
require(len(binding_cells) == 7, f"complete handbook has {len(binding_cells)} binding notes")
require(binding_paths == expected_binding_paths, "complete handbook binding notes are incomplete")
require(len(complete_code_cells) == 75, f"complete handbook has {len(complete_code_cells)} workbench sources")
require(len(complete_output_cells) == 75, f"complete handbook has {len(complete_output_cells)} attached observations")
require(longest_evidence_run <= 2, "complete handbook recreates a standalone evidence annex")
require(
not any(
"technical-atlas-divider" in cell.get("metadata", {}).get("tags", [])
for cell in complete_cells
),
"complete handbook retains a standalone atlas transition",
)
for old_surface in (
"PART TWO · THE TECHNICAL ATLAS",
"ATLAS · FULL EVIDENCE",
"# The executable workbench",
):
require(old_surface not in reader_source, f"complete handbook retains old surface: {old_surface}")
return code_cells, markdown_cells, output_cells, markdown_words
def validate_build_provenance() -> Mapping[str, object]:
"""Validate deterministic upstream logs without claiming post-render work."""
require(BUILD_PROVENANCE.is_file(), "missing notebook build provenance")
payload = json.loads(BUILD_PROVENANCE.read_text(encoding="utf-8"))
require(payload.get("schema_version") == 1, "unsupported build provenance schema")
steps = payload.get("steps")
require(isinstance(steps, list), "build provenance steps are not a list")
expected_steps = (
"scripts/build_chronological_index.py",
"scripts/build_curriculum_notebooks.py",
"scripts/execute_notebooks.py",
)
require(
tuple(step.get("script") for step in steps) == expected_steps,
"build provenance has the wrong upstream order",
)
for step in steps:
require(step.get("returncode") == 0, f"failed recorded build step: {step.get('script')}")
require(isinstance(step.get("arguments"), list), "build arguments are not stable JSON")
require(isinstance(step.get("stdout"), str), "build stdout is not text")
require(isinstance(step.get("stderr"), str), "build stderr is not text")
source_hashes = payload.get("source_sha256", {})
for relative in expected_steps:
target = ROOT / relative
require(
source_hashes.get(relative) == hashlib.sha256(target.read_bytes()).hexdigest(),
f"stale provenance source hash: {relative}",
)
artifact_hashes = payload.get("artifact_sha256", {})
expected_artifacts = (
ROOT / "research" / "chronological_index.md",
*REQUIRED_NOTEBOOKS,
)
for target in expected_artifacts:
relative = target.relative_to(ROOT).as_posix()
require(
artifact_hashes.get(relative) == hashlib.sha256(target.read_bytes()).hexdigest(),
f"stale provenance artifact hash: {relative}",
)
serialized = json.dumps(payload, sort_keys=True)
require(str(ROOT) not in serialized, "build provenance contains an absolute repo path")
forbidden = ("timestamp", "duration", "pid", "cwd", "interpreter", "preview_sha")
require(
not any(f'"{field}"' in serialized.lower() for field in forbidden),
"build provenance contains nondeterministic fields",
)
return payload
def validate_system_design_catalog() -> Tuple[Mapping[str, object], Tuple[Mapping[str, object], ...]]:
"""Validate the source catalog for the dedicated system-design collection."""
def text(value: object) -> bool:
return isinstance(value, str) and bool(value.strip()) and not set("<>") & set(value)
require(SYSTEM_DESIGNS.is_file(), "missing system-design catalog")
payload = json.loads(SYSTEM_DESIGNS.read_text(encoding="utf-8"))
require(
isinstance(payload, dict) and set(payload) == {"schema_version", "families"}
and payload.get("schema_version") == 1,
"invalid system catalog envelope",
)
families = payload["families"]
require(isinstance(families, list) and len(families) == 12, "expected twelve system families")
slugs = set()
for family in families:
require(
isinstance(family, dict) and set(family) == {
"slug", "title", "category", "summary", "systems", "lanes", "reading"
}, "invalid system family fields",
)
slug = family["slug"]
require(
text(slug) and re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", str(slug))
and slug not in slugs, f"bad system slug: {slug!r}",
)
slugs.add(slug)
require(all(text(family[key]) for key in ("title", "category", "summary")), f"{slug}: bad copy")
systems = family["systems"]
require(
isinstance(systems, list) and systems and all(text(name) for name in systems)
and len(systems) == len(set(systems)), f"{slug}: bad systems",
)
lanes = family["lanes"]
require(isinstance(lanes, list) and 2 <= len(lanes) <= 5, f"{slug}: expected 2-5 lanes")
lane_labels = set()
assigned: List[str] = []
for lane in lanes:
require(
isinstance(lane, dict) and set(lane) in (
{"label", "steps", "systems"}, {"label", "steps", "systems", "feedback"}
) and text(lane.get("label")) and lane["label"] not in lane_labels,
f"{slug}: bad lane",
)
lane_labels.add(lane["label"])
steps = lane["steps"]
require(
isinstance(steps, list) and 3 <= len(steps) <= 6
and all(
isinstance(step, dict) and set(step) == {"role", "label"}
and text(step["role"]) and text(step["label"])
for step in steps
), f"{slug}/{lane['label']}: bad flow",
)
lane_systems = lane["systems"]
require(
isinstance(lane_systems, list) and lane_systems and all(text(name) for name in lane_systems)
and len(lane_systems) == len(set(lane_systems))
and ("feedback" not in lane or text(lane["feedback"])),
f"{slug}/{lane['label']}: bad assignments",
)
assigned.extend(lane_systems)
require(Counter(assigned) == Counter(systems), f"{slug}: systems need one lane each")
reading = family["reading"]
require(
isinstance(reading, list) and 1 <= len(reading) <= 3
and all(
isinstance(item, dict) and set(item) == {"label", "href"} and text(item["label"])
and isinstance(item["href"], str)
and item["href"].startswith(("/read/", "/research/"))
for item in reading
)
and len({item["href"] for item in reading}) == len(reading), f"{slug}: bad reading links",
)
require(str(ROOT) not in json.dumps(payload), "absolute path in system catalog")
return payload, tuple(families)
def publication_slug(path: Path) -> str:
"""Return the renderer's stable public slug for Markdown and notebooks."""
if path.name == "README.md":
return "method-and-scope"
stem = re.sub(r"^\d+_", "", path.stem)
return stem.replace("_", "-")
def python_source_slug(path: Path) -> str:
"""Return the public slug used for one standalone Python source page."""
relative = path.relative_to(ROOT)
if relative.parts[:2] == ("src", "rag_evolution"):
stem = "package" if path.stem == "__init__" else path.stem
return f"rag-evolution-{stem.replace('_', '-')}"
return path.stem.replace("_", "-")
def public_python_sources() -> Tuple[Path, ...]:
"""Enumerate every first-party Python file exposed by the publication."""
package_sources = tuple(sorted((ROOT / "src" / "rag_evolution").glob("*.py")))
sources = (*REQUIRED_SITE_SCRIPTS, *package_sources)
require(all(path.is_file() for path in sources), "public Python source set is incomplete")
require(len(sources) == len(set(sources)), "public Python source set contains duplicates")
return sources
def meta_content(audit: PreviewHTMLAudit, name: str, page: Path) -> str:
"""Read one required named meta value from an audited HTML page."""
values = [
str(attributes.get("content", ""))
for tag, attributes in audit.elements
if tag == "meta" and attributes.get("name") == name
]
require(
len(values) == 1 and bool(values[0]),
f"{page.relative_to(ROOT)} must contain one {name} meta value",
)
return values[0]
def embedded_json(document: str, identifier: str, page: Path) -> Mapping[str, object]:
"""Parse one embedded JSON script without accepting ambiguous duplicates."""
matches = re.findall(
rf'<script id="{re.escape(identifier)}" type="application/json">(.*?)</script>',
document,
flags=re.DOTALL,
)
require(
len(matches) == 1,
f"{page.relative_to(ROOT)} must contain one {identifier} payload",
)
payload = json.loads(matches[0])
require(isinstance(payload, dict), f"{identifier} is not a JSON object in {page.name}")
return payload
def audit_publication_page(
route: str,
path: Path,
) -> Tuple[str, PreviewHTMLAudit]:
"""Validate one page's basic HTML structure, IDs, and canonical route."""
require(path.is_file(), f"missing publication page: {path.relative_to(ROOT)}")
document = path.read_text(encoding="utf-8")
require(
document.lstrip().lower().startswith("<!doctype html>"),
f"publication page has no HTML doctype: {path.relative_to(ROOT)}",
)
audit = PreviewHTMLAudit()
audit.feed(document)
audit.close()
if audit.stack:
audit.errors.append(f"unclosed tags: {', '.join(audit.stack[-8:])}")
html_error = audit.errors[0] if audit.errors else "unknown structural error"
require(
not audit.errors,
f"invalid publication HTML in {path.relative_to(ROOT)}: {html_error}",
)
tag_counts = Counter(tag for tag, _ in audit.elements)
for tag in ("html", "head", "body", "main"):
require(
tag_counts[tag] == 1,
f"{path.relative_to(ROOT)} has {tag_counts[tag]} {tag} elements",
)
duplicate_ids = sorted(
identifier
for identifier, count in Counter(audit.identifiers).items()
if count > 1
)
require(
not duplicate_ids,
f"duplicate IDs in {path.relative_to(ROOT)}: {duplicate_ids[:5]}",
)
invalid_ids = sorted(
identifier for identifier in audit.identifiers if re.search(r"\s", identifier)
)
require(
not invalid_ids,
f"whitespace in IDs in {path.relative_to(ROOT)}: {invalid_ids[:5]}",
)
canonical_links = [
str(attributes.get("href", ""))
for tag, attributes in audit.elements
if tag == "link"
and "canonical" in str(attributes.get("rel", "")).split()
]
expected_canonical = PUBLIC_SITE_ORIGIN + ("/" if route == "/" else route)
require(
canonical_links == [expected_canonical],
f"wrong canonical URL in {path.relative_to(ROOT)}: {canonical_links}",
)
return document, audit
def validate_page_set(directory: Path, expected: Sequence[str], label: str) -> None:
"""Reject missing and stale generated HTML files in one publication group."""
actual = {path.name for path in directory.glob("*.html")}
required = set(expected)
require(
actual == required,
f"{label} page set differs: missing={sorted(required - actual)}, "
f"extra={sorted(actual - required)}",
)
def validate_page_budget(path: Path, maximum: int, label: str) -> int:
"""Enforce a byte budget and return the measured size."""
size = path.stat().st_size
require(
size <= maximum,
f"{label} exceeds its HTML budget: {size:,} > {maximum:,} bytes "
f"({path.relative_to(ROOT)})",
)
return size
def validate_system_design_page(
family: Mapping[str, object],
page: Path,
document: str,
audit: PreviewHTMLAudit,
) -> None:
"""Validate one semantic system-flow page against its catalog family."""
def one(pattern: str, source: str, label: str) -> str:
matches = re.findall(pattern, source, re.DOTALL)
require(len(matches) == 1, f"{page.name}: expected one {label}")
return matches[0]
slug = str(family["slug"])
lanes = family["lanes"]
figures = [
attrs for tag, attrs in audit.elements
if tag == "figure" and "system-design" in str(attrs.get("class", "")).split()
]
caption_id = f"{slug}-diagram-caption"
require(
len(figures) == 1 and figures[0].get("data-system-diagram") == slug
and figures[0].get("data-lane-count") == str(len(lanes))
and figures[0].get("aria-labelledby") == caption_id
and sum(tag == "figcaption" and attrs.get("id") == caption_id for tag, attrs in audit.elements) == 1,
f"{page.name}: bad semantic figure",
)
figure = one(r'<figure class="system-design"[^>]*>(.*?)</figure>', document, "diagram")
require(f'<figcaption id="{caption_id}">' in figure, f"{page.name}: detached caption")
lane_elements = [
attrs for tag, attrs in audit.elements
if tag == "section" and "system-lane" in str(attrs.get("class", "")).split()
]
lane_bodies = re.findall(r'<section class="system-lane"[^>]*>(.*?)</section>', figure, re.DOTALL)
require(len(lane_elements) == len(lanes) == len(lane_bodies), f"{page.name}: wrong lanes")
rendered_systems: List[str] = []
for index, (lane, attrs, body) in enumerate(zip(lanes, lane_elements, lane_bodies), start=1):
label = str(lane["label"])
lane_id = f"{slug}-lane-{index}"
require(
attrs.get("aria-labelledby") == lane_id
and attrs.get("data-system-variant") == label
and re.search(rf'<h2 id="{re.escape(lane_id)}">{re.escape(html.escape(label))}</h2>', body),
f"{page.name}: bad lane {index}",
)
flow = one(r'<ol class="system-flow"[^>]*>(.*?)</ol>', body, f"flow {index}")
actual_steps = re.findall(
r'<li><span class="system-step-role">(.*?)</span><strong>(.*?)</strong></li>',
flow, re.DOTALL,
)
expected_steps = [
(html.escape(str(step["role"])), html.escape(str(step["label"])))
for step in lane["steps"]
]
require(3 <= len(actual_steps) <= 6 and actual_steps == expected_steps, f"{page.name}: changed flow {index}")
system_list = one(r'<ul class="system-lane-systems"[^>]*>(.*?)</ul>', body, f"systems {index}")
lane_systems = [
html.unescape(item) for item in re.findall(r"<li>([^<]*)</li>", system_list)
]
require(lane_systems == lane["systems"], f"{page.name}: changed systems {index}")
rendered_systems.extend(lane_systems)
require(Counter(rendered_systems) == Counter(family["systems"]), f"{page.name}: systems need one lane each")
require(
"system-inventory" not in document and "system-name-list" not in document,
f"{page.name}: detached system inventory",
)
reading = one(r'<section class="system-reading"[^>]*>(.*?)</section>', document, "reading section")
rendered_reading = re.findall(
r'<li><a href="([^"]+)">(.*?)</a></li>', reading, re.DOTALL,
)
require(
rendered_reading == [(item["href"], html.escape(item["label"])) for item in family["reading"]],
f"{page.name}: changed reading links",
)
def validate_publication_links(
route_pages: Mapping[str, Path],
audits: Mapping[str, PreviewHTMLAudit],
) -> int:
"""Resolve local hrefs exactly as a browser would from each public route."""
local_links = 0
public_host = urlparse(PUBLIC_SITE_ORIGIN).netloc
for route, audit in audits.items():
base = PUBLIC_SITE_ORIGIN + ("/" if route == "/" else route)
for href in audit.hrefs:
raw = href.strip()
require(bool(raw), f"empty href in {route_pages[route].relative_to(ROOT)}")
parsed_raw = urlparse(raw)
require(
parsed_raw.scheme not in {"javascript", "data"},
f"unsafe href in {route_pages[route].relative_to(ROOT)}: {href}",
)
resolved = urlparse(urljoin(base, raw))
if resolved.netloc and resolved.netloc != public_host:
continue
if resolved.scheme and resolved.scheme not in {"http", "https"}:
continue
local_links += 1
target_route = unquote(resolved.path).rstrip("/") or "/"
if target_route in route_pages:
if resolved.fragment:
require(
resolved.fragment in set(audits[target_route].identifiers),
f"missing target {target_route}#{resolved.fragment} from {route}",
)
continue
if target_route == "/book":
require(BOOK_PDF.is_file(), "publication links a missing reader PDF")
continue
target = (ROOT / target_route.lstrip("/")).resolve()
require(
target.is_relative_to(ROOT),
f"publication href escapes the repository: {href}",
)
require(
target.exists(),
f"broken local href in {route_pages[route].relative_to(ROOT)}: {href}",
)
return local_links
def validate_publication() -> Mapping[str, int]:
"""Prove the compact multi-page site is complete, faithful, and deterministic."""
provenance = validate_build_provenance()
_, system_families = validate_system_design_catalog()
require(PUBLICATION_INDEX.is_file(), "missing publication index")
require(SITE_MANIFEST.is_file(), "missing publication site manifest")
require(not RETIRED_MONOLITH.exists(), "retired 1-page preview still bloats publication")
manifest = json.loads(SITE_MANIFEST.read_text(encoding="utf-8"))
require(isinstance(manifest, dict), "site manifest is not a JSON object")
require(manifest.get("schema_version") == 3, "unsupported site manifest schema")
require(
set(manifest) == {
"schema_version",
"complete_notebook",
"reader",
"systems",
"research",
"notebooks",
"python",
},
"site manifest has missing or unreviewed fields",
)
reader_manifest = {
path.relative_to(ROOT).as_posix(): f"/read/{slug}"
for path, slug in zip(REQUIRED_FIELD_NOTEBOOK, READER_ROUTE_SLUGS)
}
research_paths = tuple(ROOT / "research" / name for name in RESEARCH_PAGE_ORDER)
research_manifest = {
path.relative_to(ROOT).as_posix(): f"/research/{publication_slug(path)}"
for path in research_paths
}
focused_notebooks = tuple(ROOT / "notebooks" / name for name in FOCUSED_NOTEBOOK_NAMES)
notebook_manifest = {
path.relative_to(ROOT).as_posix(): f"/notebooks/view/{publication_slug(path)}"
for path in focused_notebooks
}
python_sources = public_python_sources()
python_manifest = {
path.relative_to(ROOT).as_posix(): f"source/{python_source_slug(path)}.html"
for path in python_sources
}
system_routes = {
str(family["slug"]): f'/systems/{family["slug"]}'
for family in system_families
}
system_manifest = {
"source": SYSTEM_DESIGNS.relative_to(ROOT).as_posix(),
"sha256": hashlib.sha256(SYSTEM_DESIGNS.read_bytes()).hexdigest(),
"routes": system_routes,
}
require(manifest.get("reader") == reader_manifest, "site manifest has wrong reader order")
require(
isinstance(manifest.get("systems"), dict)
and set(manifest["systems"]) == {"source", "sha256", "routes"}
and manifest["systems"] == system_manifest,
"site manifest has stale or invalid system-design provenance",
)
require(manifest.get("research") == research_manifest, "site manifest has wrong research order")
require(manifest.get("notebooks") == notebook_manifest, "site manifest has wrong notebook order")
require(manifest.get("python") == python_manifest, "site manifest has wrong Python source set")
complete_payload = json.loads(REQUIRED_NOTEBOOKS[0].read_text(encoding="utf-8"))
expected_complete = {
"path": REQUIRED_NOTEBOOKS[0].relative_to(ROOT).as_posix(),
"sha256": hashlib.sha256(REQUIRED_NOTEBOOKS[0].read_bytes()).hexdigest(),
"cells": len(complete_payload.get("cells", [])),
}
require(
manifest.get("complete_notebook") == expected_complete,
"site manifest has stale complete-notebook provenance",
)
serialized_manifest = json.dumps(manifest, sort_keys=True)
require(str(ROOT) not in serialized_manifest, "site manifest contains an absolute path")
require(
not any(field in serialized_manifest.lower() for field in ('"timestamp"', '"duration"', '"pid"', '"cwd"')),
"site manifest contains nondeterministic fields",
)
reader_pages = {
route: PREVIEW_DIRECTORY / "read" / f"{route.rsplit('/', 1)[-1]}.html"
for route in reader_manifest.values()
}
research_pages = {
route: PREVIEW_DIRECTORY / "research" / f"{route.rsplit('/', 1)[-1]}.html"
for route in research_manifest.values()
}
notebook_pages = {
route: PREVIEW_DIRECTORY / "notebooks" / f"{route.rsplit('/', 1)[-1]}.html"
for route in notebook_manifest.values()
}
python_pages = {
f"/python/{Path(relative_page).stem}": PREVIEW_DIRECTORY / relative_page
for relative_page in python_manifest.values()
}
system_pages = {
route: PREVIEW_DIRECTORY / "systems" / f"{slug}.html"
for slug, route in system_routes.items()
}
route_pages: Dict[str, Path] = {
"/": PUBLICATION_INDEX,
"/systems": PREVIEW_DIRECTORY / "systems" / "index.html",
"/research": PREVIEW_DIRECTORY / "research" / "index.html",
"/notebooks": PREVIEW_DIRECTORY / "notebooks" / "index.html",
"/python": PREVIEW_DIRECTORY / "source" / "index.html",
**reader_pages,
**system_pages,
**research_pages,
**notebook_pages,
**python_pages,
}
require(len(reader_pages) == 6, "publication does not have six reader pages")
require(len(system_pages) == 12, "publication does not have twelve system-design pages")
require(len(research_pages) == 18, "publication does not have eighteen research pages")
require(len(notebook_pages) == 8, "publication does not have eight notebook pages")
require(
len(python_pages) == len(python_sources),
"publication does not have one page per public Python source",
)
require(
len(route_pages) == len(set(route_pages.values())),
"multiple public routes point at the same generated HTML page",
)
validate_page_set(PREVIEW_DIRECTORY / "read", [path.name for path in reader_pages.values()], "reader")
validate_page_set(
PREVIEW_DIRECTORY / "systems",
["index.html", *(path.name for path in system_pages.values())],
"system design",
)
validate_page_set(
PREVIEW_DIRECTORY / "research",
["index.html", *(path.name for path in research_pages.values())],
"research",
)
validate_page_set(
PREVIEW_DIRECTORY / "notebooks",
["index.html", *(path.name for path in notebook_pages.values())],
"notebook",
)
validate_page_set(
PREVIEW_DIRECTORY / "source",
["index.html", *(path.name for path in python_pages.values())],
"Python source",
)
documents: Dict[str, str] = {}
audits: Dict[str, PreviewHTMLAudit] = {}
for route, page in route_pages.items():
document, audit = audit_publication_page(route, page)
documents[route] = document
audits[route] = audit
root_audit = audits["/"]
require(elements_with_class(root_audit, "home-page") == 1, "root is not a compact publication index")
require(
not any(attributes.get("data-code-cell") == "true" for _, attributes in root_audit.elements),
"root index embeds notebook cells",
)
require(
set(reader_pages).issubset(root_audit.hrefs),
"root index does not expose the complete six-part reading order",
)
for collection_route in ("/systems", "/research", "/notebooks", "/python", "/book"):
require(collection_route in root_audit.hrefs, f"root index omits {collection_route}")
system_index = documents["/systems"]
system_index_entries = re.findall(
r'<ol class="publication-list">(.*?)</ol>',
system_index,
flags=re.DOTALL,
)
require(
len(system_index_entries) == 1
and system_index_entries[0].count("<li>") == 12,
"system-design index must contain exactly twelve family entries",
)
for route in system_routes.values():
require(
system_index_entries[0].count(f'href="{route}"') == 1,
f"system-design index does not link {route} exactly once",
)
catalog_digest = hashlib.sha256(SYSTEM_DESIGNS.read_bytes()).hexdigest()
for family in system_families:
slug = str(family["slug"])
route = system_routes[slug]
page = system_pages[route]
require(
meta_content(audits[route], "source-sha256", page) == catalog_digest,
f"stale system-design catalog hash in {page.name}",
)
validate_system_design_page(family, page, documents[route], audits[route])
source_page_by_source = {
source: python_pages[f"/python/{python_source_slug(source)}"]
for source in python_sources
}
for source, page in source_page_by_source.items():
route = f"/python/{python_source_slug(source)}"
audit = audits[route]
source_elements = [
attributes
for tag, attributes in audit.elements
if tag == "code" and attributes.get("data-source-path")
]
relative = source.relative_to(ROOT).as_posix()
digest = hashlib.sha256(source.read_bytes()).hexdigest()
require(
len(source_elements) == 1,
f"{page.relative_to(ROOT)} does not contain one source block",
)
attributes = source_elements[0]
require(attributes.get("data-source-path") == relative, f"wrong source path in {page.name}")
require(attributes.get("data-source-sha256") == digest, f"stale source hash in {page.name}")
require(attributes.get("data-highlighted-python") == "true", f"unhighlighted source in {page.name}")
match = re.search(
rf'<code(?=[^>]*data-source-path="{re.escape(relative)}")[^>]*>(.*?)</code>',
documents[route],
flags=re.DOTALL,
)
require(match is not None, f"missing rendered source body in {page.name}")
rendered_source = html.unescape(re.sub(r"<[^>]+>", "", match.group(1)))
source_text = source.read_text(encoding="utf-8")
require(rendered_source == source_text, f"rendered Python differs from {relative}")
source_size = len(source.read_bytes())
require(
page.stat().st_size <= source_size * 8 + 16 * 1024,
f"highlighted source expansion is excessive: {page.relative_to(ROOT)}",
)
aggregate_code = 0
aggregate_outputs = 0
aggregate_highlighted = 0
for path in focused_notebooks:
route = notebook_manifest[path.relative_to(ROOT).as_posix()]
page = notebook_pages[route]
payload = json.loads(path.read_text(encoding="utf-8"))
cells = payload.get("cells", [])
code = [cell for cell in cells if cell.get("cell_type") == "code"]
expected_stats = {
"notebooks": 1,
"cells": len(cells),
"code_cells": len(code),
"executed_cells": sum(cell.get("execution_count") is not None for cell in code),
"output_cells": sum(bool(cell.get("outputs")) for cell in code),
}
digest = hashlib.sha256(path.read_bytes()).hexdigest()
require(meta_content(audits[route], "source-sha256", page) == digest, f"stale notebook hash in {page.name}")
page_manifest = embedded_json(documents[route], "artifact-manifest", page)
require(
page_manifest == {"notebook": path.name, "sha256": digest, "stats": expected_stats},
f"stale notebook manifest in {page.name}",
)
rendered_code = sum(
attributes.get("data-code-cell") == "true"
for _, attributes in audits[route].elements
)
highlighted = sum(
attributes.get("data-highlighted-python") == "true"
for _, attributes in audits[route].elements
)
require(rendered_code == len(code), f"{page.name} omits notebook code cells")
require(audits[route].workbench_notes == len(code), f"{page.name} has detached code cells")
require(
audits[route].workbench_notes_with_output == expected_stats["output_cells"],
f"{page.name} has detached saved outputs",
)
require(highlighted == len(code), f"{page.name} has unhighlighted notebook code")
require("Execution error" not in documents[route], f"{page.name} renders an execution error")
aggregate_code += rendered_code
aggregate_outputs += audits[route].workbench_notes_with_output
aggregate_highlighted += highlighted
require(aggregate_code == 75, f"publication renders {aggregate_code}/75 notebook code cells")
require(aggregate_outputs == 75, f"publication renders {aggregate_outputs}/75 notebook outputs")
require(aggregate_highlighted == 75, f"publication highlights {aggregate_highlighted}/75 notebook cells")
for source_path, route in zip(REQUIRED_FIELD_NOTEBOOK, reader_manifest.values()):
page = reader_pages[route]
require(
meta_content(audits[route], "source-sha256", page)
== hashlib.sha256(source_path.read_bytes()).hexdigest(),
f"stale reader source hash in {page.name}",
)
for source_path in research_paths:
route = research_manifest[source_path.relative_to(ROOT).as_posix()]
page = research_pages[route]
require(
meta_content(audits[route], "source-sha256", page)
== hashlib.sha256(source_path.read_bytes()).hexdigest(),
f"stale research source hash in {page.name}",
)
math_routes = (*reader_pages, *research_pages, *notebook_pages)
math_documents = [documents[route] for route in math_routes]
math_audits = [audits[route] for route in math_routes]
math_counts = {
"inline": sum(elements_with_class(audit, "math-inline") for audit in math_audits),
"display": sum(elements_with_class(audit, "math-display") for audit in math_audits),
"round": sum(document.count('data-math-delimiter="\\("') for document in math_documents),
"dollar": sum(document.count('data-math-delimiter="$"') for document in math_documents),
"bracket": sum(document.count('data-math-delimiter="\\["') for document in math_documents),
"double_dollar": sum(document.count('data-math-delimiter="$$"') for document in math_documents),
}
require(math_counts["inline"] == 273, f"publication has {math_counts['inline']}/273 inline equations")
require(math_counts["display"] == 160, f"publication has {math_counts['display']}/160 display equations")
require(math_counts["round"] == 267, "publication lost round-delimited TeX provenance")
require(math_counts["dollar"] == 6, "publication lost dollar-delimited TeX provenance")
require(math_counts["bracket"] == 158, "publication lost bracket-delimited TeX provenance")
require(math_counts["double_dollar"] == 2, "publication lost double-dollar TeX provenance")
for route in math_routes:
document = documents[route]
require(
'<script id="mathjax-runtime" defer' in document
and "https://cdn.jsdelivr.net/npm/mathjax@4.1.3/tex-chtml.js" in document,
f"{route} does not load pinned MathJax 4",
)
require("displayOverflow: 'linebreak'" in document, f"{route} disables MathJax line breaking")
require(
re.search(r'class="equation-note math-display"[^>]*>\s*<pre', document) is None,
f"{route} places a display equation inside pre",
)
local_links = validate_publication_links(route_pages, audits)
sizes = [validate_page_budget(PUBLICATION_INDEX, ROOT_PAGE_BUDGET, "root index")]
sizes.append(
validate_page_budget(
route_pages["/systems"],
SYSTEM_INDEX_BUDGET,
"system-design index",
)
)
for route in ("/research", "/notebooks", "/python"):
sizes.append(validate_page_budget(route_pages[route], COLLECTION_PAGE_BUDGET, "collection index"))
sizes.extend(
validate_page_budget(path, SYSTEM_PAGE_BUDGET, "system-design page")
for path in system_pages.values()
)
sizes.extend(validate_page_budget(path, READER_PAGE_BUDGET, "reader page") for path in reader_pages.values())
sizes.extend(validate_page_budget(path, RESEARCH_PAGE_BUDGET, "research page") for path in research_pages.values())
sizes.extend(validate_page_budget(path, NOTEBOOK_PAGE_BUDGET, "notebook page") for path in notebook_pages.values())
sizes.extend(validate_page_budget(path, PYTHON_PAGE_BUDGET, "Python source page") for path in python_pages.values())
total_size = sum(sizes)
require(
total_size <= PUBLICATION_BUDGET,
f"publication exceeds aggregate HTML budget: {total_size:,} > {PUBLICATION_BUDGET:,} bytes",
)
require(len(provenance.get("steps", [])) == 3, "wrong recorded build-step count")
sys.path.insert(0, str(ROOT))
from scripts.render_field_notebook_preview import render # pylint: disable=import-outside-toplevel
expected_files = {
path.relative_to(PREVIEW_DIRECTORY).as_posix()
for path in route_pages.values()
} | {SITE_MANIFEST.name}
with TemporaryDirectory(prefix="rag-publication-") as directory:
fresh_root = Path(directory)
render(REQUIRED_NOTEBOOKS[0], fresh_root / "index.html")
fresh_files = {
path.relative_to(fresh_root).as_posix()
for path in fresh_root.rglob("*.html")
} | {"site_manifest.json"}
require(fresh_files == expected_files, "fresh renderer emits the wrong publication file set")
for relative in sorted(expected_files):
require(
(fresh_root / relative).read_bytes()
== (PREVIEW_DIRECTORY / relative).read_bytes(),
f"generated publication differs from a fresh render: {relative}",
)
unique_ids = sum(len(set(audit.identifiers)) for audit in audits.values())
return {
"pages": len(route_pages),
"reader_pages": len(reader_pages),
"system_pages": len(system_pages),
"research_pages": len(research_pages),
"notebook_pages": len(notebook_pages),
"python_pages": len(python_pages),
"code_cells": aggregate_code,
"outputs": aggregate_outputs,
"math": math_counts["inline"] + math_counts["display"],
"ids": unique_ids,
"links": local_links,
"bytes": total_size,
}
def validate_reference_implementation() -> Tuple[int, int]:
package = ROOT / "src" / "rag_evolution"
for name in REQUIRED_MODULES:
require((package / name).is_file(), f"missing reference module: {name}")
test_files = sorted((ROOT / "tests").glob("test_*.py"))
test_methods = 0
for path in test_files:
tree = ast.parse(path.read_text(encoding="utf-8"))
test_methods += sum(
isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name.startswith("test_")
for node in ast.walk(tree)
)
require(test_methods >= 45, f"too few reference tests: {test_methods}")
return len(REQUIRED_MODULES), test_methods
def validate_demo_sources() -> int:
sys.path.insert(0, str(ROOT / "src"))
from rag_evolution.demo_data import demo_documents # pylint: disable=import-outside-toplevel
documents = demo_documents()
require(len(documents) >= 12, "teaching corpus is unexpectedly small")
for document in documents:
parsed = urlparse(document.source)
require(parsed.scheme == "https" and parsed.netloc, f"invalid demo source: {document.id}")
require(bool(document.date), f"missing demo date: {document.id}")
require(bool(document.metadata.get("entities")), f"missing graph entities: {document.id}")
return len(documents)
def main() -> int:
try:
source_count, status_counts, topic_counts = validate_sources()
word_count, link_count = validate_markdown()
chronology_rows, coverage_rows, glossary_terms = validate_research_structure()
field_words, margin_notes, experiments = validate_field_notebook()
code_cells, markdown_cells, output_cells, notebook_words = validate_notebooks()
publication = validate_publication()
module_count, test_count = validate_reference_implementation()
document_count = validate_demo_sources()
except (ValidationError, json.JSONDecodeError) as error:
print(f"FAIL: {error}", file=sys.stderr)
return 1
print(f"PASS sources: {source_count} ({dict(status_counts)})")
print(f"PASS source topics: {len(topic_counts)} distinct tags")
print(f"PASS research: {word_count} words, {link_count} distinct external links")
print(
"PASS coverage: "
f"{chronology_rows} chronological works, {coverage_rows} matrix rows, "
f"{glossary_terms} glossary terms"
)
print(
"PASS field notebook: "
f"{field_words} words, {margin_notes} margin notes, {experiments} experiments"
)
print(
"PASS notebooks: "
f"{code_cells} executed code cells, {markdown_cells} Markdown cells, "
f"{output_cells} cells with saved output, {notebook_words} Markdown words"
)
print(
"PASS publication: "
f"{publication['pages']} pages "
f"({publication['reader_pages']} reader, {publication['system_pages']} systems, "
f"{publication['research_pages']} research, "
f"{publication['notebook_pages']} notebooks, {publication['python_pages']} Python), "
f"{publication['code_cells']} code cells, {publication['outputs']} outputs, "
f"{publication['math']} equations, {publication['ids']} unique page-local ids, "
f"{publication['links']} local links, {publication['bytes']:,} bytes, exact render"
)
print(f"PASS implementation: {module_count} required modules, {test_count} test methods")
print(f"PASS teaching corpus: {document_count} dated source-linked documents")
return 0
if __name__ == "__main__":
raise SystemExit(main())