Spaces:

defenestrate
/

asa

Sleeping

asa / scripts /lib /build_derived.py

ASA Space: squashed history (instructions and settings live in the private pack)

526950e 24 days ago

23.9 kB

"""Build pipeline orchestrator for derived/student-chat/.
Reads the index (filtered to student_access=true), loads each source doc,
extracts a concept profile, embeds via the injected Embedder (cache-aware),
and writes the full catalog artifact atomically.
Incremental: per-doc reuse if (source_hash, build_config_fingerprint) match
the prior manifest. --full-rebuild bypasses cache.
--check mode validates inputs and reports without mutating.
--dry-run extracts profiles + computes hashes but writes no artifacts.
"""
from __future__ import annotations
import hashlib
import json
import os
import shutil
import subprocess
import tarfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from scripts.lib.build_config import (
    build_config_dict, BUILD_CONFIG_SCHEMA_VERSION,
    CATALOG_SCHEMA_VERSION, MANIFEST_SCHEMA_VERSION,
)
from scripts.lib.concept_profile import assemble as assemble_profile, PROFILE_SCHEMA_VERSION
from scripts.lib.embedder import Embedder
from scripts.lib.embedding_cache import EmbeddingCache
from scripts.lib.lecture_chunker import chunk_lecture_jsonl, row_lookup_to_json
from scripts.lib.ranker import _cosine_similarity
from scripts.lib.syntopicon_reference_parser import parse_all_references
from scripts.lib.term_extraction import build_regex


_DERIVED_DIR = "derived/student-chat"
_INDEX_SCHEMA_VERSION = 1


@dataclass
class BuildStats:
    docs_processed: int
    docs_reused: int
    cache_hits: int
    api_calls: int


def _sha256_text(text: str) -> str:
    return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()


def _sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            h.update(chunk)
    return "sha256:" + h.hexdigest()


def _git_head(repo_root: Path) -> str:
    try:
        out = subprocess.run(
            ["git", "-C", str(repo_root), "rev-parse", "HEAD"],
            capture_output=True, text=True, check=True,
        )
        return out.stdout.strip()
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        raise RuntimeError(
            f"unable to determine source_commit: git rev-parse HEAD failed in {repo_root} "
            f"and index.json did not provide source_commit. The catalog contract requires a "
            f"real commit SHA matching ^[0-9a-f]{{7,40}}$; refusing to emit a placeholder."
        ) from e


_REDACTED_KINDS = frozenset({"lecture_problem", "lecture_turn"})


def _privacy_tier_for(kind: str) -> str:
    return "redacted_classroom_speech" if kind in _REDACTED_KINDS else "public_course_content"


def _doc_id_for(file_record: dict) -> str:
    """Stable doc_id derived from kind + path stem + short path hash.
    The path hash discriminator ensures uniqueness across same-stem files
    in different directories.
    """
    path = file_record.get("path", "")
    stem = Path(path).stem
    kind = file_record.get("kind", "")
    # 8-hex-char hash of the full path for collision resistance
    path_disc = hashlib.sha256(path.encode("utf-8")).hexdigest()[:8]
    return f"{kind}_{stem}_{path_disc}"


def _student_visible_files(index: dict) -> list[tuple[str, dict]]:
    """Yield (topic, file_record) for every student-visible record in the
    index — including syntopicon entries, which the index stores in their
    own per-topic array with a lighter shape than FileRecord."""
    out: list[tuple[str, dict]] = []
    for topic, record in index.get("topics", {}).items():
        for f in record.get("files", []):
            if not f.get("student_access", False):
                continue
            if f.get("kind") == "syntopicon_registry":
                continue
            out.append((topic, f))
        for entry in record.get("syntopicon_entries", []):
            # Normalize the SyntopiconEntry shape to a FileRecord so the
            # rest of the pipeline can treat it uniformly. Entries are
            # student-accessible by classification (course.yaml).
            normalized = {
                "path": entry["path"],
                "kind": "syntopicon_entry",
                "topic": topic,
                "sha256": entry.get("sha256", ""),
                "visibility": "student",
                "student_access": True,
                "source_system": "syntopicon",
                "release_status": "released",
            }
            out.append((topic, normalized))
    return out


def _load_prior_manifest(manifest_path: Path) -> dict:
    if not manifest_path.is_file():
        return {}
    return json.loads(manifest_path.read_text(encoding="utf-8")).get("docs", {})


def _atomic_replace_dir(staging: Path, target: Path) -> None:
    """Replace target/ with staging/ atomically (with backup-rename dance for Windows).
    Recovers from an orphaned .bak left by a prior crash: if target is missing
    but .bak exists, restores .bak to target first. The remaining window
    between (target -> bak) and (staging -> target) is small but still not
    truly atomic; in the worst case (kill mid-rename), next build sees a
    missing derived/ and treats all docs as new (correct but expensive).
    """
    target.parent.mkdir(parents=True, exist_ok=True)
    bak = target.with_name(target.name + ".bak")
    # Recovery: if target was lost in a prior crash but bak exists, restore it
    if not target.exists() and bak.exists():
        os.rename(bak, target)
    if bak.exists():
        shutil.rmtree(bak)
    if target.exists():
        os.rename(target, bak)
    os.rename(staging, target)
    if bak.exists():
        shutil.rmtree(bak)


