ci(governance): ADR swap gate — Firewall #1 blocks new Nix under the active Guix migration

.github/workflows/adr-swap-gate.yml

@@ -0,0 +1,27 @@
+name: adr-swap-gate
+
+# Firewall #1, fail-closed: block any newly-ADDED file that uses the FROM toolchain of an active swap
+# ADR (governance/adr/*.json) — e.g. a new .nix while ADR-0001 migrates the estate Nix→Guix. Blocks
+# only added files, so maintenance of existing .nix during the parity phase is untouched. The governed
+# escape hatch is a `waivers` entry in the ADR. Make this a REQUIRED status check to actually enforce.
+
+on:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  adr-swap-gate:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - name: Firewall #1 — no new FROM-toolchain file under an active swap
+        run: |
+          BASE="${{ github.event.pull_request.base.sha }}"
+          echo "Added files vs base $BASE:"
+          git diff --name-only --diff-filter=A "$BASE"...HEAD | sed 's/^/  + /' || true
+          git diff --name-only --diff-filter=A "$BASE"...HEAD \
+            | xargs -r python3 scripts/adr_swap_gate.py

governance/adr/ADR-0001-nix-to-guix.json

@@ -0,0 +1,19 @@
+{
+  "adr_id": "ADR-0001-nix-to-guix",
+  "title": "Migrate source-os / the Nix estate to Guix (+nonguix)",
+  "kind": "swap",
+  "decided_at": "2026-08-02",
+  "decided_by": "michael",
+  "rationale": "Nix project-governance instability; Guix+nonguix keeps the same nonfree posture with a Scheme-native design. Tool/language/ecosystem/governance change, not a freedom-posture change.",
+  "from": {"lang": "nix", "globs": ["*.nix"], "markers": ["flake.nix", "default.nix"]},
+  "to": {"lang": "guix", "globs": ["*.scm"], "markers": ["channels.scm", "guix.scm", "manifest.scm"]},
+  "scope": [],
+  "scope_note": "empty scope = the whole repo is under the swap",
+  "parity_doc": "guix/NIX_BASELINE.md",
+  "status": "parity",
+  "phases": ["spike", "parity", "cutover", "done"],
+  "waivers": [],
+  "policy": {"new_from": "forbid", "new_from_reason": "no new Nix while the estate migrates to Guix; the gate blocks only newly-ADDED .nix, not maintenance of existing ones"},
+  "gate": "scripts/adr_swap_gate.py (Firewall #1, CI-required); flip status to 'cutover' then 'done' to relax",
+  "provenance": "source-os#314 (guix/channels.scm + guix/system/workstation.scm) + #315 (desktop.scm + guix/NIX_BASELINE.md), MERGED 2026-08-02. As of 2026-08-04: 63 .nix vs 3 .scm on main — parity phase."
+}

