feat(exclusion): re-land --globus robustness layer dropped from #13 merge

archive.py

@@ -691,8 +691,14 @@ def git_coverage_preconditions(coverage, repo_root, asset_subtree, now=None):
     ctx = {}
     now = time.time() if now is None else now
 
-    # 1a. Repo present.
+    # 1a. git present + repo present. rc 127 from _git means the git binary is
+    # not on PATH — a common --globus compute-node gap (like hsi), reported
+    # distinctly so a disabled-exclusion run is diagnosable rather than being
+    # mistaken for a missing repo.
     rc, out, _ = _git(repo_root, "rev-parse", "--show-toplevel")
+    if rc == 127:
+        return (False, "git not available on PATH (compute node may need a "
+                       "module load / PATH fix)", ctx)
     if rc != 0 or not out.strip():
         return (False, f"git repo not found at {repo_root}", ctx)
     toplevel = os.path.normpath(out.strip())
@@ -805,8 +811,13 @@ def git_file_verdict(repo_root, full_path, ctx):
     require_pushed = ctx.get("require_pushed", True)
     remote_ref = ctx.get("remote_ref")
 
-    # L — tracked in the index.
+    # L — tracked in the index. git-missing (rc 127) -> ERROR, never a false
+    # UNTRACKED (which would look like "back up" for the right reason but the
+    # wrong cause — the operator must see that git itself was unavailable).
     rc, _, _ = _git(repo_root, "ls-files", "--error-unmatch", "--", rel)
+    if rc == 127:
+        ev["reason"] = "git unavailable"
+        return (COV_ERROR, ev)
     if rc != 0:
         return (COV_UNTRACKED, ev)
 
@@ -874,6 +885,204 @@ def git_file_verdict(repo_root, full_path, ctx):
     return (COV_COVERED, ev)
 
 