def _build_chunk_to_referencing_cards(catalog_records: list[dict]) -> dict[str, list[dict]]:
    by_doc = {rec["doc_id"]: rec for rec in catalog_records}
    chunk_to_cards: dict[str, list[dict]] = {}
    for rec in catalog_records:
        if rec.get("kind") != "syntopicon_entry":
            continue
        card_id = rec["doc_id"]
        card_vec = rec.get("embedding_vector", [])
        for ref in rec.get("references", []) or []:
            chunk_id = ref.get("lecture_problem_doc_id")
            if not chunk_id or chunk_id not in by_doc:
                continue
            chunk_vec = by_doc[chunk_id].get("embedding_vector", [])
            sim = _cosine_similarity(card_vec, chunk_vec)
            chunk_to_cards.setdefault(chunk_id, []).append({
                "syntopicon_doc_id": card_id,
                "similarity_score": sim,
            })
    return {
        chunk_id: sorted(cards, key=lambda c: c["similarity_score"], reverse=True)[:5]
        for chunk_id, cards in chunk_to_cards.items()
    }


def _build_term_to_docs(catalog_records: list[dict]) -> dict[str, list[str]]:
    term_to_docs: dict[str, list[str]] = {}
    for rec in catalog_records:
        if rec.get("kind") == "lecture_turn":
            continue
        for slug in rec.get("concepts_present", []):
            term_to_docs.setdefault(slug, []).append(rec["doc_id"])
    return {k: sorted(set(v)) for k, v in sorted(term_to_docs.items())}


def _build_concept_graph(registry: dict) -> dict[str, dict]:
    concept_graph: dict[str, dict] = {}
    for slug, term in registry.get("terms", {}).items():
        depends_on = term.get("depends_on", []) or []
        concept_graph[slug] = {"depends_on": sorted(depends_on), "dependents": []}
    for slug, info in concept_graph.items():
        for dep in info["depends_on"]:
            if dep in concept_graph:
                concept_graph[dep]["dependents"].append(slug)
    for info in concept_graph.values():
        info["dependents"] = sorted(set(info["dependents"]))
    return concept_graph