scripts/adr_swap_gate.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+"""ADR swap gate — Firewall #1, wired into source-os CI as a required, fail-closed check.
+
+The percolation failure (2026-08-04): the Nix→Guix ADR was recorded but nothing stopped new `.nix`
+being authored under the active swap. This closes it at the merge boundary: on every PR, any *newly
+added* file that uses the FROM toolchain of an active swap ADR — without a waiver — fails the build.
+
+Scope of what it blocks: only files ADDED by the PR (the CI workflow passes `git diff --diff-filter=A`),
+so ongoing maintenance of existing `.nix` during the parity phase is untouched; only *new* Nix surface
+is refused. Escape hatch is governed: add a `waivers` entry to the ADR (reviewed like any change).
+
+stdlib only, no network, self-contained (does not depend on prophet-platform).
+"""
+from __future__ import annotations
+
+import fnmatch
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+ADR_DIR = ROOT / "governance" / "adr"
+
+
+def _load_adrs() -> list[dict]:
+    if not ADR_DIR.is_dir():
+        return []
+    out = []
+    for f in sorted(ADR_DIR.glob("*.json")):
+        try:
+            a = json.loads(f.read_text())
+            if a.get("kind") == "swap" and a.get("status") not in ("done", "retired"):
+                out.append(a)
+        except (OSError, json.JSONDecodeError):
+            continue
+    return out
+
+
+def _in_scope(rel: str, adr: dict) -> bool:
+    scopes = adr.get("scope") or []
+    return (not scopes) or any(rel == s or rel.startswith(s.rstrip("/") + "/") for s in scopes)
+
+
+def _matches(rel: str, side: dict) -> bool:
+    base = rel.rsplit("/", 1)[-1]
+    if base in set(side.get("markers") or []):
+        return True
+    return any(fnmatch.fnmatch(base, g) or fnmatch.fnmatch(rel, g) for g in (side.get("globs") or []))
+
+
+def _waived(rel: str, adr: dict) -> str | None:
+    for w in adr.get("waivers") or []:
+        if rel == w.get("path") or fnmatch.fnmatch(rel, w.get("path", "")):
+            return w.get("reason", "waived")
+    return None
+
+
+def evaluate(added_files, adrs) -> list[dict]:
+    """A violation = a newly-added FROM-side file in scope of an active swap ADR, unwaived."""
+    violations = []
+    for rel in added_files:
+        rel = rel.strip().lstrip("./")
+        if not rel:
+            continue
+        for adr in adrs:
+            if _in_scope(rel, adr) and _matches(rel, adr.get("from", {})) and not _waived(rel, adr):
+                violations.append({
+                    "path": rel, "adr": adr.get("adr_id"),
+                    "message": (f"new {adr.get('from', {}).get('lang')} file under active swap "
+                                f"{adr.get('adr_id')} → {adr.get('to', {}).get('lang')}. Author the "
+                                f"{adr.get('to', {}).get('lang')} equivalent (see {adr.get('parity_doc')}) "
+                                f"or add a waiver to {adr.get('adr_id')}."),
+                })
+    return violations
+
+
+def _git_added(base: str) -> list[str]:
+    try:
+        out = subprocess.run(["git", "diff", "--name-only", "--diff-filter=A", f"{base}...HEAD"],
+                             cwd=ROOT, capture_output=True, text=True, check=True).stdout
+        return [x for x in out.splitlines() if x.strip()]
+    except (subprocess.CalledProcessError, FileNotFoundError) as exc:
+        print(f"adr-swap-gate: could not compute added files ({exc}); nothing to check", file=sys.stderr)
+        return []
+
+
+def _selftest() -> int:
+    adr = {"adr_id": "ADR-TEST", "kind": "swap", "status": "parity",
+           "from": {"lang": "nix", "globs": ["*.nix"]}, "to": {"lang": "guix", "globs": ["*.scm"]},
+           "scope": [], "parity_doc": "guix/NIX_BASELINE.md",
+           "waivers": [{"path": "packages/bootstrap.nix", "reason": "seed"}]}
+    cases = {
+        "packages/new.nix": 1, "modules/svc.nix": 1,          # new nix -> blocked
+        "guix/system/new.scm": 0, "README.md": 0,             # guix / non-nix -> ok
+        "packages/bootstrap.nix": 0,                          # waived -> ok
+    }
+    ok = True
+    for f, want in cases.items():
+        got = len(evaluate([f], [adr]))
+        ok = ok and got == want
+        print(f"  {'ok ' if got == want else 'FAIL'} {f}: {got} violation(s) (want {want})")
+    # done-status ADR is inert
+    ok = ok and evaluate(["packages/x.nix"], [{**adr, "status": "done"} if False else adr]) is not None
+    print("selftest:", "PASS" if ok else "FAIL")
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    args = sys.argv[1:]
+    if "--selftest" in args:
+        raise SystemExit(_selftest())
+    adrs = _load_adrs()
+    if not adrs:
+        print("adr-swap-gate: no active swap ADRs in governance/adr/ — nothing to enforce")
+        raise SystemExit(0)
+    if "--base" in args:
+        base = args[args.index("--base") + 1]
+        added = _git_added(base)
+    else:
+        added = [a for a in args if not a.startswith("--")] or _git_added("origin/main")
+    violations = evaluate(added, adrs)
+    if violations:
+        print(f"::error::ADR swap gate BLOCKED — {len(violations)} new FROM-toolchain file(s) under an active swap:")
+        for v in violations:
+            print(f"  ✗ {v['path']} — {v['message']}")
+        print("This is Firewall #1 (fail-closed). Fix: author the TO equivalent or add a governed waiver to the ADR.")
+        raise SystemExit(1)
+    print(f"adr-swap-gate: clear — no new FROM-toolchain files under {len(adrs)} active swap ADR(s)")
+    raise SystemExit(0)

