feat(exclusion): resume-under-exclusion + post-hoc re-verifier (finish Phase 1)

CLAUDE.md

@@ -90,6 +90,7 @@ python archive.py --globus /path/to/configs/config_<name>.json
 archive.py               # Single-file script — all archive logic lives here
 archive_local.sbatch     # Default path: runs archive.py (local mode) as a Slurm job
 reclaim.py               # Inverse of archive.py: verify archives, reclaim source space
+verify_exclusions.py     # Post-hoc re-verifier: prove archive-time exclusions are still git-covered
 reclaim_holds.example.txt# Template holds file (pass a real one via --holds)
 config.example.json      # Annotated config template
 setup_keyring.py         # One-time credential helper (legacy — now uses personal Globus identity)
@@ -296,9 +297,20 @@ metadata-only change reads `SAFE` (no false DRIFT → the backup skips the re-sh
 this is what actually delivers the byte savings); an excluded file that is no longer
 `COVERED` (edited/regenerated since the archive, so its current bytes aren't on tape)
 reads `DRIFTED`, never a silent `--delete`. Logs without an `exclusion` block behave
-exactly as before. Still pending: the standalone post-hoc re-verifier, and
-resume-under-exclusion (the archive currently forces a fresh zip when an
-`exclude_spec` is set).
+exactly as before.
+
+**Resume under exclusion** is honored: the resume staleness guard uses
+`enumerate_source_excluded` (survivors vs the exclusion-filtered zip), which also
+supplies the exclusion block the resumed run records (so reclaim stays correct). A
+changed/added/removed `exclude_spec` shifts the survivor count → `[STALE]` abort →
+re-zip with `--fresh`.
+
+**`verify_exclusions.py`** is the standalone post-hoc re-verifier (the exclusion
+analogue of reclaim's double-check): given run-log JSON(s), it re-confirms every
+recorded exclusion is still git-`COVERED` (`OK`), notes ones whose committed content
+changed since (`CHANGED`), and fails on any that are no longer covered (`LOST`) —
+so "prove nothing real was dropped" is answerable months later. Exits non-zero on
+any `LOST`.
 
 ---
 

archive.py

@@ -360,6 +360,72 @@ def enumerate_source(source_folder, file_pattern):
     return count, newest_mtime
 
 
+def enumerate_source_excluded(source_folder, file_pattern, exclude_spec=None,
+                              exclude_optional=()):
+    """
+    Exclusion-aware sibling of enumerate_source, for the resume staleness guard.
+    Walks the live source, applies the SAME exclusion filter make_zip_files would
+    (reads the spec here, so it works under --globus), and returns:
+
+        (survivor_count, newest_survivor_mtime, exclusion_report)
+
+    counting/timing only the SURVIVORS (matched − excluded). This lets the resume
+    guard compare an exclusion-filtered staged zip against the exclusion-filtered
+    live set (else it would spuriously [STALE]-abort every resume — INV-G). The
+    returned exclusion_report is also what the RESUMED run records in its log, so
+    a resumed exclusion archive keeps its exclusion block (and reclaim's INV-E
+    stays correct). If the spec changed since the zip was built, the survivor set
+    won't match the zip's members → the guard [STALE]-aborts → re-zip with
+    --fresh. With no exclude_spec this is just enumerate_source + an OFF report.
+
+    evaluate_exclusions never raises; imports stay inside the body (Globus
+    Compute / dill, invariant #1).
+    """
+    import os
+    import re
+    import json
+
+    pattern = re.compile(file_pattern)
+    source_folder = os.path.normpath(source_folder)
+    matched = []
+    for dirpath, dirnames, filenames in os.walk(source_folder):
+        for fname in filenames:
+            full_path = os.path.join(dirpath, fname)
+            arcname = os.path.relpath(full_path, source_folder)
+            if pattern.search(arcname):
+                matched.append((full_path, arcname))
+
+    report = {"status": "OFF", "spec_path": exclude_spec}
+    if exclude_spec:
+        try:
+            with open(exclude_spec) as fh:
+                spec = json.load(fh)
+        except Exception as e:
+            report = {"status": "DISABLED", "spec_path": exclude_spec,
+                      "disabled_reason": f"could not read/parse exclude_spec "
+                                         f"{exclude_spec!r}: {e}",
+                      "excluded": [], "excluded_arcnames": [], "demoted": [],
+                      "contested_data": [], "companion_violations": [],
+                      "counts": {}, "excluded_bytes": 0, "error": str(e)}
+            spec = None
+        if spec is not None:
+            report = evaluate_exclusions(matched, source_folder, spec,
+                                         exclude_optional)
+            report["spec_path"] = exclude_spec
+
+    drop = set(report.get("excluded_arcnames") or [])
+    survivors = [(fp, arc) for (fp, arc) in matched if arc not in drop]
+    newest = 0.0
+    for full_path, _arc in survivors:
+        try:
+            m = os.path.getmtime(full_path)
+        except OSError:
+            continue
+        if m > newest:
+            newest = m
+    return len(survivors), newest, report
+
+
 # ---------------------------------------------------------------------------
 # Leveled-incremental primitives (see docs/RFC_incremental_v2.md, PR 1).
 #
@@ -1919,16 +1985,11 @@ def run(fn, *fn_args):
     # make_zip_files entirely and recompute checksums on a compute node. Lets us
     # recover from any failure in later steps (upload, verify) without redoing the zip.
     #
-    # INTERIM (Phase 1b-1): resume-with-exclusion is a follow-up. When an
-    # exclude_spec is configured we always build a fresh zip, so the staleness
-    # guard never compares an exclusion-filtered zip's member count against a
-    # non-exclusion source walk (which would spuriously [STALE]-abort). Proper
-    # resume-under-exclusion (exclusion-aware enumerate_source + a spec-sha guard)
-    # lands next.
-    if exclude_spec and resume_enabled:
-        print("  [EXCLUSION] exclude_spec set — building a fresh zip this run "
-              "(resume-with-exclusion is a follow-up); ignoring any existing zip.")
-        resume_enabled = False
+    # Resume-under-exclusion: the staleness guard below re-evaluates exclusion on
+    # the LIVE source (enumerate_source_excluded) so it compares survivors against
+    # the exclusion-filtered zip, and so this resumed run records the correct
+    # exclusion block for reclaim (INV-E). A changed/added/removed exclude_spec
+    # shifts the survivor set → the guard [STALE]-aborts → re-zip with --fresh.
     existing_zip = find_existing_zip(tmp_dir, project_name) if resume_enabled else None
     resume_result = None
     if existing_zip:
@@ -1942,6 +2003,32 @@ def run(fn, *fn_args):
     if resume_result is not None:
         zip_path, zip_checksum, file_checksums, members, source_bytes, exclusion = resume_result
 
+        # Re-evaluate the live source for the staleness guard. Under an
+        # exclude_spec we use the exclusion-aware walk: it counts SURVIVORS (so it
+        # lines up with the exclusion-filtered zip) AND returns the exclusion
+        # report this resumed run must record for reclaim (INV-E). Computed even
+        # under --force-resume so the log is correct; the staleness comparison
+        # itself is what --force-resume skips.
+        src_count = src_newest_mtime = None
+        if exclude_spec or not force_resume:
+            try:
+                if exclude_spec:
+                    src_count, src_newest_mtime, exclusion = run(
+                        enumerate_source_excluded, source_folder, file_pattern,
+                        exclude_spec, exclude_optional)
+                    _es = exclusion.get("status", "OFF")
+                    print(f"  [RESUME] Exclusion re-evaluated on live source: {_es}"
+                          + (f" ({len(exclusion.get('excluded', []))} excluded)"
+                             if _es == "ON" else ""))
+                else:
+                    src_count, src_newest_mtime = run(
+                        enumerate_source, source_folder, file_pattern)
+            except Exception as e:
+                print(f"\n  ERROR: could not enumerate source to validate resume: {e}")
+                send_alert_email(emails, f"[FAILED] Fortress archive resume check: {project_name}",
+                                 f"source enumeration failed for {project_name}:\n{e}")
+                sys.exit(1)
+
         # Staleness guard: an existing zip is only safe to reuse if it still reflects
         # the live source. Reusing a zip from an earlier run after the source has
         # grown/shrunk/changed silently archives an incomplete snapshot (this is what
@@ -1957,25 +2044,20 @@ def run(fn, *fn_args):
                                                    "%Y%m%d%H%M%S").timestamp()
                 except ValueError:
                     zip_ts = None
-            try:
-                src_count, src_newest_mtime = run(enumerate_source, source_folder, file_pattern)
-            except Exception as e:
-                print(f"\n  ERROR: could not enumerate source to validate resume: {e}")
-                send_alert_email(emails, f"[FAILED] Fortress archive resume check: {project_name}",
-                                 f"enumerate_source failed for {project_name}:\n{e}")
-                sys.exit(1)
             problems = []
             if src_count != len(members):
-                problems.append(f"file count differs: source has {src_count}, "
-                                f"zip has {len(members)}")
-            if zip_ts is not None and src_newest_mtime > zip_ts:
+                problems.append(f"file count differs: source has {src_count} "
+                                f"(after exclusion), zip has {len(members)}")
+            if zip_ts is not None and src_newest_mtime is not None and src_newest_mtime > zip_ts:
                 problems.append(f"source modified after the zip was made "
                                 f"(newest source mtime is later than the zip timestamp)")
             if problems:
                 detail = "\n".join(f"  - {p}" for p in problems)
                 msg = (
                     f"Existing zip is STALE — it no longer matches the live source, so\n"
-                    f"resuming from it would archive an incomplete/outdated snapshot.\n\n"
+                    f"resuming from it would archive an incomplete/outdated snapshot.\n"
+                    f"(A changed/added/removed exclude_spec also shifts the file count\n"
+                    f"and shows up here — re-zip with --fresh in that case.)\n\n"
                     f"Project:       {project_name}\n"
                     f"Source folder: {source_folder}\n"
                     f"Existing zip:  {existing_zip}\n\n"

tests/test_exclusion_polish.py

@@ -0,0 +1,193 @@
+"""
+Tests for the exclusion polish (finishing Phase 1):
+  - archive.enumerate_source_excluded — the exclusion-aware resume staleness walk
+    (survivor count + newest survivor mtime + the exclusion report the resumed
+    run records for reclaim/INV-E).
+  - verify_exclusions.verify_log — the standalone post-hoc re-verifier that proves
+    every recorded exclusion is still git-COVERED.
+
+Real throwaway git repo (data + metadata committed + pushed to a local bare
+remote); a spec + a constructed archive log point at it. All local; no network.
+
+Run: python3 -m unittest discover -s tests
+"""
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import archive
+import verify_exclusions
+
+
+def _git(cwd, *args):
+    r = subprocess.run(["git", "-C", cwd, *args], capture_output=True, text=True)
+    if r.returncode != 0:
+        raise RuntimeError(f"git {' '.join(args)} failed: {r.stderr}")
+    return r
+
+
+def _have_git():
+    try:
+        subprocess.run(["git", "--version"], capture_output=True)
+        return True
+    except OSError:
+        return False
+
+
+@unittest.skipUnless(_have_git(), "git not available")
+class _RepoBase(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.remote = os.path.join(self.tmp, "remote.git")
+        self.work = os.path.join(self.tmp, "work")
+        self.asset = os.path.join(self.work, "asset")
+        subprocess.run(["git", "init", "--bare", "-b", "main", self.remote],
+                       capture_output=True, check=True)
+        subprocess.run(["git", "init", "-b", "main", self.work],
+                       capture_output=True, check=True)
+        for k, v in [("user.email", "t@t"), ("user.name", "t"),
+                     ("core.autocrlf", "false"), ("commit.gpgsign", "false")]:
+            _git(self.work, "config", k, v)
+        _git(self.work, "remote", "add", "origin", self.remote)
+        self._write("asset/scan.tiff", b"\x89imgdata")     # DATA
+        self._write("asset/values.csv", b"a,b\n1,2\n")      # DATA
+        self._write("asset/X_status.json", b'{"s":1}')      # EXCLUDE
+        self._write("asset/catalog.json", b'{"c":1}')       # EXCLUDE
+        _git(self.work, "add", "-A")
+        _git(self.work, "commit", "-m", "c")
+        _git(self.work, "push", "-u", "origin", "main")
+
+    def tearDown(self):
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _write(self, rel, content):
+        p = os.path.join(self.work, rel)
+        os.makedirs(os.path.dirname(p), exist_ok=True)
+        with open(p, "wb") as fh:
+            fh.write(content)
+        return p
+
+    def _blob(self, rel):
+        return _git(self.work, "rev-parse", f"HEAD:asset/{rel}").stdout.strip()
+
+    def _spec_file(self):
+        spec = {
+            "version": 1,
+            "coverage": {"oracle": "git", "repo_root": self.work,
+                         "require_pushed": True, "remote_durable_attestation": True,
+                         "forbid_remote_under": ["/depot/"], "max_fetch_age": "24h"},
+            "data_identity": [
+                {"id": "tabular", "kind": "regex", "pattern": r"\.(csv|tsv|xlsx)$"},
+                {"id": "imagery", "kind": "regex", "pattern": r"\.(png|tif|tiff)$"}],
+            "exclude_identity": [
+                {"id": "status", "kind": "regex", "pattern": r"(^|/)[^/]+_status\.json$"},
+                {"id": "catalog", "kind": "regex", "pattern": r"(^|/)(catalog|manifest)\.json$"}],
+            "optional_identity": [],
+        }
+        p = os.path.join(self.tmp, "exclude_spec.json")
+        with open(p, "w") as fh:
+            json.dump(spec, fh)
+        return p
+
+
+class TestEnumerateSourceExcluded(_RepoBase):
+    def test_no_spec_counts_everything(self):
+        count, newest, report = archive.enumerate_source_excluded(self.asset, ".*")
+        self.assertEqual(count, 4)
+        self.assertEqual(report["status"], "OFF")
+
+    def test_spec_counts_only_survivors(self):
+        count, newest, report = archive.enumerate_source_excluded(
+            self.asset, ".*", self._spec_file())
+        self.assertEqual(report["status"], "ON", report.get("disabled_reason"))
+        self.assertEqual(count, 2)     # data only; the 2 metadata excluded
+        self.assertEqual(sorted(report["excluded_arcnames"]),
+                         ["X_status.json", "catalog.json"])
+
+    def test_unreadable_spec_counts_everything(self):
+        count, newest, report = archive.enumerate_source_excluded(
+            self.asset, ".*", os.path.join(self.tmp, "nope.json"))
+        self.assertEqual(report["status"], "DISABLED")
+        self.assertEqual(count, 4)     # default-safe: nothing subtracted
+
+
+class TestVerifyExclusions(_RepoBase):
+    def _log(self, excluded_entries):
+        return {
+            "project": "T", "run_timestamp": "T_20260101_000000",
+            "source_folder": self.asset, "fortress_tar": "/f/T.tar",
+            "source_files": {"scan.tiff": "m", "values.csv": "m"},
+            "status": "success",
+            "exclusion": {
+                "status": "ON", "spec_sha256": "abc",
+                "excluded": excluded_entries,
+                "excluded_arcnames": [e["arcname"] for e in excluded_entries],
+                "coverage": {"repo_root": self.work, "require_pushed": True,
+                             "remote_durable_attestation": True,
+                             "forbid_remote_under": ["/depot/"],
+                             "max_fetch_age": "24h"},
+            },
+        }
+
+    def _write_log(self, log):
+        p = os.path.join(self.tmp, "run.json")
+        with open(p, "w") as fh:
+            json.dump(log, fh)
+        return p
+
+    def test_ok_when_unchanged(self):
+        log = self._log([
+            {"arcname": "X_status.json", "blob_oid": self._blob("X_status.json")},
+            {"arcname": "catalog.json", "blob_oid": self._blob("catalog.json")}])
+        status, results = verify_exclusions.verify_log(self._write_log(log))
+        self.assertEqual(status, "OK")
+        self.assertEqual(len(results["ok"]), 2)
+        self.assertEqual(results["lost"], [])
+
+    def test_changed_when_regenerated_and_recommitted(self):
+        # Record the OLD blob, then regenerate + re-commit + push -> still covered
+        # but a different blob -> CHANGED (informational, not a failure).
+        old = self._blob("X_status.json")
+        log = self._log([{"arcname": "X_status.json", "blob_oid": old}])
+        self._write("asset/X_status.json", b'{"s":2,"new":true}')
+        _git(self.work, "add", "-A")
+        _git(self.work, "commit", "-m", "regen")
+        _git(self.work, "push", "origin", "main")
+        status, results = verify_exclusions.verify_log(self._write_log(log))
+        self.assertEqual(status, "OK")      # CHANGED is not a failure
+        self.assertEqual(len(results["changed"]), 1)
+        self.assertEqual(results["lost"], [])
+
+    def test_lost_when_edited_not_pushed(self):
+        log = self._log([
+            {"arcname": "X_status.json", "blob_oid": self._blob("X_status.json")}])
+        self._write("asset/X_status.json", b'{"s":99,"edited":true}')
+        _git(self.work, "add", "-A")
+        _git(self.work, "commit", "-m", "edit, no push")  # blob not on remote
+        status, results = verify_exclusions.verify_log(self._write_log(log))
+        self.assertEqual(status, "LOST")
+        self.assertEqual(len(results["lost"]), 1)
+        self.assertEqual(results["lost"][0][0], "X_status.json")
+
+    def test_skip_when_no_exclusion(self):
+        log = self._log([])
+        log["exclusion"] = {"status": "OFF"}
+        status, results = verify_exclusions.verify_log(self._write_log(log))
+        self.assertEqual(status, "SKIP")
+
+    def test_gate_off_is_lost(self):
+        log = self._log([
+            {"arcname": "X_status.json", "blob_oid": self._blob("X_status.json")}])
+        self._write("asset/loose.txt", b"uncommitted")   # dirties subtree -> gate off
+        status, results = verify_exclusions.verify_log(self._write_log(log))
+        self.assertEqual(status, "LOST")
+
+
+if __name__ == "__main__":
+    unittest.main()