Skip to content

feat(exclusion): wire the second filter into the archive path (Phase 1b-1) #15

Merged
Jarrod S Doucette (jdoucett) merged 1 commit into from
Jul 1, 2026
+410 −16
Merged

feat(exclusion): wire the second filter into the archive path (Phase 1b-1) #15

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions CLAUDE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,9 +269,27 @@ Personal auth (endpoint owner) never needs `allowed_principals`. The local
| `log_dir` | No | `tmp_dir` | Where log pairs are written |
| `compression` | No | `deflate` | `deflate` (zip DEFLATE) or `store` (no compression — faster for already-compressed media) |
| `cleanup_zip_on_success` | No | `false` | Remove the staging zip after the round-trip verification passes (recurring/automated archives; see invariant #5) |
| `exclude_spec` | No | `null` | PATH to a shared exclusion spec JSON (see `docs/EXCLUSION_SPEC.md`) that leaves git-versioned regenerable metadata off the backup. Classified by role-bearing path (never extension); dropped only on a positive git `COVERED` proof — any uncertainty backs the file up. Default-safe; read where the zip is built so it works under `--globus`. Omit to disable. |
| `exclude_optional` | No | `[]` | Per-asset opt-out list of Tier-2 (OPTIONAL) rule ids from the spec (regenerable provenance/QC). Honored only if the asset also ships the raw inputs. Omit to back up all OPTIONAL artifacts. |

See `config.example.json` for an annotated template.

## Exclusion (leaving git-versioned metadata off the backup)

`docs/EXCLUSION_SPEC.md` is the authoritative design. In brief: after `file_pattern`,
a second **subtractive** filter drops files that are (a) classified EXCLUDE (or
OPTIONAL opted-out) by a **role-bearing relpath regex** — never by extension, a bare
`*.json` rule is rejected at compile time — **and** (b) proven git-`COVERED`
(tracked + committed-at-HEAD + clean + blob on a fresh-fetched remote off `/depot` +
real bytes, not an LFS pointer + no EOL filter). Every other verdict → **back up**.
`evaluate_exclusions()` **never raises**: a bad spec, an unreachable git remote, or a
missing `git` on a compute node all degrade to "back up everything" with a recorded
reason (banner + run-log `exclusion` block + completion email). The **companion
guard** (INV-C) suppresses exclusion run-wide if a directory would ship zero members,
and the run continues backing up everything with a loud `[EXCLUSION-ABORT]` alert.
Runs under `--globus` too (the spec PATH is read on the compute node). Reclaim must
re-apply the recorded excluded set + re-run the gate (INV-E) — pending.

---

## Known issues / history
Expand Down
202 changes: 190 additions & 12 deletions archive.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,26 @@ def load_config(config_path):
f"got: {config['cleanup_zip_on_success']!r}"
)

# exclude_spec: optional PATH to an exclusion spec JSON (docs/EXCLUSION_SPEC.md).
# Only the TYPE is validated here — the file is READ at run time, on the
# compute node that has /depot, so exclusion works under --globus (the
# launching host may have no filesystem access). All spec validation + the git
# coverage gate run in evaluate_exclusions, which is default-safe: any problem
# yields "back up everything" with a recorded reason (never a hard failure).
config.setdefault("exclude_spec", None)
if config["exclude_spec"] is not None and not isinstance(config["exclude_spec"], str):
raise ValueError(
f"config 'exclude_spec' must be a path string or null, "
f"got: {config['exclude_spec']!r}"
)
# exclude_optional: per-asset Tier-2 opt-out list (OPTIONAL rule ids to drop).
config.setdefault("exclude_optional", [])
if not isinstance(config["exclude_optional"], list):
raise ValueError(
f"config 'exclude_optional' must be a list of rule ids, "
f"got: {config['exclude_optional']!r}"
)

return config


Expand Down Expand Up @@ -205,9 +225,14 @@ def find_existing_zip(tmp_dir, project_name):

