diff --git a/.gitignore b/.gitignore
index d924502..e71a2dd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,11 +7,15 @@
 !.polytoken
 .playwright-mcp
 node_modules/
+**/__pycache__/
+*.py[cod]
 .polytoken/permissions.local.yaml
 /e2e/test-results/**
 /e2e/playwright-report/**
 /docs/**
 .scratch
+# Runtime coordination state for the event-sourced GC executor.
+/.current-gc/
 # Generated by build.rs from POLYMODEL_* env vars (WASM bundle config).
 /src/env.rs
 # Generated renderer worker artifacts (built by `just build-renderer-worker`).
diff --git a/justfile b/justfile
index 3ec940d..d3f58bb 100644
--- a/justfile
+++ b/justfile
@@ -109,6 +109,14 @@ verify-renderer-split: build-web
 # Regenerate crates/polymodel-api from authored lexicons via the local Jacquard checkout.
 generate-api:
     nix run ../jacquard
+# Validate portable event-sourced GC coordination state.
+validate-gc-state *ARGS:
+    python3 tools/validate-gc-state.py {{ ARGS }}
+
+# Validate GC jj workspace/bookmark rails before or after forgetting workspaces.
+validate-gc-jj-state *ARGS:
+    python3 tools/validate-gc-jj-state.py {{ ARGS }}
+
 # Create a sibling jj workspace for a Jira ticket or feature slug, then start a fresh change.
 workspace-create name:
     root="$(jj workspace root)" && workspace_path="$(dirname "$root")/{{ name }}" && jj workspace add "$workspace_path" -r @
diff --git a/tools/validate-gc-jj-state.py b/tools/validate-gc-jj-state.py
new file mode 100755
index 0000000..54cf59d
--- /dev/null
+++ b/tools/validate-gc-jj-state.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+"""Validate the GC workspace rail before and after final bookmark cleanup."""
+from __future__ import annotations
+
+import argparse
+import subprocess
+import sys
+
+RUN_ID = "gc-2026-08-01-catchup"
+PARENT = "gc-2026-08-01-catchup"
+CHILDREN = [f"{PARENT}-task-{unit}" for unit in "ABCDEFG"]
+
+
+def jj(*args: str) -> str:
+    result = subprocess.run(["jj", *args], text=True, capture_output=True)
+    if result.returncode:
+        raise RuntimeError(result.stderr.strip() or result.stdout.strip())
+    return result.stdout
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("run_id", nargs="?", default=RUN_ID)
+    parser.add_argument("--parent", default=PARENT)
+    parser.add_argument("--pre-forget", action="store_true")
+    parser.add_argument("--post-forget", action="store_true")
+    parser.add_argument("--self-test", action="store_true")
+    args = parser.parse_args()
+    try:
+        if args.run_id != RUN_ID or args.parent != PARENT:
+            raise RuntimeError("unexpected run id or parent")
+        if args.self_test:
+            print("validate-gc-jj-state self-test passed")
+            return 0
+        workspaces = jj("workspace", "list")
+        if args.pre_forget:
+            for name in [PARENT, *CHILDREN]:
+                if name not in workspaces:
+                    raise RuntimeError(f"missing expected workspace {name}")
+            if PARENT not in jj("bookmark", "list"):
+                raise RuntimeError(f"missing parent bookmark {PARENT}")
+            print(f"JJ pre-forget state valid: {RUN_ID}")
+        elif args.post_forget:
+            for name in [PARENT, *CHILDREN]:
+                if name in workspaces:
+                    raise RuntimeError(f"workspace still present after forget: {name}")
+            print(f"JJ post-forget state valid: {RUN_ID}")
+        else:
+            raise RuntimeError("choose --pre-forget or --post-forget")
+    except (RuntimeError, FileNotFoundError) as exc:
+        print(f"JJ state invalid: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tools/validate-gc-state.py b/tools/validate-gc-state.py
new file mode 100755
index 0000000..dffc6b1
--- /dev/null
+++ b/tools/validate-gc-state.py
@@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+"""Validate portable state for an event-sourced garbage-collection run."""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+RUN_ID = "gc-2026-08-01-catchup"
+UNITS = set("ABCDEFG")
+CLUSTERS = {f"C{i}" for i in range(1, 14)}
+REQUIRED = ("manifest.txt", "scope.md", "workflow.dot", "events.log", "gradebook.jsonl", "clusters.jsonl", "deferrals.jsonl", "todos.jsonl")
+EVENTS = {"setup", "raw_discovered", "clustered", "verified", "rejected", "needs_more_evidence", "cleanup_ready", "planned", "in_progress", "cleaned", "deferred", "blocked", "ejected_from_gc", "accepted_risk", "stale_expired"}
+TERMINAL = {"done", "blocked", "deferred", "rejected", "ejected_from_gc", "partially_cleaned"}
+
+
+def fail(message: str) -> None:
+    raise ValueError(message)
+
+
+def read_jsonl(path: Path) -> list[dict[str, Any]]:
+    if not path.exists():
+        fail(f"missing {path}")
+    records: list[dict[str, Any]] = []
+    for line_no, line in enumerate(path.read_text().splitlines(), 1):
+        if not line.strip():
+            continue
+        try:
+            value = json.loads(line)
+        except json.JSONDecodeError as exc:
+            fail(f"{path}:{line_no}: invalid JSON: {exc}")
+        if not isinstance(value, dict):
+            fail(f"{path}:{line_no}: record is not an object")
+        records.append(value)
+    return records
+
+
+def validate_event(record: dict[str, Any], source: str) -> None:
+    for field in ("schema_version", "run_id", "event", "at", "by", "notes"):
+        if field not in record:
+            fail(f"{source}: event missing {field}")
+    if record["schema_version"] != 1 or record["run_id"] != RUN_ID:
+        fail(f"{source}: wrong schema version or run id")
+    if record["event"] not in EVENTS:
+        fail(f"{source}: unknown event {record['event']!r}")
+    if not isinstance(record["at"], str) or not record["at"]:
+        fail(f"{source}: event has no timestamp")
+    if not isinstance(record["by"], str) or not record["by"]:
+        fail(f"{source}: event has no actor/session")
+    if not isinstance(record["notes"], str) or not record["notes"]:
+        fail(f"{source}: event has no notes")
+    if "cluster_id" in record and record["cluster_id"] not in CLUSTERS:
+        fail(f"{source}: unknown cluster {record['cluster_id']!r}")
+    if "cleanup_unit" in record and record["cleanup_unit"] not in UNITS:
+        fail(f"{source}: unknown cleanup unit {record['cleanup_unit']!r}")
+    if record["event"] in {"blocked", "deferred", "rejected", "ejected_from_gc"} and not record.get("evidence"):
+        fail(f"{source}: {record['event']} needs evidence")
+    if record["event"] in {"cleaned", "partially_cleaned"} and not record.get("evidence"):
+        fail(f"{source}: terminal cleanup event needs evidence")
+
+
+def validate_state(root: Path, require_dispositions: bool, require_reviews: bool) -> None:
+    state = root / ".current-gc"
+    if not state.is_dir():
+        fail(".current-gc is missing")
+    for rel in REQUIRED:
+        if not (state / rel).is_file():
+            fail(f"missing .current-gc/{rel}")
+    for unit in UNITS:
+        for rel in (f"cleanup-units/{unit}.md", f"plans/plan-{unit}.md", f"findings/findings-{unit}.jsonl"):
+            if not (state / rel).is_file():
+                fail(f"missing .current-gc/{rel}")
+    if not (state / "findings/findings-final.jsonl").is_file():
+        fail("missing .current-gc/findings/findings-final.jsonl")
+    gitignore = (root / ".gitignore").read_text() if (root / ".gitignore").exists() else ""
+    if not re.search(r"(?m)^/?\.current-gc/?$", gitignore):
+        fail(".gitignore does not contain the exact .current-gc/ rule")
+    if f"run_id={RUN_ID}" not in (state / "manifest.txt").read_text():
+        fail("manifest does not name the run id")
+    events: list[dict[str, Any]] = []
+    for rel in ("events.log", "gradebook.jsonl"):
+        records = read_jsonl(state / rel)
+        for index, record in enumerate(records, 1):
+            validate_event(record, f".current-gc/{rel}:{index}")
+        events.extend(records)
+    clusters = read_jsonl(state / "clusters.jsonl")
+    seen_clusters: set[str] = set()
+    for index, record in enumerate(clusters, 1):
+        if record.get("run_id") != RUN_ID or record.get("schema_version") != 1:
+            fail(f"clusters.jsonl:{index}: wrong run metadata")
+        cluster = record.get("cluster_id")
+        if cluster not in CLUSTERS:
+            fail(f"clusters.jsonl:{index}: invalid cluster")
+        seen_clusters.add(cluster)
+    if seen_clusters != CLUSTERS:
+        fail(f"clusters.jsonl does not cover all clusters: {sorted(CLUSTERS - seen_clusters)}")
+    deferrals = read_jsonl(state / "deferrals.jsonl")
+    for index, record in enumerate(deferrals, 1):
+        for field in ("cluster_id", "disposition", "rationale", "evidence", "follow_up_condition", "owner"):
+            if not record.get(field):
+                fail(f"deferrals.jsonl:{index}: missing {field}")
+        if record["cluster_id"] not in CLUSTERS or record["owner"] != "future implementation":
+            fail(f"deferrals.jsonl:{index}: invalid deferral")
+    if not {"C3", "C4", "C7", "C12"}.issubset({record["cluster_id"] for record in deferrals}):
+        fail("required C3/C4/C7/C12 deferrals are missing")
+    todos = read_jsonl(state / "todos.jsonl")
+    if not todos:
+        fail("todos.jsonl is empty")
+    for index, record in enumerate(todos, 1):
+        for field in ("id", "title", "status", "dependencies", "checkpoint_evidence"):
+            if field not in record:
+                fail(f"todos.jsonl:{index}: missing {field}")
+        if record["status"] not in TERMINAL | {"pending", "in_progress"}:
+            fail(f"todos.jsonl:{index}: invalid status")
+    if require_dispositions:
+        event_clusters = {record.get("cluster_id") for record in events}
+        for cluster in ("C3", "C4", "C5", "C7", "C12", "C13"):
+            if cluster not in event_clusters:
+                fail(f"required disposition event missing for {cluster}")
+        dispositions = {record.get("cluster_id"): record["event"] for record in events if record["event"] in {"deferred", "blocked", "rejected", "ejected_from_gc"}}
+        for cluster in ("C3", "C4", "C7", "C12"):
+            if dispositions.get(cluster) not in {"deferred", "blocked"}:
+                fail(f"{cluster} has no deferred/blocked disposition")
+        if dispositions.get("C5") not in {"rejected", "ejected_from_gc"} or dispositions.get("C13") not in {"rejected", "ejected_from_gc"}:
+            fail("C5/C13 must be rejected or ejected")
+    if require_reviews:
+        findings = []
+        for rel in [f"findings/findings-{unit}.jsonl" for unit in UNITS] + ["findings/findings-final.jsonl"]:
+            findings.extend(read_jsonl(state / rel))
+        if not findings:
+            fail("review records are required")
+        for index, finding in enumerate(findings, 1):
+            for field in ("finding_id", "severity", "disposition", "resolution"):
+                if not finding.get(field):
+                    fail(f"review record {index} missing {field}")
+            if finding["severity"] in {"Critical", "High"} and finding["disposition"] not in {"fixed", "rebutted", "accepted_risk"}:
+                fail(f"unresolved {finding['severity']} finding {finding['finding_id']}")
+        if any(record["status"] not in TERMINAL for record in todos):
+            fail("all todos must be terminal for --require-review-records")
+    print(f"GC state valid: {RUN_ID}")
+
+
+def self_test() -> None:
+    with tempfile.TemporaryDirectory() as directory:
+        root = Path(directory)
+        state = root / ".current-gc"
+        state.mkdir()
+        (root / ".gitignore").write_text(".current-gc/\n")
+        for rel in REQUIRED:
+            (state / rel).write_text(f"run_id={RUN_ID}\n" if rel == "manifest.txt" else "")
+        for unit in UNITS:
+            for rel in (f"cleanup-units/{unit}.md", f"plans/plan-{unit}.md", f"findings/findings-{unit}.jsonl"):
+                path = state / rel
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_text("" if path.suffix == ".jsonl" else "fixture")
+        final = state / "findings/findings-final.jsonl"
+        final.parent.mkdir(parents=True, exist_ok=True)
+        final.write_text("")
+        event = {"schema_version": 1, "run_id": RUN_ID, "event": "verified", "at": "2026-08-01T00:00:00Z", "by": "self-test", "notes": "fixture"}
+        (state / "events.log").write_text(json.dumps(event) + "\n")
+        (state / "gradebook.jsonl").write_text(json.dumps(event) + "\n")
+        (state / "clusters.jsonl").write_text("\n".join(json.dumps({"schema_version": 1, "run_id": RUN_ID, "cluster_id": f"C{i}", "disposition": "verified"}) for i in range(1, 14)) + "\n")
+        (state / "deferrals.jsonl").write_text("\n".join(json.dumps({"cluster_id": cluster, "disposition": "deferred", "rationale": "fixture", "evidence": ["fixture"], "follow_up_condition": "future", "owner": "future implementation"}) for cluster in ("C3", "C4", "C7", "C12")) + "\n")
+        (state / "todos.jsonl").write_text(json.dumps({"id": "1", "title": "fixture", "status": "pending", "dependencies": [], "checkpoint_evidence": ["fixture"]}) + "\n")
+        validate_state(root, False, False)
+        (state / "events.log").write_text("not json\n")
+        try:
+            validate_state(root, False, False)
+        except ValueError:
+            pass
+        else:
+            fail("negative fixture unexpectedly passed")
+        print("validate-gc-state self-test passed")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--self-test", action="store_true")
+    parser.add_argument("--require-dispositions", action="store_true")
+    parser.add_argument("--require-review-records", action="store_true")
+    parser.add_argument("run_id", nargs="?", default=RUN_ID)
+    args = parser.parse_args()
+    try:
+        if args.run_id != RUN_ID:
+            fail(f"unsupported run id {args.run_id}")
+        if args.self_test:
+            self_test()
+        else:
+            validate_state(Path.cwd(), args.require_dispositions, args.require_review_records)
+    except ValueError as exc:
+        print(f"GC state invalid: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())