def run(
    *,
    repo_root: Path,
    embedder: Embedder,
    full_rebuild: bool = False,
    dry_run: bool = False,
) -> BuildStats:
    """Run the build pipeline. Returns BuildStats."""
    registry_path = repo_root / "content" / "syntopicon-mgrecon780" / "_registry.json"
    topic_link_path = repo_root / "index" / "inputs" / "syntopicon-links" / "topic-to-entry.json"
    rules_path = repo_root / "config" / "course.yaml"
    index_path = repo_root / "index" / "index.json"

    if not registry_path.is_file():
        raise FileNotFoundError(f"missing slim registry at {registry_path}")
    if not topic_link_path.is_file():
        raise FileNotFoundError(f"missing topic-to-entry map at {topic_link_path}")
    if not rules_path.is_file():
        raise FileNotFoundError(f"missing course.yaml at {rules_path}")
    if not index_path.is_file():
        raise FileNotFoundError(f"missing index.json at {index_path}")

    cfg = build_config_dict(
        model=embedder.model_name, dims=embedder.dimensions,
        registry_path=registry_path, topic_link_path=topic_link_path,
        classification_rules_path=rules_path,
        index_schema_version=_INDEX_SCHEMA_VERSION,
    )
    build_fp = cfg["build_config_fingerprint"]

    index = json.loads(index_path.read_text(encoding="utf-8"))
    registry = json.loads(registry_path.read_text(encoding="utf-8"))
    topic_links = json.loads(topic_link_path.read_text(encoding="utf-8")).get("mappings", {})

    term_regex = build_regex(registry)

    derived_path = repo_root / _DERIVED_DIR
    cache = EmbeddingCache(derived_path / "embedding-cache.jsonl")
    # full_rebuild=True deliberately skips cache.load() so test-mode counting embedders see the same call count as a fresh build.
    if not dry_run and not full_rebuild:
        cache.load()
    prior_manifest = {} if full_rebuild else _load_prior_manifest(derived_path / "manifest.json")
    prior_catalog: dict[str, dict] = {}
    prior_catalog_path = derived_path / "catalog.jsonl"
    if not full_rebuild and prior_catalog_path.is_file():
        for line in prior_catalog_path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if line:
                rec = json.loads(line)
                prior_catalog[rec["doc_id"]] = rec

    catalog_records: list[dict] = []
    new_manifest_docs: dict[str, dict] = {}
    cache_hits = 0
    api_calls = 0
    docs_reused = 0
    docs_visited = 0
    source_commit = index.get("source_commit") or _git_head(repo_root)
    index_fingerprint = index["content_fingerprint"]

    # --- Lecture chunker pre-expansion -------------------------------------
    # Files classified as kind == "lecture_session" are replaced by the
    # chunker's per-problem and per-turn record stream. lecture_session is
    # retired and never emitted into the catalog.
    registry_terms = set(registry.get("terms", {}).keys())
    chunker_outputs: list = []  # list[ChunkerOutput]
    expanded_files: list[tuple[str, dict]] = []
    for topic, file_record in _student_visible_files(index):
        if file_record.get("kind") != "lecture_session":
            expanded_files.append((topic, file_record))
            continue
        source_path_rel = file_record["path"]
        out = chunk_lecture_jsonl(repo_root / source_path_rel, term_registry=registry_terms)
        chunker_outputs.append(out)
        # Synthesize pseudo-FileRecords for each lecture_problem; lecture_turn
        # records are emitted directly to catalog_records below (no embedding,
        # no profile, no main-loop pass).
        for prob in out.lecture_problems:
            expanded_files.append((topic, {
                "path": source_path_rel,
                "kind": "lecture_problem",
                "topic": topic,
                "sha256": file_record.get("sha256", ""),
                "visibility": "student",
                "student_access": True,
                "source_system": "lectures_jsonl",
                "release_status": file_record.get("release_status", "released"),
                "_chunker_record": prob,  # carries the chunker payload
            }))

    for topic, file_record in expanded_files:
        source_path_rel = file_record["path"]
        source_path = repo_root / source_path_rel
        if not source_path.is_file():
            continue
        docs_visited += 1
        source_text = source_path.read_text(encoding="utf-8", errors="replace")
        source_hash = _sha256_file(source_path)
        chunker_record = file_record.get("_chunker_record")
        if chunker_record is not None:
            # Replace source_text and downstream metadata with chunker output.
            source_text = chunker_record["text"]
            doc_id = chunker_record["doc_id"]
            kind = chunker_record["kind"]
        else:
            doc_id = _doc_id_for(file_record)
            kind = file_record["kind"]
        linked_entry = topic_links.get(topic, {}).get("entry")

        prior = prior_manifest.get(doc_id, {})
        if (prior.get("source_hash") == source_hash
                and prior.get("build_config_fingerprint") == build_fp
                and doc_id in prior_catalog):
            catalog_records.append(prior_catalog[doc_id])
            new_manifest_docs[doc_id] = prior
            docs_reused += 1
            continue

        # Track whether only the build_config changed (not the source).
        # In that case we force a fresh embedding call to ensure the new
        # config is fully reflected; relying on a stale cache entry would
        # silently produce embeddings that pre-date the config change.
        build_config_changed = (
            prior.get("source_hash") == source_hash
            and prior.get("build_config_fingerprint") != build_fp
            and bool(prior)  # there WAS a prior entry
        )

        doc_meta = {
            "doc_id": doc_id, "kind": kind, "topic": topic,
            "source_path": source_path_rel,
            "linked_syntopicon_entry": linked_entry,
        }
        profile = assemble_profile(doc_meta, source_text, term_regex, registry)
        text_hash = _sha256_text(profile.text)

        if dry_run:
            continue

        cached_vec = None if build_config_changed else cache.get(model=embedder.model_name, text=profile.text)
        if cached_vec is not None and len(cached_vec) == embedder.dimensions:
            vector = cached_vec
            cache_hits += 1
        else:
            vector = embedder.embed_one(profile.text)
            cache.put(model=embedder.model_name, text=profile.text, vector=vector)
            api_calls += 1

        record = {
            "schema_version": CATALOG_SCHEMA_VERSION,
            "doc_id": doc_id,
            "kind": kind,
            "topic": topic,
            "privacy_tier": _privacy_tier_for(kind),
            "release_status": file_record.get("release_status", "released"),
            "source_path": source_path_rel,
            "source_sha256": source_hash,
            "source_commit": source_commit,
            "classification": {
                "kind": kind,
                "visibility": file_record.get("visibility", ""),
                "student_access": file_record.get("student_access", False),
            },
            "concept_profile": profile.text,
            "profile_schema_version": PROFILE_SCHEMA_VERSION,
            "embedding_text_sha256": text_hash,
            "embedding_model": embedder.model_name,
            "embedding_vector": vector,
            "concepts_present": (
                chunker_record["concepts_present"]
                if chunker_record is not None
                else profile.concepts_present
            ),
            "connected_concepts": profile.connected_concepts,
            "linked_syntopicon_entry": linked_entry,
            "token_count": profile.token_count,
            "sections": profile.sections,
            "packing_hint": profile.packing_hint,
            "build_config_fingerprint": build_fp,
            "index_content_fingerprint": index_fingerprint,
            "validation_status": "passed",
        }
        catalog_records.append(record)
        new_manifest_docs[doc_id] = {
            "source_hash": source_hash,
            "build_config_fingerprint": build_fp,
            "embedding_text_sha256": text_hash,
            "concept_count": len(profile.concepts_present),
        }

    if dry_run:
        return BuildStats(
            docs_processed=docs_visited,
            docs_reused=docs_reused, cache_hits=cache_hits, api_calls=api_calls,
        )

    # --- Append lecture_turn records (citation-only, no embedding) ---------
    for out in chunker_outputs:
        for turn in out.lecture_turns:
            catalog_records.append({
                "schema_version": CATALOG_SCHEMA_VERSION,
                "doc_id": turn["doc_id"],
                "kind": "lecture_turn",
                "topic": "",  # turns aren't topic-anchored
                "privacy_tier": _privacy_tier_for("lecture_turn"),
                "release_status": "released",
                "source_path": "",  # not file-addressable
                "source_sha256": "",
                "source_commit": source_commit,
                "classification": {
                    "kind": "lecture_turn",
                    "visibility": "student",
                    "student_access": True,
                },
                "lecture_date": turn["lecture_date"],
                "row_index": turn["row_index"],
                "turn_id": turn["turn_id"],
                "start": turn["start"],
                "end": turn["end"],
                "speaker": turn["speaker"],
                "label": turn["label"],
                "text": turn["text"],
                "word_count": turn["word_count"],
                "duration": turn["duration"],
                "parent_problem_doc_id": turn["parent_problem_doc_id"],
                "build_config_fingerprint": build_fp,
                "index_content_fingerprint": index_fingerprint,
                "validation_status": "passed",
            })

    # --- Combined row lookup map (used by reference parser AND sidecar) ---
    combined_lookup: dict = {}
    for out in chunker_outputs:
        combined_lookup.update(out.row_lookup)

    # --- Reference parsing on syntopicon records --------------------------
    syntopicon_dir = repo_root / "content" / "syntopicon-mgrecon780"
    refs_by_term = parse_all_references(syntopicon_dir, combined_lookup)
    for rec in catalog_records:
        if rec.get("kind") != "syntopicon_entry":
            continue
        term_slug = rec.get("topic", "")
        rec["references"] = refs_by_term.get(term_slug, [])
        rec["schema_version"] = 2  # bumped: now carries `references`

    chunk_to_cards = _build_chunk_to_referencing_cards(catalog_records)
    term_to_docs = _build_term_to_docs(catalog_records)
    concept_graph = _build_concept_graph(registry)

    # Atomic write to staging dir
    staging_root = derived_path.parent / (_DERIVED_DIR.split("/")[-1] + ".staging")
    if staging_root.exists():
        shutil.rmtree(staging_root)
    staging_root.mkdir(parents=True)
    try:
        with open(staging_root / "catalog.jsonl", "w", encoding="utf-8") as f:
            for rec in catalog_records:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        artifact_input = "".join(sorted(d.get("source_hash", "") for d in new_manifest_docs.values())) + build_fp
        artifact_id = "sha256:" + hashlib.sha256(artifact_input.encode("utf-8")).hexdigest()
        now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
        (staging_root / "manifest.json").write_text(json.dumps({
            "schema_version": MANIFEST_SCHEMA_VERSION,
            "build_timestamp_utc": now_utc,
            "artifact_version_id": artifact_id,
            "build_config_fingerprint": build_fp,
            "embedding_model": embedder.model_name,
            "docs": new_manifest_docs,
        }, indent=2), encoding="utf-8")
        (staging_root / "build-config.json").write_text(json.dumps(cfg, indent=2), encoding="utf-8")
        (staging_root / "term-to-docs.json").write_text(json.dumps(term_to_docs, indent=2, sort_keys=True), encoding="utf-8")
        (staging_root / "concept-graph.json").write_text(json.dumps(concept_graph, indent=2, sort_keys=True), encoding="utf-8")
        # Persist combined lecture_row_lookup for downstream debugging
        (staging_root / "lecture_row_lookup.json").write_text(
            row_lookup_to_json(combined_lookup), encoding="utf-8",
        )
        (staging_root / "chunk_to_referencing_cards.json").write_text(
            json.dumps(chunk_to_cards, indent=2, sort_keys=True), encoding="utf-8",
        )
        (staging_root / "build-info.json").write_text(json.dumps({
            "schema_version": 1,
            "artifact_version_id": artifact_id,
            "build_config_fingerprint": build_fp,
            "build_timestamp_utc": now_utc,
            "source_commit": source_commit,
            "embedding_model": embedder.model_name,
            "docs_processed": len(catalog_records),
            "docs_reused": docs_reused,
            "cache_hits": cache_hits,
            "api_calls_made": api_calls,
            "validation_status": "passed",
        }, indent=2), encoding="utf-8")

        # Persist embedding cache and copy into staging
        cache.flush()
        if derived_path.is_dir():
            cache_file = derived_path / "embedding-cache.jsonl"
            if cache_file.is_file():
                shutil.copy2(cache_file, staging_root / "embedding-cache.jsonl")

        _atomic_replace_dir(staging_root, derived_path)
    except Exception:
        if staging_root.exists():
            shutil.rmtree(staging_root, ignore_errors=True)
        raise

    # 11. Emit reproducible release tarball
    release_dir = derived_path / "release"
    release_dir.mkdir(parents=True, exist_ok=True)
    tarball_name = f"student-chat-{artifact_id.replace(':', '-')}.tar.gz"
    tarball_path = release_dir / tarball_name
    if tarball_path.exists():
        tarball_path.unlink()
    with tarfile.open(tarball_path, "w:gz") as tar:
        for name in ("catalog.jsonl", "manifest.json", "build-config.json",
                     "term-to-docs.json", "concept-graph.json", "build-info.json",
                     "chunk_to_referencing_cards.json", "lecture_row_lookup.json"):
            full = derived_path / name
            if full.is_file():
                tar.add(full, arcname=name)
    tar_hash = hashlib.sha256(tarball_path.read_bytes()).hexdigest()
    (release_dir / (tarball_name + ".sha256")).write_text(
        f"sha256:{tar_hash}  {tarball_name}\n", encoding="utf-8"
    )

    return BuildStats(
        docs_processed=docs_visited,
        docs_reused=docs_reused,
        cache_hits=cache_hits,
        api_calls=api_calls,
    )


