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