def reconstruct_from_zip(zip_path):
"""
Rebuild (zip_path, zip_checksum, file_checksums, members, source_bytes)
from an existing zip on disk — the same tuple make_zip_files would have
returned.
Rebuild (zip_path, zip_checksum, file_checksums, members, source_bytes,
exclusion) from an existing zip on disk — the same 6-tuple make_zip_files
returns. The zip's members are whatever was archived (already
exclusion-filtered if it was built under an exclude_spec), so the exclusion
element is a resume marker only: {"status": "OFF", "note": "resumed"}. Proper
resume-under-exclusion (re-verifying the staged zip against the current spec)
is handled in the caller; today the wiring forces a fresh build whenever an
exclude_spec is configured, so this path never runs under exclusion.
Validates zip integrity via zipfile.testzip() first. Per-file checksums are
computed by reading each member's bytes back through MD5; the result is
Expand Down Expand Up @@ -260,7 +285,8 @@ def reconstruct_from_zip(zip_path):
fh.close()
zip_checksum = md5.hexdigest()

return zip_path, zip_checksum, file_checksums, members, source_bytes
exclusion = {"status": "OFF", "note": "resumed from existing zip"}
return zip_path, zip_checksum, file_checksums, members, source_bytes, exclusion


def try_reconstruct(existing_zip, runner):
Expand Down Expand Up @@ -1084,7 +1110,8 @@ def evaluate_exclusions(matched_members, source_folder, spec,


def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
compression="deflate", allow_empty_files=False):
compression="deflate", allow_empty_files=False,
exclude_spec=None, exclude_optional=()):
"""
On the RCAC compute system: find files in source_folder matching
file_pattern (regex), zip them preserving relative paths, verify
Expand All @@ -1103,9 +1130,21 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
0-byte member. Set True to archive legitimately-empty files as-is (e.g.
.gitkeep placeholders, empty __init__.py markers); they still get a valid
(empty-string) MD5 and are counted, so reclaim's drift check stays exact.
exclude_spec: optional PATH to an exclusion spec JSON (docs/EXCLUSION_SPEC.md).
The path is READ HERE — on the compute system, which has /depot — so exclusion
works under --globus too (the launching host may have no /depot). Files that
are git-COVERED regenerable metadata are subtracted from the matched set;
exclude_optional is the per-asset Tier-2 opt-out list. evaluate_exclusions
NEVER raises: any problem yields "back up everything" with a recorded reason.
Returns (zip_path, zip_checksum, file_checksums, members, source_bytes,
exclusion) — the 6th element is the exclusion report (status OFF/DISABLED/ON
+ excluded/demoted/counts) for the run log, banner, and completion email.
"""
import os
import re
import json
import zipfile
import hashlib
import datetime
Expand All @@ -1129,6 +1168,40 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
f"such as 'raw/IMG_001.raw')"
)

# Second, subtractive filter (docs/EXCLUSION_SPEC.md): drop git-COVERED
# regenerable metadata. Read the spec here (compute node has /depot) so it
# works under --globus. evaluate_exclusions NEVER raises; on any problem it
# returns status DISABLED + an empty excluded set (back up everything).
exclusion = {"status": "OFF", "spec_path": exclude_spec}
if exclude_spec:
try:
with open(exclude_spec) as _fh:
_spec = json.load(_fh)
except Exception as e:
exclusion = {"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:
exclusion = evaluate_exclusions(matched_files, source_folder,
_spec, exclude_optional)
exclusion["spec_path"] = exclude_spec
drop = set(exclusion.get("excluded_arcnames") or [])
if drop:
matched_files = [(fp, arc) for (fp, arc) in matched_files
if arc not in drop]
if not matched_files:
# Should be unreachable: the companion guard suppresses exclusion
# rather than let a directory ship zero members. Guard anyway — never
# produce an empty archive silently.
raise RuntimeError(
f"Exclusion removed every matched file for '{project_name}' — "
f"refusing to build an empty archive (this indicates a spec bug; "
f"the companion guard should have prevented it)")

os.makedirs(tmp_dir, exist_ok=True)

timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
Expand Down Expand Up @@ -1186,13 +1259,13 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
# preserves mtime) — the one drift case count+mtime alone cannot see.
source_bytes = sum(os.path.getsize(fp) for fp, _ in matched_files)

return zip_path, zip_checksum, file_checksums, members, source_bytes
return zip_path, zip_checksum, file_checksums, members, source_bytes, exclusion


def send_to_fortress(zip_path, zip_checksum, file_checksums, members,
source_bytes, source_folder, file_pattern,
fortress_base_dir, emails, project_name, log_dir,
cleanup_zip_on_success=False):
cleanup_zip_on_success=False, exclusion=None):
"""
On the RCAC compute system: transfer the zip to Fortress via htar_large,
then performs full round-trip retrieval and MD5 verification.
Expand Down Expand Up @@ -1261,6 +1334,11 @@ def send_to_fortress(zip_path, zip_checksum, file_checksums, members,
"source_bytes": source_bytes,
"file_pattern": file_pattern,
"recipients": emails,
# Full exclusion report (docs/EXCLUSION_SPEC.md): status, excluded set,
# demoted, coverage evidence, counts. reclaim reads exclusion.excluded /
# excluded_arcnames + re-runs the gate (INV-E). Absent/OFF for archives
# made without an exclude_spec — fully back-compatible with old logs.
"exclusion": (exclusion or {"status": "OFF"}),
"events": [],
"status": "running",
"error": None,
Expand Down Expand Up @@ -1336,6 +1414,56 @@ def send_email(subject, body):
for fname, cksum in file_checksums.items():
write_log(f"Source file MD5: {fname} = {cksum}")

# ── Exclusion audit (docs/EXCLUSION_SPEC.md §6) ────────────────────────────
# An operator must be able to PROVE nothing real was dropped. The .txt gets a
# summary + sampled events; the full excluded/demoted lists live in the JSON
# run_record. On a companion-guard trip we CONTINUE (exclusion already
# suppressed run-wide → everything backed up) and send a loud alert.
excl = exclusion or {"status": "OFF"}
estatus = excl.get("status", "OFF")
if estatus == "ON":
write_log(f"Exclusion ON (spec {str(excl.get('spec_sha256') or '')[:12]}): "
f"dropped {len(excl.get('excluded', []))} file(s), "
f"{excl.get('excluded_bytes', 0)} bytes; "
f"demoted {len(excl.get('demoted', []))} to backup")
cov = excl.get("coverage") or {}
if cov.get("repo_head"):
write_log(f"Exclusion coverage: git {str(cov.get('repo_head'))[:12]} "
f"on {cov.get('remote_ref')} "
f"(pushed={cov.get('require_pushed')}, "
f"attested={cov.get('remote_durable_attestation')})")
for e in excl.get("excluded", [])[:100]:
write_log(f"[EXCLUDE] {e.get('arcname')} rule={e.get('rule_id')} "
f"{e.get('verdict')} blob={str(e.get('blob_oid') or '')[:12]} "
f"@ {e.get('remote_ref')}")
for d in excl.get("demoted", [])[:100]:
write_log(f"[EXCLUDE-DEMOTE] {d.get('arcname')} "
f"reason={d.get('verdict')} → backed up")
for c in excl.get("contested_data", []):
write_log(f"[EXCLUDE-CONTEST] {c.get('arcname')} kept as DATA "
f"(also matched rule {c.get('also_matched')})")
elif estatus == "DISABLED":
write_log(f"Exclusion DISABLED this run — {excl.get('disabled_reason')}; "
f"everything backed up", level="WARNING")
# estatus OFF: no exclude_spec configured — nothing to log.

if excl.get("companion_violations"):
detail = "\n".join(f" - {v.get('directory') or '(root)'}: {v.get('reason')}"
for v in excl["companion_violations"])
write_log("[EXCLUSION-ABORT] companion guard tripped — exclusion suppressed, "
"backing up EVERYTHING; investigate classification", level="ERROR")
send_email(
f"[EXCLUSION-ABORT] Fortress archive: {project_name}",
f"Project: {project_name}\n\n"
f"The exclusion companion guard tripped: a directory that had matched "
f"files would ship zero members (or drop a DATA member). Exclusion was "
f"SUPPRESSED for this run — the archive proceeds and backs up "
f"EVERYTHING, so no data is lost — but the exclude spec / classification "
f"needs review.\n\n{detail}\n\n"
f"Log (TXT): {txt_path}\n"
f"Log (JSON): {json_path}"
)

subprocess.run(f'hsi "mkdir -p {fortress_base_dir}"',
shell=True, capture_output=True)

Expand Down Expand Up @@ -1570,14 +1698,36 @@ def send_email(subject, body):

finalize_log("success")

# Exclusion summary for the completion email — self-explains the run so an
# operator can see at a glance whether exclusion fired and why (§6).
if estatus == "ON":
c = excl.get("counts") or {}
excl_email = (
f"\nExclusion: ON (spec {str(excl.get('spec_sha256') or '')[:12]})\n"
f" excluded: {len(excl.get('excluded', []))} file(s), "
f"{excl.get('excluded_bytes', 0)} bytes (all git-COVERED)\n"
f" demoted to backup: {len(excl.get('demoted', []))} "
f"(not COVERED — see log)\n"
f" tiers: DATA-allowlist={c.get('data_allowlist', 0)} "
f"DATA-default={c.get('data_default', 0)} "
f"OPTIONAL-shipped={c.get('optional_shipped', 0)}\n"
)
elif estatus == "DISABLED":
excl_email = (f"\nExclusion: DISABLED — "
f"{excl.get('disabled_reason')}\n"
f" (everything was backed up)\n")
else:
excl_email = ""

send_email(
f"[SUCCESS] Fortress archive complete: {filename}",
f"Project: {project_name}\n\n"
f"Archive workflow completed successfully.\n\n"
f"Source (Depot): {source_folder}\n"
f"Destination (Fortress): {fortress_tar}\n"
f"Local zip: {zip_note}\n"
f"Zip MD5: {zip_checksum}\n\n"
f"Zip MD5: {zip_checksum}\n"
f"{excl_email}\n"
f"Log (TXT): {txt_path}\n"
f"Log (JSON): {json_path}\n\n"
f"Full log is available at the path above."
Expand Down Expand Up @@ -1643,6 +1793,8 @@ def send_email(subject, body):
compression = config["compression"]
allow_empty_files = config["allow_empty_files"]
cleanup_zip_on_success = config["cleanup_zip_on_success"]
exclude_spec = config["exclude_spec"]
exclude_optional = config["exclude_optional"]

# `run` abstracts how a worker function is executed: in-process by default, or
# submitted to the endpoint and waited on with --globus. Step 1/Step 2 call run()
Expand Down Expand Up @@ -1749,11 +1901,25 @@ def run(fn, *fn_args):
print(f" Zip location: {tmp_dir}/{project_name}_YYYYMMDD_HHMMSS.zip")
print(f" Compression: {compression}")
print(f" Allow empty: {allow_empty_files}")
if exclude_spec:
print(f" Exclude spec: {exclude_spec}")
print(f" Opt-outs: {exclude_optional or 'none'}")
print("=" * 60)

# Resume short-circuit: if a previous run left a valid zip in tmp_dir, skip
# 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
existing_zip = find_existing_zip(tmp_dir, project_name) if resume_enabled else None
resume_result = None
if existing_zip:
Expand All @@ -1765,7 +1931,7 @@ def run(fn, *fn_args):
if resume_result is None:
existing_zip = None
if resume_result is not None:
zip_path, zip_checksum, file_checksums, members, source_bytes = resume_result
zip_path, zip_checksum, file_checksums, members, source_bytes, exclusion = resume_result

# 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
Expand Down Expand Up @@ -1819,9 +1985,9 @@ def run(fn, *fn_args):
print(f" [OK] {len(members)} file(s) verified from existing zip")
else:
try:
zip_path, zip_checksum, file_checksums, members, source_bytes = run(
zip_path, zip_checksum, file_checksums, members, source_bytes, exclusion = run(
make_zip_files, source_folder, file_pattern, tmp_dir, project_name,
compression, allow_empty_files)
compression, allow_empty_files, exclude_spec, exclude_optional)
except Exception as e:
msg = (
f"Step 1 (make_zip_files) failed.\n\n"
Expand All @@ -1837,6 +2003,18 @@ def run(fn, *fn_args):
print(f" [OK] Zip: {zip_path}")
print(f" [OK] Zip MD5: {zip_checksum}")

# Exclusion status banner (known only after Step 1 — the gate runs where the
# zip is built, i.e. remotely under --globus). The full detail is in the run
# log + completion email; this is the at-a-glance line.
_estatus = (exclusion or {}).get("status", "OFF")
if _estatus == "ON":
print(f" [OK] Exclusion: ON — dropped {len(exclusion.get('excluded', []))} "
f"file(s) ({exclusion.get('excluded_bytes', 0)} bytes), "
f"demoted {len(exclusion.get('demoted', []))} to backup")
elif _estatus == "DISABLED":
print(f" WARNING: Exclusion DISABLED — {exclusion.get('disabled_reason')}")
print(f" (everything is being backed up — safe, but exclusion did not fire)")

# Step 2: Transfer to Fortress and verify
print()
print("=" * 60)
Expand All @@ -1850,7 +2028,7 @@ def run(fn, *fn_args):
send_to_fortress,
zip_path, zip_checksum, file_checksums, members,
source_bytes, source_folder, file_pattern, fortress_base,
emails, project_name, log_dir, cleanup_zip_on_success
emails, project_name, log_dir, cleanup_zip_on_success, exclusion
)
except Exception as e:
msg = (
Expand Down
Loading