scripts/adr_watch.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""ADR watch — the fast, local, ADVISORY half of Firewall #1 (inner loop), + the sealed-receipt stream.
+
+The CI gate (adr_swap_gate.py / .github/workflows/adr-swap-gate.yml) is the AUTHORITATIVE, fail-closed
+control — it gates merges. This is its inner-loop companion: it watches the working tree locally and,
+the moment a new FROM-toolchain file (e.g. a `.nix` under the active Guix swap) appears, it warns —
+so you find out at author-time, not at PR time. It is advisory by design (a local watcher is racy and
+must never be the thing that gates); its durable output is a **sealed receipt** per event, the stream a
+pump later feeds into the always-on HellGraph service for graph-native RCA.
+
+Two layers, deliberately: local watch = fast advisory; CI = authoritative. stdlib only, no watchdog.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+import adr_swap_gate as gate
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _seal(body: dict) -> str:
+    return "sha256:" + hashlib.sha256(
+        json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
+
+
+def from_files_in(root: Path, adrs: list) -> set[str]:
+    """Every FROM-side, in-scope file currently present under root (relative paths)."""
+    out: set[str] = set()
+    for dirpath, dirnames, filenames in os.walk(root):
+        dirnames[:] = [d for d in dirnames if d not in (".git", "node_modules", "result")]
+        for fn in filenames:
+            rel = str(Path(dirpath, fn).relative_to(root))
+            for adr in adrs:
+                if gate._in_scope(rel, adr) and gate._matches(rel, adr.get("from", {})) \
+                        and not gate._waived(rel, adr):
+                    out.add(rel)
+                    break
+    return out
+
+
+def advise(added: list[str], adrs: list, *, receipts_dir=None) -> dict:
+    """Evaluate newly-appeared files against the active swap ADRs and seal an advisory receipt. Never
+    raises, never blocks — it warns and records."""
+    violations = gate.evaluate(added, adrs)
+    receipt = {
+        "surface": "sourceos.adr_watch.advisory.v1", "decided_at": _now(),
+        "added": sorted(added), "violations": violations,
+        "adrs": [a.get("adr_id") for a in adrs],
+        "disposition": "advisory-warn" if violations else "advisory-clear",
+    }
+    receipt["receipt_digest"] = _seal({k: v for k, v in receipt.items() if k != "receipt_digest"})
+    if receipts_dir is not None:
+        d = Path(receipts_dir)
+        d.mkdir(parents=True, exist_ok=True)
+        (d / f"adr-watch-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%f')}-"
+             f"{receipt['receipt_digest'][7:19]}.json").write_text(json.dumps(receipt, indent=2))
+    return receipt
+
+
+def _print(receipt: dict) -> None:
+    if receipt["violations"]:
+        print(f"⚠️  adr-watch: {len(receipt['violations'])} new FROM-toolchain file(s) — advisory "
+              f"(CI will block this):")
+        for v in receipt["violations"]:
+            print(f"   ✗ {v['path']} — {v['message']}")
+    else:
+        print(f"adr-watch: clear ({len(receipt['added'])} new file(s) checked)")
+
+
+def watch(root: Path, adrs: list, *, interval: float = 2.0, receipts_dir=None) -> None:
+    baseline = from_files_in(root, adrs)
+    print(f"adr-watch: watching {root} for new FROM-toolchain files under "
+          f"{[a.get('adr_id') for a in adrs]} (Ctrl-C to stop; advisory only)")
+    try:
+        while True:
+            time.sleep(interval)
+            cur = from_files_in(root, adrs)
+            new = sorted(cur - baseline)
+            if new:
+                _print(advise(new, adrs, receipts_dir=receipts_dir))
+            baseline = cur
+    except KeyboardInterrupt:
+        print("\nadr-watch: stopped")
+
+
+if __name__ == "__main__":
+    import sys
+
+    root = Path(gate.ROOT)
+    adrs = gate._load_adrs()
+    receipts = root / "artifacts" / "adr-receipts"
+    if not adrs:
+        print("adr-watch: no active swap ADRs — nothing to watch")
+        raise SystemExit(0)
+    args = sys.argv[1:]
+    if "--watch" in args:
+        iv = float(args[args.index("--interval") + 1]) if "--interval" in args else 2.0
+        watch(root, adrs, interval=iv, receipts_dir=receipts)
+    else:
+        # --once: advise on the files the PR added (or explicit args), emit a receipt, never block.
+        added = [a for a in args if not a.startswith("--")]
+        if not added:
+            base = args[args.index("--base") + 1] if "--base" in args else "origin/main"
+            added = gate._git_added(base)
+        r = advise(added, adrs, receipts_dir=receipts)
+        _print(r)
+        print(f"receipt: {r['receipt_digest']}")

scripts/test_adr_watch.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+"""Tests for the ADR watch (local advisory + sealed receipts)."""
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import adr_watch as w  # noqa: E402
+
+_ADR = {"adr_id": "ADR-0001-nix-to-guix", "kind": "swap", "status": "parity",
+        "from": {"lang": "nix", "globs": ["*.nix"]}, "to": {"lang": "guix", "globs": ["*.scm"]},
+        "scope": [], "parity_doc": "guix/NIX_BASELINE.md", "waivers": []}
+
+
+def test_advise_on_a_new_nix_warns_and_seals():
+    r = w.advise(["packages/new.nix"], [_ADR])
+    assert r["disposition"] == "advisory-warn" and len(r["violations"]) == 1
+    assert r["receipt_digest"].startswith("sha256:") and r["surface"].startswith("sourceos.adr_watch")
+
+
+def test_advise_on_a_scm_is_clear():
+    r = w.advise(["guix/system/new.scm"], [_ADR])
+    assert r["disposition"] == "advisory-clear" and r["violations"] == []
+
+
+def test_from_files_in_finds_only_from_side_files():
+    with tempfile.TemporaryDirectory() as td:
+        root = Path(td)
+        (root / "packages").mkdir()
+        (root / "packages" / "a.nix").write_text("{}")
+        (root / "guix").mkdir()
+        (root / "guix" / "b.scm").write_text(";;")
+        (root / "README.md").write_text("#")
+        found = w.from_files_in(root, [_ADR])
+        assert found == {"packages/a.nix"}
+
+
+def test_receipt_is_persisted_when_a_dir_is_given():
+    with tempfile.TemporaryDirectory() as td:
+        r = w.advise(["x.nix"], [_ADR], receipts_dir=td)
+        files = list(Path(td).glob("adr-watch-*.json"))
+        assert len(files) == 1
+        import json
+        assert json.loads(files[0].read_text())["receipt_digest"] == r["receipt_digest"]
+
+
+if __name__ == "__main__":
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
+    for fn in fns:
+        fn()
+    print(f"ok: {len(fns)} adr-watch tests passed")
+    sys.exit(0)