"""/refresh orchestrator. Faculty mode (this task) and maintainer mode (Task 27).""" from __future__ import annotations import json import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Optional from lib import git_ops, index_generator, local_state, source_state, sync from lib.paths import repo_root class IndexDriftError(RuntimeError): pass class MissingStateError(RuntimeError): """Raised when STATE.json is missing — the maintainer hasn't run /refresh yet.""" class MissingIndexError(RuntimeError): """Raised when committed index/index.json is missing.""" _REFRESH_ALLOWLIST = [".me-hub", ".claude/settings.local.json"] def _ensure_clean_worktree() -> None: if not git_ops.worktree_clean(allowlist=_REFRESH_ALLOWLIST): raise RuntimeError( "refresh requires a clean worktree (allowed exceptions: .me-hub/, " "settings.local.json). Resolve outstanding changes first." ) def run_faculty_mode() -> None: """git pull -> regenerate index to compare -> error on drift -> update local-state. Named invariant: this function MUST NOT mutate tracked files. It only writes to the gitignored .me-hub/local-state.json. On any drift between regenerated and committed index, it raises IndexDriftError without writing anything. """ _ensure_clean_worktree() try: git_ops.pull() except subprocess.CalledProcessError: pass state = source_state.read() if state is None: raise MissingStateError( "STATE.json missing. Faculty mode expects a maintainer-produced STATE.json on main. " "Run /refresh on the maintainer machine first, then git pull on this machine." ) sha = git_ops.current_commit() regenerated = index_generator.generate(source_commit=sha) committed_path = repo_root() / "index" / "index.json" if not committed_path.is_file(): raise MissingIndexError("committed index/index.json is missing") committed = json.loads(committed_path.read_text(encoding="utf-8")) # Compare on content_fingerprint only — generated_utc and source_commit # differ legitimately across maintainer-write and faculty-regenerate. if regenerated["content_fingerprint"] != committed["content_fingerprint"]: raise IndexDriftError( f"regenerated index differs from committed index.\n" f" regenerated content_fingerprint: {regenerated['content_fingerprint']}\n" f" committed content_fingerprint: {committed['content_fingerprint']}\n" "Investigate (out-of-date local fixtures or maintainer regression). " "Faculty mode never rewrites tracked files." ) local_state.update_last_refresh(state_commit=state["source_commit"]) print("OK faculty refresh: index matches committed; local-state updated.") def run_maintainer_mode( *, box_togo_path: Path, tolk_syntopicon_path: Path, ferpa_lectures_raw: Optional[Path], allow_deletes: bool, push: bool, ) -> None: """Maintainer-mode refresh: per-source ingestion + regenerate + commit + push. Per-source allowlist commits: each `git add` is by exact path. We never use `git add .` or `git add -A`, so untracked files in other parts of the tree don't get accidentally committed. """ _ensure_clean_worktree() try: git_ops.pull() except subprocess.CalledProcessError: pass repo = repo_root() _ingest_source( src=box_togo_path, tgt=repo / "content" / "togo", commit_paths=["content/togo"], commit_message="refresh: ingest togo content", allow_deletes=allow_deletes, ) from lib import syntopicon_ingest syntopicon_ingest.ingest( tolk_syntopicon_path=tolk_syntopicon_path, target_dir=repo / "content" / "syntopicon-mgrecon780", course_pattern="MGRECON780_", expand_hops=1, allow_deletes=allow_deletes, ) # syntopicon_ingest stamps `filter.ingested_utc` every run; restore the prior # _registry.json if the only change is that timestamp, so unchanged inputs # don't produce phantom diffs. _restore_file_if_only_timestamp_changed( repo / "content" / "syntopicon-mgrecon780" / "_registry.json", ("filter", "ingested_utc"), ) git_ops.add_and_commit_if_changes( ["content/syntopicon-mgrecon780"], "refresh: ingest syntopicon (filtered to MGRECON780+1-hop)", ) if ferpa_lectures_raw and ferpa_lectures_raw.is_dir(): _normalize_lectures(ferpa_lectures_raw, repo / "content" / "lectures-jsonl" / "normalized") git_ops.add_and_commit_if_changes( ["content/lectures-jsonl/normalized"], "refresh: normalize lecture jsonl", ) sha = git_ops.current_commit() idx = index_generator.generate(source_commit=sha) index_generator.write_index(idx) state = _build_state(idx=idx, source_commit=sha) source_state.write(state) git_ops.add_and_commit( ["index/index.json", "STATE.json"], "refresh: regenerate index", ) # Auto-generate topic-to-syntopicon-term mapping from lib import topic_links links_dir = repo / "index" / "inputs" / "syntopicon-links" topic_links.run( index_path=repo / "index" / "index.json", registry_path=repo / "content" / "syntopicon-mgrecon780" / "_registry.json", output_dir=links_dir, ) # topic_links.run() stamps `generated_utc` into both output files every call; # restore the prior content if the only change is that timestamp. _restore_file_if_only_timestamp_changed(links_dir / "topic-to-entry.json", ("generated_utc",)) _restore_file_if_only_timestamp_changed(links_dir / "_unresolved.json", ("generated_utc",)) git_ops.add_and_commit_if_changes( ["index/inputs/syntopicon-links"], "refresh: regenerate topic-to-entry map", ) if push: git_ops.push() print("OK maintainer refresh complete.") def _ingest_source( *, src: Path, tgt: Path, commit_paths: list[str], commit_message: str, allow_deletes: bool, ) -> None: if not src.is_dir(): raise RuntimeError(f"source missing: {src}. Path required in maintainer mode.") tgt.mkdir(parents=True, exist_ok=True) diff = sync.stage_and_diff(src, tgt) if not (diff.additions or diff.modifications or diff.deletions): return print(f" staged sync of {src} -> {tgt.relative_to(repo_root())}:") print(sync.format_diff_summary(diff)) sync.apply(diff, src, tgt, allow_deletes=allow_deletes) git_ops.add_and_commit(commit_paths, commit_message) def _restore_file_if_only_timestamp_changed( file_path: Path, timestamp_field_path: tuple[str, ...] ) -> None: """If file_path differs from the HEAD version ONLY in the named timestamp field, restore the HEAD content. Keeps maintainer-mode refresh idempotent on unchanged upstream input (which would otherwise re-commit every run because the generator stamps the timestamp on every write). `timestamp_field_path` is a sequence of dict keys naming the location of the timestamp inside the JSON document. Examples: ("generated_utc",) # top-level field ("filter", "ingested_utc") # nested under "filter" No-op if the file isn't tracked yet (first ingest), JSON parsing fails, or the structural content (everything except the named timestamp) actually differs. """ if not file_path.is_file(): return repo = repo_root() try: rel_path = file_path.resolve().relative_to(repo.resolve()) except ValueError: # Path isn't under the repo root; we can't address it via git. return rel_posix = rel_path.as_posix() try: head_blob = subprocess.run( ["git", "show", f"HEAD:{rel_posix}"], check=True, capture_output=True, text=True, ).stdout except subprocess.CalledProcessError: # Not tracked yet; nothing to restore. return try: current = json.loads(file_path.read_text(encoding="utf-8")) head = json.loads(head_blob) except json.JSONDecodeError: return def _stripped(doc: dict) -> dict: # Walk into nested dicts following timestamp_field_path[:-1], pop the # leaf key. Returns a structurally-equal copy with the timestamp removed. if not timestamp_field_path: return doc out = dict(doc) cursor = out for key in timestamp_field_path[:-1]: inner = cursor.get(key) if not isinstance(inner, dict): return out cursor[key] = dict(inner) cursor = cursor[key] cursor.pop(timestamp_field_path[-1], None) return out if _stripped(current) == _stripped(head): file_path.write_text(head_blob, encoding="utf-8") def _normalize_lectures(ferpa_raw_dir: Path, target_dir: Path) -> None: """Apply redaction policy to every raw JSONL in ferpa_raw_dir. Each input file is scrubbed via redaction.scrub_to_instructor_only(), which drops Anonymous Student turns, redacts known student names from instructor text and problem outcomes, and writes the redacted JSONL atomically. The .redaction-report.txt files contain absolute source/output paths (machine-specific, not deterministic across maintainers) and would trip index_generator's unmatched-files diagnostic if left under content/. We pass an explicit report_path under .me-hub/redaction-reports/ (gitignored) so the report is written directly there, avoiding the non-atomic scrub-then-move window where a SIGKILL could leave a stray report under content/. No source_hash short-circuit: scrub_to_instructor_only is deterministic given identical input + config, so the rewritten JSONL is byte-identical on repeat runs and add_and_commit_if_changes handles the no-op case. """ from lib import redaction cfg = redaction.default_config() target_dir.mkdir(parents=True, exist_ok=True) # Defensive: clean up any pre-existing stray report files (in case a prior # version wrote them here). index_generator would otherwise treat them as # unmatched. for stray in target_dir.glob("*.redaction-report.txt"): stray.unlink() reports_dir = repo_root() / ".me-hub" / "redaction-reports" reports_dir.mkdir(parents=True, exist_ok=True) for raw_jsonl in sorted(ferpa_raw_dir.glob("*.jsonl")): out = target_dir / raw_jsonl.name report = reports_dir / (raw_jsonl.stem + ".redaction-report.txt") redaction.scrub_to_instructor_only(raw_jsonl, out, cfg, report_path=report) def _build_state(idx: dict, source_commit: str) -> dict: repo = repo_root() # Ensure all source directories exist for fingerprinting; empty dir → empty-input digest. (repo / "content" / "togo").mkdir(parents=True, exist_ok=True) (repo / "content" / "lectures-jsonl" / "normalized").mkdir(parents=True, exist_ok=True) (repo / "content" / "syntopicon-mgrecon780").mkdir(parents=True, exist_ok=True) now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") return { "schema_version": 1, "snapshot_utc": now_iso, "source_commit": source_commit, "sources": { "togo": { "ingested_utc": now_iso, "file_count": _count_files(repo / "content" / "togo"), "content_fingerprint": source_state.compute_directory_fingerprint( repo / "content" / "togo" ), }, "lectures_jsonl": { "ingested_utc": now_iso, "file_count": _count_files(repo / "content" / "lectures-jsonl" / "normalized"), "content_fingerprint": source_state.compute_directory_fingerprint( repo / "content" / "lectures-jsonl" / "normalized" ), "redaction_policy": "v1.0", }, "syntopicon": { "ingested_utc": now_iso, "entry_count": _count_files(repo / "content" / "syntopicon-mgrecon780"), "content_fingerprint": source_state.compute_directory_fingerprint( repo / "content" / "syntopicon-mgrecon780" ), }, }, "index": { "generator_version": idx["generator_version"], "topics": len(idx["topics"]), "files_indexed": sum(len(t.get("files", [])) for t in idx["topics"].values()), "content_fingerprint": idx["content_fingerprint"], }, } def _count_files(d: Path) -> int: if not d.is_dir(): return 0 return sum(1 for p in d.rglob("*") if p.is_file() and p.name != ".gitkeep")