+def companion_guard(member_records):
+    """
+    INV-C smoke alarm (EXCLUSION_SPEC.md §4). `member_records` is a list of
+    {arcname, tier, excluded(bool)}. Returns a list of violation dicts. A
+    directory that had matched files must still ship at least one member AND must
+    never have a DATA-classified member excluded. DATA is never a candidate so a
+    dropped-DATA violation signals a logic bug; the zero-survivor case catches a
+    directory whose entire matched set was excluded (a misclassification hint).
+    """
+    import os
+    by_dir = {}
+    for r in member_records:
+        by_dir.setdefault(os.path.dirname(r["arcname"]), []).append(r)
+    violations = []
+    for d, recs in sorted(by_dir.items()):
+        survivors = [r for r in recs if not r["excluded"]]
+        dropped_data = [r["arcname"] for r in recs
+                        if r["excluded"] and r["tier"] == EXCL_DATA]
+        if dropped_data:
+            violations.append({"directory": d, "reason": "would drop DATA member(s)",
+                               "members": dropped_data})
+        elif not survivors:
+            violations.append({"directory": d,
+                               "reason": "directory would ship zero members",
+                               "members": [r["arcname"] for r in recs]})
+    return violations
+
+
+def evaluate_exclusions(matched_members, source_folder, spec,
+                        exclude_optional=(), now=None):
+    """
+    Decide which already-matched members to EXCLUDE, with exhaustive error
+    handling and a fully serializable diagnostic report. **NEVER raises** — any
+    failure (malformed spec, git unavailable, fetch failure, a per-file gate
+    error, or any unexpected exception) degrades to "exclude nothing, back up
+    everything" with a recorded reason. So exclusion can never fail an archive,
+    and — crucially for --globus, where this runs on a compute node — the report
+    is plain dicts/lists/strings that travel back to the launcher intact for the
+    startup banner + completion email.
+
+    matched_members: iterable of (full_path, arcname) already matched by
+                     file_pattern (the first filter). `spec` is the RAW dict
+                     (recompiled here, so it is safe to ship across the Compute
+                     boundary). Returns a report dict:
+
+      status: "OFF"      — no spec configured (feature not requested)
+              "DISABLED" — spec present but exclusion could not run safely
+                           (disabled_reason set) -> everything backed up
+              "ON"       — exclusion ran (some candidates may still be demoted)
+      disabled_reason, error, spec_sha256, coverage,
+      excluded[], excluded_arcnames[], demoted[], contested_data[],
+      companion_violations[], counts{}, excluded_bytes
+
+    On a companion-guard violation, exclusion is suppressed for the whole run
+    (drop nothing) AND the violation is reported, so the wiring can raise a loud
+    [EXCLUSION-ABORT] while the data is already safe regardless.
+    """
+    import os
+
+    members = list(matched_members)
+    report = {
+        "status": "OFF", "disabled_reason": None, "error": None,
+        "spec_sha256": None, "coverage": {},
+        "excluded": [], "excluded_arcnames": [], "demoted": [],
+        "contested_data": [], "companion_violations": [],
+        "counts": {}, "excluded_bytes": 0,
+    }
+    if not spec:
+        return report
+
+    try:
+        try:
+            compiled = compile_exclude_spec(spec, exclude_optional)
+        except ValueError as e:
+            report["status"] = "DISABLED"
+            report["disabled_reason"] = f"invalid exclude spec: {e}"
+            return report
+
+        report["spec_sha256"] = compiled["spec_sha256"]
+        coverage = compiled["coverage"]
+        repo_root = coverage.get("repo_root")
+        if not repo_root:
+            report["status"] = "DISABLED"
+            report["disabled_reason"] = "exclude spec coverage has no repo_root"
+            return report
+
+        try:
+            ok, reason, ctx = git_coverage_preconditions(
+                coverage, repo_root, source_folder, now=now)
+        except Exception as e:
+            report["status"] = "DISABLED"
+            report["disabled_reason"] = f"coverage preconditions errored: {e}"
+            return report
+        if not ok:
+            report["status"] = "DISABLED"
+            report["disabled_reason"] = reason
+            return report
+
+        report["status"] = "ON"
+        report["coverage"] = {
+            "repo_head": ctx.get("repo_head"),
+            "remote_ref": ctx.get("remote_ref"),
+            "require_pushed": ctx.get("require_pushed"),
+            "remote_durable_attestation": ctx.get("remote_durable_attestation"),
+            "fetch_ts": ctx.get("fetch_ts"),
+        }
+
+        counts = {k: 0 for k in (
+            "data_allowlist", "data_default", "exclude_excluded",
+            "exclude_demoted", "optional_shipped", "optional_dropped_excluded",
+            "optional_dropped_demoted")}
+        # Sub-tier rules, for flagging a DATA file that ALSO matched one (proof
+        # that precedence — not luck — kept it; EXCLUSION_SPEC.md §6).
+        sub_rules = list(compiled["exclude_identity"]) + \
+            [(rid, rx, note) for (rid, rx, _d, note) in compiled["optional_identity"]]
+
+        member_records = []
+        for full_path, arcname in members:
+            cls = classify_tier(arcname, compiled)
+            rec = {"arcname": arcname, "tier": cls["tier"], "excluded": False}
+            member_records.append(rec)
+
+            if cls["tier"] == EXCL_DATA:
+                if cls["data_kind"] == "DATA-allowlist":
+                    counts["data_allowlist"] += 1
+                    hit = next((rid for (rid, rx, _n) in sub_rules
+                                if rx.search(arcname)), None)
+                    if hit:
+                        report["contested_data"].append(
+                            {"arcname": arcname, "rule_id": cls["rule_id"],
+                             "also_matched": hit})
+                else:
+                    counts["data_default"] += 1
+                continue
+
+            if not cls["candidate"]:
+                counts["optional_shipped"] += 1   # OPTIONAL, default backup
+                continue
+
+            # Candidate (EXCLUDE, or OPTIONAL opted-out): demand a COVERED proof.
+            try:
+                verdict, ev = git_file_verdict(repo_root, full_path, ctx)
+            except Exception as e:
+                verdict, ev = COV_ERROR, {"reason": f"{type(e).__name__}: {e}"}
+
+            is_excl = cls["tier"] == EXCL_EXCLUDE
+            if verdict == COV_COVERED:
+                rec["excluded"] = True
+                report["excluded"].append(
+                    {"arcname": arcname, "rule_id": cls["rule_id"],
+                     "verdict": verdict, "blob_oid": ev.get("blob_oid"),
+                     "remote_ref": ev.get("remote_ref")})
+                counts["exclude_excluded" if is_excl
+                       else "optional_dropped_excluded"] += 1
+                try:
+                    report["excluded_bytes"] += os.path.getsize(full_path)
+                except OSError:
+                    pass
+            else:
+                report["demoted"].append(
+                    {"arcname": arcname, "rule_id": cls["rule_id"],
+                     "verdict": verdict, "reason": ev.get("reason", "")})
+                counts["exclude_demoted" if is_excl
+                       else "optional_dropped_demoted"] += 1
+
+        report["counts"] = counts
+
+        # INV-C companion guard. On any violation: default-safe — suppress ALL
+        # exclusion this run (drop nothing) and report it loudly so the wiring
+        # can raise [EXCLUSION-ABORT]; the data is already safe either way.
+        violations = companion_guard(member_records)
+        if violations:
+            report["companion_violations"] = violations
+            report["status"] = "DISABLED"
+            report["disabled_reason"] = (
+                "companion guard tripped (a directory would ship zero members or "
+                "drop a DATA member) — exclusion suppressed, everything backed "
+                "up; investigate classification")
+            report["excluded"] = []
+            report["excluded_arcnames"] = []
+            report["excluded_bytes"] = 0
+            return report
+
+        report["excluded_arcnames"] = [e["arcname"] for e in report["excluded"]]
+        return report
+
+    except Exception as e:
+        # Absolute backstop: exclusion can NEVER fail an archive. Anything that
+        # slipped past the targeted handlers above lands here -> back up all.
+        report["status"] = "DISABLED"
+        report["error"] = f"{type(e).__name__}: {e}"
+        report["disabled_reason"] = f"unexpected exclusion error: {e}"
+        report["excluded"] = []
+        report["excluded_arcnames"] = []
+        report["excluded_bytes"] = 0
+        return report
+
+
 def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
                    compression="deflate", allow_empty_files=False):
     """