def check(*, repo_root: Path, embedder: Embedder) -> int:
    """--check doctor mode. Validates inputs without mutating."""
    issues: list[str] = []
    for rel in ("config/course.yaml", "index/index.json",
                "content/syntopicon-mgrecon780/_registry.json",
                "index/inputs/syntopicon-links/topic-to-entry.json"):
        if not (repo_root / rel).is_file():
            issues.append(f"missing: {rel}")
    if issues:
        for i in issues:
            print(f"FAIL: {i}")
        return 1
    print("OK: all required inputs present")
    print(f"OK: embedder model = {embedder.model_name} ({embedder.dimensions} dims)")
    registry_path = repo_root / "content" / "syntopicon-mgrecon780" / "_registry.json"
    topic_link_path = repo_root / "index" / "inputs" / "syntopicon-links" / "topic-to-entry.json"
    rules_path = repo_root / "config" / "course.yaml"
    cfg = build_config_dict(
        model=embedder.model_name, dims=embedder.dimensions,
        registry_path=registry_path, topic_link_path=topic_link_path,
        classification_rules_path=rules_path,
        index_schema_version=_INDEX_SCHEMA_VERSION,
    )
    print(f"OK: build_config_fingerprint = {cfg['build_config_fingerprint']}")
    return 0

	"""Build pipeline orchestrator for derived/student-chat/.

	Reads the index (filtered to student_access=true), loads each source doc,
	extracts a concept profile, embeds via the injected Embedder (cache-aware),
	and writes the full catalog artifact atomically.

	Incremental: per-doc reuse if (source_hash, build_config_fingerprint) match
	the prior manifest. --full-rebuild bypasses cache.
	--check mode validates inputs and reports without mutating.
	--dry-run extracts profiles + computes hashes but writes no artifacts.
	"""
	from __future__ import annotations
	import hashlib
	import json
	import os
	import shutil
	import subprocess
	import tarfile
	from dataclasses import dataclass
	from datetime import datetime, timezone
	from pathlib import Path

	from scripts.lib.build_config import (
	build_config_dict, BUILD_CONFIG_SCHEMA_VERSION,
	CATALOG_SCHEMA_VERSION, MANIFEST_SCHEMA_VERSION,
	)
	from scripts.lib.concept_profile import assemble as assemble_profile, PROFILE_SCHEMA_VERSION
	from scripts.lib.embedder import Embedder
	from scripts.lib.embedding_cache import EmbeddingCache
	from scripts.lib.lecture_chunker import chunk_lecture_jsonl, row_lookup_to_json
	from scripts.lib.ranker import _cosine_similarity
	from scripts.lib.syntopicon_reference_parser import parse_all_references
	from scripts.lib.term_extraction import build_regex


	_DERIVED_DIR = "derived/student-chat"
	_INDEX_SCHEMA_VERSION = 1


	@dataclass
	class BuildStats:
	docs_processed: int
	docs_reused: int
	cache_hits: int
	api_calls: int


	def _sha256_text(text: str) -> str:
	return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest()


	def _sha256_file(path: Path) -> str:
	h = hashlib.sha256()
	with open(path, "rb") as f:
	for chunk in iter(lambda: f.read(8192), b""):
	h.update(chunk)
	return "sha256:" + h.hexdigest()


	def _git_head(repo_root: Path) -> str:
	try:
	out = subprocess.run(
	["git", "-C", str(repo_root), "rev-parse", "HEAD"],
	capture_output=True, text=True, check=True,
	)
	return out.stdout.strip()
	except (subprocess.CalledProcessError, FileNotFoundError) as e:
	raise RuntimeError(
	f"unable to determine source_commit: git rev-parse HEAD failed in {repo_root} "
	f"and index.json did not provide source_commit. The catalog contract requires a "
	f"real commit SHA matching ^[0-9a-f]{{7,40}}$; refusing to emit a placeholder."
	) from e


	_REDACTED_KINDS = frozenset({"lecture_problem", "lecture_turn"})


	def _privacy_tier_for(kind: str) -> str:
	return "redacted_classroom_speech" if kind in _REDACTED_KINDS else "public_course_content"


	def _doc_id_for(file_record: dict) -> str:
	"""Stable doc_id derived from kind + path stem + short path hash.

	The path hash discriminator ensures uniqueness across same-stem files
	in different directories.
	"""
	path = file_record.get("path", "")
	stem = Path(path).stem
	kind = file_record.get("kind", "")
	# 8-hex-char hash of the full path for collision resistance
	path_disc = hashlib.sha256(path.encode("utf-8")).hexdigest()[:8]
	return f"{kind}_{stem}_{path_disc}"


	def _student_visible_files(index: dict) -> list[tuple[str, dict]]:
	"""Yield (topic, file_record) for every student-visible record in the
	index — including syntopicon entries, which the index stores in their
	own per-topic array with a lighter shape than FileRecord."""
	out: list[tuple[str, dict]] = []
	for topic, record in index.get("topics", {}).items():
	for f in record.get("files", []):
	if not f.get("student_access", False):
	continue
	if f.get("kind") == "syntopicon_registry":
	continue
	out.append((topic, f))
	for entry in record.get("syntopicon_entries", []):
	# Normalize the SyntopiconEntry shape to a FileRecord so the
	# rest of the pipeline can treat it uniformly. Entries are
	# student-accessible by classification (course.yaml).
	normalized = {
	"path": entry["path"],
	"kind": "syntopicon_entry",
	"topic": topic,
	"sha256": entry.get("sha256", ""),
	"visibility": "student",
	"student_access": True,
	"source_system": "syntopicon",
	"release_status": "released",
	}
	out.append((topic, normalized))
	return out


	def _load_prior_manifest(manifest_path: Path) -> dict:
	if not manifest_path.is_file():
	return {}
	return json.loads(manifest_path.read_text(encoding="utf-8")).get("docs", {})


	def _atomic_replace_dir(staging: Path, target: Path) -> None:
	"""Replace target/ with staging/ atomically (with backup-rename dance for Windows).

	Recovers from an orphaned .bak left by a prior crash: if target is missing
	but .bak exists, restores .bak to target first. The remaining window
	between (target -> bak) and (staging -> target) is small but still not
	truly atomic; in the worst case (kill mid-rename), next build sees a
	missing derived/ and treats all docs as new (correct but expensive).
	"""
	target.parent.mkdir(parents=True, exist_ok=True)
	bak = target.with_name(target.name + ".bak")
	# Recovery: if target was lost in a prior crash but bak exists, restore it
	if not target.exists() and bak.exists():
	os.rename(bak, target)
	if bak.exists():
	shutil.rmtree(bak)
	if target.exists():
	os.rename(target, bak)
	os.rename(staging, target)
	if bak.exists():
	shutil.rmtree(bak)


	def _build_chunk_to_referencing_cards(catalog_records: list[dict]) -> dict[str, list[dict]]:
	by_doc = {rec["doc_id"]: rec for rec in catalog_records}
	chunk_to_cards: dict[str, list[dict]] = {}
	for rec in catalog_records:
	if rec.get("kind") != "syntopicon_entry":
	continue
	card_id = rec["doc_id"]
	card_vec = rec.get("embedding_vector", [])
	for ref in rec.get("references", []) or []:
	chunk_id = ref.get("lecture_problem_doc_id")
	if not chunk_id or chunk_id not in by_doc:
	continue
	chunk_vec = by_doc[chunk_id].get("embedding_vector", [])
	sim = _cosine_similarity(card_vec, chunk_vec)
	chunk_to_cards.setdefault(chunk_id, []).append({
	"syntopicon_doc_id": card_id,
	"similarity_score": sim,
	})
	return {
	chunk_id: sorted(cards, key=lambda c: c["similarity_score"], reverse=True)[:5]
	for chunk_id, cards in chunk_to_cards.items()
	}


	def _build_term_to_docs(catalog_records: list[dict]) -> dict[str, list[str]]:
	term_to_docs: dict[str, list[str]] = {}
	for rec in catalog_records:
	if rec.get("kind") == "lecture_turn":
	continue
	for slug in rec.get("concepts_present", []):
	term_to_docs.setdefault(slug, []).append(rec["doc_id"])
	return {k: sorted(set(v)) for k, v in sorted(term_to_docs.items())}


	def _build_concept_graph(registry: dict) -> dict[str, dict]:
	concept_graph: dict[str, dict] = {}
	for slug, term in registry.get("terms", {}).items():
	depends_on = term.get("depends_on", []) or []
	concept_graph[slug] = {"depends_on": sorted(depends_on), "dependents": []}
	for slug, info in concept_graph.items():
	for dep in info["depends_on"]:
	if dep in concept_graph:
	concept_graph[dep]["dependents"].append(slug)
	for info in concept_graph.values():
	info["dependents"] = sorted(set(info["dependents"]))
	return concept_graph


	def run(
	*,
	repo_root: Path,
	embedder: Embedder,
	full_rebuild: bool = False,
	dry_run: bool = False,
	) -> BuildStats:
	"""Run the build pipeline. Returns BuildStats."""
	registry_path = repo_root / "content" / "syntopicon-mgrecon780" / "_registry.json"
	topic_link_path = repo_root / "index" / "inputs" / "syntopicon-links" / "topic-to-entry.json"
	rules_path = repo_root / "config" / "course.yaml"
	index_path = repo_root / "index" / "index.json"

	if not registry_path.is_file():
	raise FileNotFoundError(f"missing slim registry at {registry_path}")
	if not topic_link_path.is_file():
	raise FileNotFoundError(f"missing topic-to-entry map at {topic_link_path}")
	if not rules_path.is_file():
	raise FileNotFoundError(f"missing course.yaml at {rules_path}")
	if not index_path.is_file():
	raise FileNotFoundError(f"missing index.json at {index_path}")

	cfg = build_config_dict(
	model=embedder.model_name, dims=embedder.dimensions,
	registry_path=registry_path, topic_link_path=topic_link_path,
	classification_rules_path=rules_path,
	index_schema_version=_INDEX_SCHEMA_VERSION,
	)
	build_fp = cfg["build_config_fingerprint"]

	index = json.loads(index_path.read_text(encoding="utf-8"))
	registry = json.loads(registry_path.read_text(encoding="utf-8"))
	topic_links = json.loads(topic_link_path.read_text(encoding="utf-8")).get("mappings", {})

	term_regex = build_regex(registry)

	derived_path = repo_root / _DERIVED_DIR
	cache = EmbeddingCache(derived_path / "embedding-cache.jsonl")
	# full_rebuild=True deliberately skips cache.load() so test-mode counting embedders see the same call count as a fresh build.
	if not dry_run and not full_rebuild:
	cache.load()
	prior_manifest = {} if full_rebuild else _load_prior_manifest(derived_path / "manifest.json")
	prior_catalog: dict[str, dict] = {}
	prior_catalog_path = derived_path / "catalog.jsonl"
	if not full_rebuild and prior_catalog_path.is_file():
	for line in prior_catalog_path.read_text(encoding="utf-8").splitlines():
	line = line.strip()
	if line:
	rec = json.loads(line)
	prior_catalog[rec["doc_id"]] = rec

	catalog_records: list[dict] = []
	new_manifest_docs: dict[str, dict] = {}
	cache_hits = 0
	api_calls = 0
	docs_reused = 0
	docs_visited = 0
	source_commit = index.get("source_commit") or _git_head(repo_root)
	index_fingerprint = index["content_fingerprint"]

	# --- Lecture chunker pre-expansion -------------------------------------
	# Files classified as kind == "lecture_session" are replaced by the
	# chunker's per-problem and per-turn record stream. lecture_session is
	# retired and never emitted into the catalog.
	registry_terms = set(registry.get("terms", {}).keys())
	chunker_outputs: list = [] # list[ChunkerOutput]
	expanded_files: list[tuple[str, dict]] = []
	for topic, file_record in _student_visible_files(index):
	if file_record.get("kind") != "lecture_session":
	expanded_files.append((topic, file_record))
	continue
	source_path_rel = file_record["path"]
	out = chunk_lecture_jsonl(repo_root / source_path_rel, term_registry=registry_terms)
	chunker_outputs.append(out)
	# Synthesize pseudo-FileRecords for each lecture_problem; lecture_turn
	# records are emitted directly to catalog_records below (no embedding,
	# no profile, no main-loop pass).
	for prob in out.lecture_problems:
	expanded_files.append((topic, {
	"path": source_path_rel,
	"kind": "lecture_problem",
	"topic": topic,
	"sha256": file_record.get("sha256", ""),
	"visibility": "student",
	"student_access": True,
	"source_system": "lectures_jsonl",
	"release_status": file_record.get("release_status", "released"),
	"_chunker_record": prob, # carries the chunker payload
	}))

	for topic, file_record in expanded_files:
	source_path_rel = file_record["path"]
	source_path = repo_root / source_path_rel
	if not source_path.is_file():
	continue
	docs_visited += 1
	source_text = source_path.read_text(encoding="utf-8", errors="replace")
	source_hash = _sha256_file(source_path)
	chunker_record = file_record.get("_chunker_record")
	if chunker_record is not None:
	# Replace source_text and downstream metadata with chunker output.
	source_text = chunker_record["text"]
	doc_id = chunker_record["doc_id"]
	kind = chunker_record["kind"]
	else:
	doc_id = _doc_id_for(file_record)
	kind = file_record["kind"]
	linked_entry = topic_links.get(topic, {}).get("entry")

	prior = prior_manifest.get(doc_id, {})
	if (prior.get("source_hash") == source_hash
	and prior.get("build_config_fingerprint") == build_fp
	and doc_id in prior_catalog):
	catalog_records.append(prior_catalog[doc_id])
	new_manifest_docs[doc_id] = prior
	docs_reused += 1
	continue

	# Track whether only the build_config changed (not the source).
	# In that case we force a fresh embedding call to ensure the new
	# config is fully reflected; relying on a stale cache entry would
	# silently produce embeddings that pre-date the config change.
	build_config_changed = (
	prior.get("source_hash") == source_hash
	and prior.get("build_config_fingerprint") != build_fp
	and bool(prior) # there WAS a prior entry
	)

	doc_meta = {
	"doc_id": doc_id, "kind": kind, "topic": topic,
	"source_path": source_path_rel,
	"linked_syntopicon_entry": linked_entry,
	}
	profile = assemble_profile(doc_meta, source_text, term_regex, registry)
	text_hash = _sha256_text(profile.text)

	if dry_run:
	continue

	cached_vec = None if build_config_changed else cache.get(model=embedder.model_name, text=profile.text)
	if cached_vec is not None and len(cached_vec) == embedder.dimensions:
	vector = cached_vec
	cache_hits += 1
	else:
	vector = embedder.embed_one(profile.text)
	cache.put(model=embedder.model_name, text=profile.text, vector=vector)
	api_calls += 1

	record = {
	"schema_version": CATALOG_SCHEMA_VERSION,
	"doc_id": doc_id,
	"kind": kind,
	"topic": topic,
	"privacy_tier": _privacy_tier_for(kind),
	"release_status": file_record.get("release_status", "released"),
	"source_path": source_path_rel,
	"source_sha256": source_hash,
	"source_commit": source_commit,
	"classification": {
	"kind": kind,
	"visibility": file_record.get("visibility", ""),
	"student_access": file_record.get("student_access", False),
	},
	"concept_profile": profile.text,
	"profile_schema_version": PROFILE_SCHEMA_VERSION,
	"embedding_text_sha256": text_hash,
	"embedding_model": embedder.model_name,
	"embedding_vector": vector,
	"concepts_present": (
	chunker_record["concepts_present"]
	if chunker_record is not None
	else profile.concepts_present
	),
	"connected_concepts": profile.connected_concepts,
	"linked_syntopicon_entry": linked_entry,
	"token_count": profile.token_count,
	"sections": profile.sections,
	"packing_hint": profile.packing_hint,
	"build_config_fingerprint": build_fp,
	"index_content_fingerprint": index_fingerprint,
	"validation_status": "passed",
	}
	catalog_records.append(record)
	new_manifest_docs[doc_id] = {
	"source_hash": source_hash,
	"build_config_fingerprint": build_fp,
	"embedding_text_sha256": text_hash,
	"concept_count": len(profile.concepts_present),
	}

	if dry_run:
	return BuildStats(
	docs_processed=docs_visited,
	docs_reused=docs_reused, cache_hits=cache_hits, api_calls=api_calls,
	)

	# --- Append lecture_turn records (citation-only, no embedding) ---------
	for out in chunker_outputs:
	for turn in out.lecture_turns:
	catalog_records.append({
	"schema_version": CATALOG_SCHEMA_VERSION,
	"doc_id": turn["doc_id"],
	"kind": "lecture_turn",
	"topic": "", # turns aren't topic-anchored
	"privacy_tier": _privacy_tier_for("lecture_turn"),
	"release_status": "released",
	"source_path": "", # not file-addressable
	"source_sha256": "",
	"source_commit": source_commit,
	"classification": {
	"kind": "lecture_turn",
	"visibility": "student",
	"student_access": True,
	},
	"lecture_date": turn["lecture_date"],
	"row_index": turn["row_index"],
	"turn_id": turn["turn_id"],
	"start": turn["start"],
	"end": turn["end"],
	"speaker": turn["speaker"],
	"label": turn["label"],
	"text": turn["text"],
	"word_count": turn["word_count"],
	"duration": turn["duration"],
	"parent_problem_doc_id": turn["parent_problem_doc_id"],
	"build_config_fingerprint": build_fp,
	"index_content_fingerprint": index_fingerprint,
	"validation_status": "passed",
	})

	# --- Combined row lookup map (used by reference parser AND sidecar) ---
	combined_lookup: dict = {}
	for out in chunker_outputs:
	combined_lookup.update(out.row_lookup)

	# --- Reference parsing on syntopicon records --------------------------
	syntopicon_dir = repo_root / "content" / "syntopicon-mgrecon780"
	refs_by_term = parse_all_references(syntopicon_dir, combined_lookup)
	for rec in catalog_records:
	if rec.get("kind") != "syntopicon_entry":
	continue
	term_slug = rec.get("topic", "")
	rec["references"] = refs_by_term.get(term_slug, [])
	rec["schema_version"] = 2 # bumped: now carries `references`

	chunk_to_cards = _build_chunk_to_referencing_cards(catalog_records)
	term_to_docs = _build_term_to_docs(catalog_records)
	concept_graph = _build_concept_graph(registry)

	# Atomic write to staging dir
	staging_root = derived_path.parent / (_DERIVED_DIR.split("/")[-1] + ".staging")
	if staging_root.exists():
	shutil.rmtree(staging_root)
	staging_root.mkdir(parents=True)
	try:
	with open(staging_root / "catalog.jsonl", "w", encoding="utf-8") as f:
	for rec in catalog_records:
	f.write(json.dumps(rec, ensure_ascii=False) + "\n")
	artifact_input = "".join(sorted(d.get("source_hash", "") for d in new_manifest_docs.values())) + build_fp
	artifact_id = "sha256:" + hashlib.sha256(artifact_input.encode("utf-8")).hexdigest()
	now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
	(staging_root / "manifest.json").write_text(json.dumps({
	"schema_version": MANIFEST_SCHEMA_VERSION,
	"build_timestamp_utc": now_utc,
	"artifact_version_id": artifact_id,
	"build_config_fingerprint": build_fp,
	"embedding_model": embedder.model_name,
	"docs": new_manifest_docs,
	}, indent=2), encoding="utf-8")
	(staging_root / "build-config.json").write_text(json.dumps(cfg, indent=2), encoding="utf-8")
	(staging_root / "term-to-docs.json").write_text(json.dumps(term_to_docs, indent=2, sort_keys=True), encoding="utf-8")
	(staging_root / "concept-graph.json").write_text(json.dumps(concept_graph, indent=2, sort_keys=True), encoding="utf-8")
	# Persist combined lecture_row_lookup for downstream debugging
	(staging_root / "lecture_row_lookup.json").write_text(
	row_lookup_to_json(combined_lookup), encoding="utf-8",
	)
	(staging_root / "chunk_to_referencing_cards.json").write_text(
	json.dumps(chunk_to_cards, indent=2, sort_keys=True), encoding="utf-8",
	)
	(staging_root / "build-info.json").write_text(json.dumps({
	"schema_version": 1,
	"artifact_version_id": artifact_id,
	"build_config_fingerprint": build_fp,
	"build_timestamp_utc": now_utc,
	"source_commit": source_commit,
	"embedding_model": embedder.model_name,
	"docs_processed": len(catalog_records),
	"docs_reused": docs_reused,
	"cache_hits": cache_hits,
	"api_calls_made": api_calls,
	"validation_status": "passed",
	}, indent=2), encoding="utf-8")

	# Persist embedding cache and copy into staging
	cache.flush()
	if derived_path.is_dir():
	cache_file = derived_path / "embedding-cache.jsonl"
	if cache_file.is_file():
	shutil.copy2(cache_file, staging_root / "embedding-cache.jsonl")

	_atomic_replace_dir(staging_root, derived_path)
	except Exception:
	if staging_root.exists():
	shutil.rmtree(staging_root, ignore_errors=True)
	raise

	# 11. Emit reproducible release tarball
	release_dir = derived_path / "release"
	release_dir.mkdir(parents=True, exist_ok=True)
	tarball_name = f"student-chat-{artifact_id.replace(':', '-')}.tar.gz"
	tarball_path = release_dir / tarball_name
	if tarball_path.exists():
	tarball_path.unlink()
	with tarfile.open(tarball_path, "w:gz") as tar:
	for name in ("catalog.jsonl", "manifest.json", "build-config.json",
	"term-to-docs.json", "concept-graph.json", "build-info.json",
	"chunk_to_referencing_cards.json", "lecture_row_lookup.json"):
	full = derived_path / name
	if full.is_file():
	tar.add(full, arcname=name)
	tar_hash = hashlib.sha256(tarball_path.read_bytes()).hexdigest()
	(release_dir / (tarball_name + ".sha256")).write_text(
	f"sha256:{tar_hash} {tarball_name}\n", encoding="utf-8"
	)

	return BuildStats(
	docs_processed=docs_visited,
	docs_reused=docs_reused,
	cache_hits=cache_hits,
	api_calls=api_calls,
	)


	def check(*, repo_root: Path, embedder: Embedder) -> int:
	"""--check doctor mode. Validates inputs without mutating."""
	issues: list[str] = []
	for rel in ("config/course.yaml", "index/index.json",
	"content/syntopicon-mgrecon780/_registry.json",
	"index/inputs/syntopicon-links/topic-to-entry.json"):
	if not (repo_root / rel).is_file():
	issues.append(f"missing: {rel}")
	if issues:
	for i in issues:
	print(f"FAIL: {i}")
	return 1
	print("OK: all required inputs present")
	print(f"OK: embedder model = {embedder.model_name} ({embedder.dimensions} dims)")
	registry_path = repo_root / "content" / "syntopicon-mgrecon780" / "_registry.json"
	topic_link_path = repo_root / "index" / "inputs" / "syntopicon-links" / "topic-to-entry.json"
	rules_path = repo_root / "config" / "course.yaml"
	cfg = build_config_dict(
	model=embedder.model_name, dims=embedder.dimensions,
	registry_path=registry_path, topic_link_path=topic_link_path,
	classification_rules_path=rules_path,
	index_schema_version=_INDEX_SCHEMA_VERSION,
	)
	print(f"OK: build_config_fingerprint = {cfg['build_config_fingerprint']}")
	return 0