"""Derive a repository's work-record projection from the forge (STATE-WP-0083-T01). `ADR-012` decision 1 makes the forge the projection source. Central does its own reading: it clones the repository's default branch and derives from that, rather than accepting a projection computed elsewhere — which `ADR-010` decision 5 forbids. This module is read-only. Deriving must be safe to run at any time, because it is what makes the reset in `T03` verifiable: you can always ask what the projection *should* be without changing anything. """ from __future__ import annotations import re import subprocess import tempfile import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any import yaml # Same derivation as ADR-007 / repo-manager, so a forge-derived projection and a # preliminary overlay compute identical identities for the same record. _WORK_RECORD_NAMESPACE = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009") _TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL) _HEADING_RE = re.compile(r"^(#{1,4})\s+(.+?)$", re.MULTILINE) DEFAULT_FORGE_BASE = "https://forgejo.coulomb.social/coulomb" def derived_record_uuid(record_id: str) -> str: return str(uuid.uuid5(_WORK_RECORD_NAMESPACE, f"helixforge\n{record_id}")) class ForgeDeriveError(RuntimeError): """The repository could not be read from the forge.""" @dataclass class DerivedTask: record_id: str uuid: str title: str | None status: str | None priority: str | None @dataclass class DerivedWorkplan: record_id: str uuid: str title: str | None status: str | None relative_path: str archived: bool tasks: list[DerivedTask] = field(default_factory=list) @dataclass class DerivedProjection: repo_slug: str commit: str workplans: list[DerivedWorkplan] = field(default_factory=list) @property def task_count(self) -> int: return sum(len(w.tasks) for w in self.workplans) def to_dict(self) -> dict[str, Any]: return { "schema": "state-hub.forge-projection.v1", "repo_slug": self.repo_slug, # Provenance is not optional: a projection that cannot name the # commit it came from cannot be audited (ADR-012 decision 2). "commit": self.commit, "workplans": [ { "record_id": w.record_id, "uuid": w.uuid, "title": w.title, "status": w.status, "relative_path": w.relative_path, "archived": w.archived, "tasks": [ { "record_id": t.record_id, "uuid": t.uuid, "title": t.title, "status": t.status, "priority": t.priority, } for t in w.tasks ], } for w in self.workplans ], } def _run_git(*args: str, cwd: str | None = None, timeout: float = 120.0) -> str: proc = subprocess.run( ["git", *args], cwd=cwd, capture_output=True, text=True, timeout=timeout ) if proc.returncode != 0: raise ForgeDeriveError((proc.stderr or proc.stdout).strip()[:400]) return proc.stdout.strip() def _split_frontmatter(text: str) -> tuple[dict, str]: if not text.startswith("---"): return {}, text end = text.find("\n---", 3) if end == -1: return {}, text raw = text[3:end] body = text[end + 4 :] try: meta = yaml.safe_load(raw) or {} except yaml.YAMLError: return {}, text return (meta if isinstance(meta, dict) else {}), body def _parse_tasks(body: str) -> list[DerivedTask]: headings = [ (m.start(), m.group(2).strip()) for m in _HEADING_RE.finditer(body) ] out: list[DerivedTask] = [] for m in _TASK_BLOCK_RE.finditer(body): try: block = yaml.safe_load(m.group(1).strip()) or {} except yaml.YAMLError: continue if not isinstance(block, dict): continue rid = str(block.get("id") or "").strip() if not rid: continue title = block.get("title") if not title: prev = [t for pos, t in headings if pos < m.start()] title = prev[-1] if prev else None out.append( DerivedTask( record_id=rid, # Derived, not read from the file: the forge projection must not # inherit an identifier the file happens to carry. uuid=derived_record_uuid(rid), title=title, status=(str(block["status"]).strip() if block.get("status") else None), priority=(str(block["priority"]).strip() if block.get("priority") else None), ) ) return out def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> DerivedProjection: """Derive a projection from an already-materialised checkout.""" proj = DerivedProjection(repo_slug=repo_slug, commit=commit) wp_dir = repo_root / "workplans" if not wp_dir.is_dir(): return proj for path in sorted(wp_dir.rglob("*.md")): if path.name.startswith("."): continue try: text = path.read_text(encoding="utf-8") except (OSError, UnicodeDecodeError): continue meta, body = _split_frontmatter(text) if str(meta.get("type") or "").strip() != "workplan": continue rid = str(meta.get("id") or "").strip() if not rid: continue proj.workplans.append( DerivedWorkplan( record_id=rid, uuid=derived_record_uuid(rid), title=(str(meta["title"]).strip() if meta.get("title") else None), status=(str(meta["status"]).strip() if meta.get("status") else None), relative_path=str(path.relative_to(repo_root).as_posix()), archived=path.parent.name == "archived", tasks=_parse_tasks(body), ) ) proj.workplans.sort(key=lambda w: w.record_id) return proj def derive_from_forge( repo_slug: str, *, forge_base: str = DEFAULT_FORGE_BASE, ref: str | None = None ) -> DerivedProjection: """Clone the repository's default branch from the forge and derive from it. A fresh shallow clone every time, deliberately: the source is what the forge holds now, and a reused working copy is how a projection ends up reflecting someone's local state instead (ADR-012 context). """ url = f"{forge_base.rstrip('/')}/{repo_slug}.git" with tempfile.TemporaryDirectory(prefix=f"forge-{repo_slug}-") as tmp: args = ["clone", "--depth", "1", "--quiet"] if ref: args += ["--branch", ref] try: _run_git(*args, url, tmp) except subprocess.TimeoutExpired as exc: raise ForgeDeriveError(f"clone timed out for {repo_slug}") from exc commit = _run_git("rev-parse", "HEAD", cwd=tmp) return derive_from_checkout(Path(tmp), repo_slug, commit) # --------------------------------------------------------------------------- # Comparison against what the hub currently holds (STATE-WP-0083-T02) # --------------------------------------------------------------------------- _ARCHIVE_PREFIX_RE = re.compile(r"^\d{6}-") def _path_key(path: str) -> str: """Normalise a workplan path so an archived copy matches its live one.""" name = path.rsplit("/", 1)[-1] return _ARCHIVE_PREFIX_RE.sub("", name).strip().lower() @dataclass class ProjectionDiff: """What a reset would change, computed without changing anything. This is what makes `ADR-012` decision 7's "verifiable" real: the reset can always be inspected before it runs, and its result compared against the forge afterwards. """ repo_slug: str commit: str missing: list[dict[str, Any]] = field(default_factory=list) # forge has, hub lacks stale: list[dict[str, Any]] = field(default_factory=list) # hub has, forge lacks differing: list[dict[str, Any]] = field(default_factory=list) # both, fields differ @property def clean(self) -> bool: return not (self.missing or self.stale or self.differing) @property def would_remove(self) -> int: return len(self.stale) def to_dict(self) -> dict[str, Any]: return { "schema": "state-hub.projection-diff.v1", "repo_slug": self.repo_slug, "commit": self.commit, "clean": self.clean, "counts": { "missing": len(self.missing), "stale": len(self.stale), "differing": len(self.differing), }, "missing": self.missing, "stale": self.stale, "differing": self.differing, } def diff_against_hub( derived: DerivedProjection, hub_workplans: list[dict[str, Any]], hub_tasks_by_workplan: dict[str, list[dict[str, Any]]], ) -> ProjectionDiff: """Compare a derived projection with the hub's current records. Pure: takes the hub's state as data rather than reading it, so the comparison is testable without a database and cannot accidentally mutate anything. """ d = ProjectionDiff(repo_slug=derived.repo_slug, commit=derived.commit) # Match on canonical identity, never on UUID. Most hub records still carry # pre-ADR-007 random identifiers, so a UUID-keyed comparison reports every # record as simultaneously missing and stale — and a reset built on that # would destroy and recreate the entire projection. The canonical record id # is what is stable across the identifier migration; the backing file is the # fallback when a hub slug was derived from a filename rather than an id. def _wp_key(record_id: str | None, slug: str | None, path: str | None) -> str: if record_id: return record_id.strip().lower() if path: return _path_key(path) return (slug or "").strip().lower() want_wp = {w.record_id.strip().lower(): w for w in derived.workplans} have_wp: dict[str, dict[str, Any]] = {} want_paths = {_path_key(w.relative_path): k for k, w in want_wp.items()} for w in hub_workplans: slug = str(w.get("slug") or "") key = slug.strip().lower() if key not in want_wp: # Slugs were not always the canonical id; fall back to the file. bp = w.get("backing_relative_path") if bp and _path_key(bp) in want_paths: key = want_paths[_path_key(bp)] else: cand = [k for k in want_wp if slug.lower().startswith(k + "-")] if len(cand) == 1: key = cand[0] have_wp[key] = w for key, w in want_wp.items(): if key not in have_wp: d.missing.append({"kind": "workplan", "record_id": w.record_id, "uuid": w.uuid}) for key, w in have_wp.items(): if key not in want_wp: uid = str(w["id"]) d.stale.append( { "kind": "workplan", "uuid": uid, "slug": w.get("slug"), "status": w.get("status"), # A stale record with no backing file is the case that must # never be destroyed silently (ADR-012 decision 7). "has_backing_file": bool( w.get("backing_filename") or w.get("backing_relative_path") ), } ) for key, w in want_wp.items(): cur = have_wp.get(key) uid = str(cur["id"]) if cur else w.uuid if not cur: continue changed = {} if (cur.get("status") or None) != (w.status or None): changed["status"] = {"hub": cur.get("status"), "forge": w.status} cur_path = cur.get("backing_relative_path") or None if cur_path != w.relative_path: changed["backing_relative_path"] = {"hub": cur_path, "forge": w.relative_path} if changed: d.differing.append( {"kind": "workplan", "record_id": w.record_id, "uuid": uid, "changed": changed} ) for w in derived.workplans: hub_rows = hub_tasks_by_workplan.get(w.uuid, []) if not hub_rows: cur = have_wp.get(w.record_id.strip().lower()) if cur: hub_rows = hub_tasks_by_workplan.get(str(cur["id"]), []) # Tasks carry no canonical id on the hub, only a title, so compare on # title. Imperfect, and the reason task removal needs the same explicit # acknowledgement as everything else. have = {(t.get("title") or "").strip().lower(): t for t in hub_rows} want = {(t.title or t.record_id).strip().lower(): t for t in w.tasks} for key, t in want.items(): if key not in have: d.missing.append( {"kind": "task", "record_id": t.record_id, "uuid": t.uuid, "workplan": w.record_id} ) for key, t in have.items(): if key not in want: uid = str(t["id"]) d.stale.append( {"kind": "task", "uuid": uid, "title": t.get("title"), "status": t.get("status"), "workplan": w.record_id, "has_backing_file": False} ) for key, t in want.items(): cur = have.get(key) if cur and (cur.get("status") or None) != (t.status or None): d.differing.append( {"kind": "task", "record_id": t.record_id, "uuid": t.uuid, "workplan": w.record_id, "changed": {"status": {"hub": cur.get("status"), "forge": t.status}}} ) return d