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_IMAGE_DIRECTORY = ROOT / "assets" / "system-images"
RETIRED_SYSTEM_DIRECTORY = PREVIEW_DIRECTORY / "systems"
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",
)
SYSTEM_IMAGE_OWNERSHIP = {
"/read/search": (
"structured-visual-and-code-rag",
"corpus-units-and-ann",
"retrieval-cascade",
),
"/read/grounding": (
"retrieve-read-and-rag",
"retrieval-pretraining-and-memory",
"query-context-and-verification",
),
"/read/agents-memory-security": (
"adaptive-and-corrective-control",
"agentic-search-policies",
"graph-and-hierarchical-rag",
"persistent-memory-and-time",
"privacy-and-robust-retrieval",
),
"/read/evaluation-production": ("production-and-optimization",),
}
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
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
SYSTEM_IMAGE_BUDGET = 64 * 1024
SYSTEM_IMAGE_TOTAL_BUDGET = 600 * 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 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 webp_dimensions(path: Path) -> Tuple[int, int]:
"""Read and validate a WebP canvas without optional image dependencies."""
data = path.read_bytes()
relative = path.relative_to(ROOT)
require(
len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP",
f"invalid RIFF/WEBP signature: {relative}",
)
require(
int.from_bytes(data[4:8], "little") + 8 == len(data),
f"invalid RIFF length: {relative}",
)
offset = 12
while offset + 8 <= len(data):
chunk_type = data[offset : offset + 4]
chunk_size = int.from_bytes(data[offset + 4 : offset + 8], "little")
start = offset + 8
end = start + chunk_size
require(end <= len(data), f"truncated WebP chunk: {relative}")
chunk = data[start:end]
if chunk_type == b"VP8 ":
require(
len(chunk) >= 10 and chunk[3:6] == b"\x9d\x01\x2a",
f"invalid lossy WebP frame: {relative}",
)
return (
int.from_bytes(chunk[6:8], "little") & 0x3FFF,
int.from_bytes(chunk[8:10], "little") & 0x3FFF,
)
if chunk_type == b"VP8L":
require(
len(chunk) >= 5 and chunk[0] == 0x2F,
f"invalid lossless WebP frame: {relative}",
)
bits = int.from_bytes(chunk[1:5], "little")
return (bits & 0x3FFF) + 1, ((bits >> 14) & 0x3FFF) + 1
if chunk_type == b"VP8X":
require(len(chunk) >= 10, f"invalid extended WebP frame: {relative}")
return (
int.from_bytes(chunk[4:7], "little") + 1,
int.from_bytes(chunk[7:10], "little") + 1,
)
offset = end + (chunk_size % 2)
raise ValidationError(f"WebP has no image dimensions: {relative}")
def validate_system_image_assets() -> Tuple[Mapping[str, Tuple[int, int]], int]:
"""Bind the exact twelve generated WebP files to dimensions and budgets."""
expected_slugs = {
slug for slugs in SYSTEM_IMAGE_OWNERSHIP.values() for slug in slugs
}
require(len(expected_slugs) == 12, "system-image ownership map is not twelve unique slugs")
require(SYSTEM_IMAGE_DIRECTORY.is_dir(), "missing system-image asset directory")
assets = {path.name: path for path in SYSTEM_IMAGE_DIRECTORY.iterdir()}
expected_names = {f"{slug}.webp" for slug in expected_slugs}
require(
set(assets) == expected_names,
"system-image asset set differs: "
f"missing={sorted(expected_names - set(assets))}, "
f"extra={sorted(set(assets) - expected_names)}",
)
dimensions: Dict[str, Tuple[int, int]] = {}
total_bytes = 0
for slug in sorted(expected_slugs):
path = assets[f"{slug}.webp"]
require(path.is_file() and not path.is_symlink(), f"invalid system-image asset: {path.name}")
size = path.stat().st_size
require(
size <= SYSTEM_IMAGE_BUDGET,
f"system image exceeds 64 KiB: {path.name} ({size:,} bytes)",
)
width, height = webp_dimensions(path)
require(width == 1440 and height > 0, f"invalid system-image dimensions: {path.name}")
dimensions[slug] = (width, height)
total_bytes += size
require(
total_bytes <= SYSTEM_IMAGE_TOTAL_BUDGET,
f"system images exceed 600 KiB: {total_bytes:,} bytes",
)
return dimensions, total_bytes
def validate_system_image_figure(
route: str,
slug: str,
document: str,
dimensions: Tuple[int, int],
) -> str:
"""Validate one rendered semantic figure against its exact WebP asset."""
fragments = re.findall(
rf'(<figure class="sketch system-image" data-system-image="{re.escape(slug)}">.*?</figure>)',
document,
flags=re.DOTALL,
)
require(len(fragments) == 1, f"{route} must render {slug} exactly once")
fragment = fragments[0]
audit = PreviewHTMLAudit()
audit.feed(fragment)
audit.close()
require(not audit.stack and not audit.errors, f"invalid {slug} figure structure in {route}")
figures = [attrs for tag, attrs in audit.elements if tag == "figure"]
images = [attrs for tag, attrs in audit.elements if tag == "img"]
captions = [attrs for tag, attrs in audit.elements if tag == "figcaption"]
require(
len(figures) == len(images) == len(captions) == 1,
f"{slug} needs one figure, image, and caption in {route}",
)
require(
set(figures[0]) == {"class", "data-system-image"}
and set(str(figures[0].get("class", "")).split()) == {"sketch", "system-image"}
and figures[0].get("data-system-image") == slug,
f"invalid semantic figure attributes for {slug}",
)
image = images[0]
require(
set(image) == {"src", "alt", "width", "height", "loading", "decoding"},
f"unreviewed image attributes for {slug}",
)
expected_src = f"/assets/system-images/{slug}.webp"
width, height = dimensions
require(image.get("src") == expected_src, f"wrong local image source for {slug}")
require(
str(image.get("width", "")).isdigit()
and int(str(image["width"])) == width == 1440,
f"wrong numeric image width for {slug}",
)
require(
str(image.get("height", "")).isdigit()
and int(str(image["height"])) == height,
f"image height differs from WebP header for {slug}",
)
require(
image.get("loading") == "lazy" and image.get("decoding") == "async",
f"{slug} must use lazy loading and asynchronous decoding",
)
alt = str(image.get("alt", "")).strip()
require(bool(alt), f"empty alternative text for {slug}")
require(
set(captions[0]) == {"class"}
and captions[0].get("class") == "sketch-caption",
f"invalid caption attributes for {slug}",
)
caption = re.findall(
r'<figcaption class="sketch-caption">(.*?)</figcaption>',
fragment,
flags=re.DOTALL,
)
require(
len(caption) == 1
and bool(html.unescape(re.sub(r"<[^>]+>", "", caption[0])).strip()),
f"empty or ambiguous caption for {slug}",
)
return alt
def validate_inline_system_images(
reader_pages: Mapping[str, Path],
documents: Mapping[str, str],
audits: Mapping[str, PreviewHTMLAudit],
dimensions: Mapping[str, Tuple[int, int]],
) -> int:
"""Enforce exact slug ownership across the six generated reader routes."""
expected_all = [
slug for slugs in SYSTEM_IMAGE_OWNERSHIP.values() for slug in slugs
]
expected_sources = {
f"/assets/system-images/{slug}.webp": slug for slug in expected_all
}
source_occurrences: Dict[str, List[str]] = {
source: [] for source in expected_sources
}
for route, audit in audits.items():
for tag, attributes in audit.elements:
system_slug = attributes.get("data-system-image")
if system_slug is not None:
require(route in reader_pages, f"system image appears outside reader pages: {route}")
source = str(attributes.get("src", ""))
if source.startswith("/assets/system-images/"):
require(tag == "img", f"system-image asset used by non-image element in {route}")
require(source in expected_sources, f"unexpected system-image source in {route}: {source}")
source_occurrences[source].append(route)
observed: List[str] = []
alt_texts: List[str] = []
for route in reader_pages:
expected = SYSTEM_IMAGE_OWNERSHIP.get(route, ())
tagged = [
(tag, attrs)
for tag, attrs in audits[route].elements
if attrs.get("data-system-image") is not None
]
require(
all(tag == "figure" for tag, _ in tagged),
f"non-figure system-image owner in {route}",
)
actual = tuple(str(attrs.get("data-system-image")) for _, attrs in tagged)
require(actual == expected, f"wrong system-image ownership or order in {route}: {actual}")
require(
not any(attrs.get("data-system-diagram") for _, attrs in audits[route].elements),
f"stale inline SVG system diagram in {route}",
)
for slug in expected:
alt_texts.append(
validate_system_image_figure(route, slug, documents[route], dimensions[slug])
)
observed.append(slug)
require(observed == expected_all, "reader routes do not own the exact twelve system images")
require(
all(len(routes) == 1 for routes in source_occurrences.values()),
"system-image assets must each appear exactly once in the publication: "
f"{source_occurrences}",
)
require(len(alt_texts) == len(set(alt_texts)), "system images reuse alternative text")
return len(observed)
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_image_dimensions, system_image_bytes = validate_system_image_assets()
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")
require(
not RETIRED_SYSTEM_DIRECTORY.exists() and not RETIRED_SYSTEM_DIRECTORY.is_symlink(),
"retired /systems generated directory still exists",
)
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") == 2, "unsupported site manifest schema")
require(
set(manifest) == {
"schema_version",
"complete_notebook",
"reader",
"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
}
require(manifest.get("reader") == reader_manifest, "site manifest has wrong reader order")
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(
"systems" not in manifest and "/systems" not in serialized_manifest,
"site manifest retains the retired /systems collection or routes",
)
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()
}
route_pages: Dict[str, Path] = {
"/": PUBLICATION_INDEX,
"/research": PREVIEW_DIRECTORY / "research" / "index.html",
"/notebooks": PREVIEW_DIRECTORY / "notebooks" / "index.html",
"/python": PREVIEW_DIRECTORY / "source" / "index.html",
**reader_pages,
**research_pages,
**notebook_pages,
**python_pages,
}
require(len(reader_pages) == 6, "publication does not have six reader 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 / "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
system_images = validate_inline_system_images(
reader_pages,
documents,
audits,
system_image_dimensions,
)
stale_system_links = [
href
for audit in audits.values()
for href in audit.hrefs
if urlparse(href).path == "/systems"
or urlparse(href).path.startswith("/systems/")
]
require(not stale_system_links, "generated publication retains retired /systems links")
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 ("/research", "/notebooks", "/python", "/book"):
require(collection_route in root_audit.hrefs, f"root index omits {collection_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")]
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, 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_images": system_images,
"system_image_bytes": system_image_bytes,
"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['research_pages']} research, "
f"{publication['notebook_pages']} notebooks, {publication['python_pages']} Python), "
f"{publication['system_images']} inline system images "
f"({publication['system_image_bytes']:,} bytes), "
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())