From 44bcbce7d50d0d466baa3d3f97ff2647a8c3d76f Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Sun, 14 Jun 2026 14:55:57 -0400 Subject: [PATCH] =?UTF-8?q?feat(reclaim):=20canonical-root=20invariant=20?= =?UTF-8?q?=E2=80=94=20paths=20that=20can=20never=20be=20deleted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a capability-level backstop to the advisory holds list. A "canonical root" (declared via --canonical-root, repeatable, or FORTRESS_CANONICAL_ROOTS) is enforced at two independent points with NO flag to override: * verify() returns a new CANONICAL verdict first, so such a source never enters the SAFE/deletable set, whatever the archive/hold/drift state. * safe_to_rm() re-checks at the actual rm site, so even a direct delete_source() with --yes --allow-no-snapshot cannot remove it. is_canonical() is a path-boundary match on the normalized absolute path (not a substring), so a root protects itself and all descendants but never a sibling. The holds list says "shouldn't"; a canonical root says "can't". Tests: helper merge/normalize/dedupe + path-boundary matching; verify() marks CANONICAL over a would-be-SAFE source and over a hold; safe_to_rm refuses; delete_source leaves the data untouched under the strongest delete invocation. All 26 tests pass (7 new, 19 existing). Co-Authored-By: Claude Opus 4.8 (1M context) --- reclaim.py | 88 ++++++++++++++++++++++---- tests/test_canonical_root.py | 116 +++++++++++++++++++++++++++++++++++ 2 files changed, 192 insertions(+), 12 deletions(-) create mode 100644 tests/test_canonical_root.py diff --git a/reclaim.py b/reclaim.py index 3d74326..adbe89d 100644 --- a/reclaim.py +++ b/reclaim.py @@ -85,6 +85,9 @@ RUNNING = "RUNNING" # archive still in progress -> check back later FAILED = "FAILED" # archive did not succeed / tar missing -> keep GONE = "GONE" # source already removed from Depot -> nothing to do +CANONICAL = "CANONICAL" # under a declared canonical root -> NEVER deletable + # (hard invariant: independent of the holds list, with + # no flag to override — see load_canonical_roots) def load_json(path): @@ -107,6 +110,43 @@ def load_holds(holds_path): return holds +def load_canonical_roots(cli_roots=None): + """Absolute, normalized paths that are NEVER deletable, merged from the + --canonical-root CLI flag (repeatable) and the FORTRESS_CANONICAL_ROOTS env + var (os.pathsep-separated). + + Unlike the holds list — advisory config that says a path "shouldn't" be + deleted — a canonical root is a capability-level invariant: a source equal + to or nested under one is protected regardless of archive/hold state and + regardless of --delete/--yes/--allow-no-snapshot. There is deliberately NO + flag to disable it. The florasense deployment passes its curated repository + here so the reclaim engine cannot remove canonical data even if the holds + file is wrong, missing, or bypassed. + """ + raw = list(cli_roots or []) + env = os.environ.get("FORTRESS_CANONICAL_ROOTS", "") + raw += [p for p in env.split(os.pathsep) if p.strip()] + roots = [] + for p in raw: + norm = os.path.abspath(os.path.expanduser(p.strip())).rstrip("/") + if norm and norm not in roots: + roots.append(norm) + return roots + + +def is_canonical(folder, canonical_roots): + """True if `folder` is a declared canonical root or lives under one. + + Path-boundary match on the normalized absolute path (NOT a substring test), + so '/depot/florasense/data/repository' protects that directory and everything + inside it, but never a sibling such as '..._repository-staging'. + """ + if not folder or not canonical_roots: + return False + f = os.path.abspath(folder).rstrip("/") + return any(f == root or f.startswith(root + "/") for root in canonical_roots) + + def run_datetime(run_timestamp): """Parse the trailing _YYYYMMDD_HHMMSS off a run_timestamp/log stem.""" m = TS_RE.search(run_timestamp or "") @@ -260,7 +300,7 @@ def resolve_source_folder(log, config): return log.get("source_folder") or (config or {}).get("source_folder") -def verify(log, source_folder, holds, file_pattern=None): +def verify(log, source_folder, holds, file_pattern=None, canonical_roots=()): """ Run the structural double-check. Returns a verdict dict with a status and human-readable reasons, plus the numbers used. @@ -316,8 +356,14 @@ def verify(log, source_folder, holds, file_pattern=None): if archived_bytes is None: live_bytes = None - # Decide verdict (most-blocking first) - if status == "running": + # Decide verdict (most-blocking first). The canonical-root invariant is + # checked before everything else and cannot be overridden by any flag: a + # source under a canonical root is never deletable, whatever the archive, + # hold, or drift state says. + if is_canonical(source_folder, canonical_roots): + verdict = CANONICAL + reasons.append("under canonical root — never deletable (hard invariant)") + elif status == "running": verdict = RUNNING elif not status_ok or not tar_ok or source_folder is None: verdict = FAILED @@ -351,8 +397,12 @@ def verify(log, source_folder, holds, file_pattern=None): } -def safe_to_rm(folder): - """Refuse to rm anything that isn't a deep, existing Depot data path.""" +def safe_to_rm(folder, canonical_roots=()): + """Refuse to rm anything that isn't a deep, existing Depot data path — or + that lives under a declared canonical root (the hard invariant, re-checked + here at the actual rm site so no code path or flag can bypass it).""" + if is_canonical(folder, canonical_roots): + return (False, "under canonical root — protected (hard invariant)") if not folder or not folder.startswith(DELETE_ROOT_PREFIX): return (False, f"path not under {DELETE_ROOT_PREFIX}") if folder.rstrip("/").count("/") < DELETE_MIN_DEPTH: @@ -398,12 +448,16 @@ def snapshot_present(folder, live_count): f"— files added since last snapshot are not captured yet") -def delete_source(result, assume_yes, allow_no_snapshot=False): - """Guarded delete: re-verify the Fortress tar AND a Depot snapshot, confirm, rm -rf.""" +def delete_source(result, assume_yes, allow_no_snapshot=False, canonical_roots=()): + """Guarded delete: re-verify the Fortress tar AND a Depot snapshot, confirm, rm -rf. + + canonical_roots is re-checked first via safe_to_rm: a canonical path is + never removed, even with --yes and --allow-no-snapshot. + """ folder = result["source_folder"] tar = result["fortress_tar"] - ok, why = safe_to_rm(folder) + ok, why = safe_to_rm(folder, canonical_roots) if not ok: print(f" SKIP {folder}: {why}") return False @@ -439,7 +493,7 @@ def delete_source(result, assume_yes, allow_no_snapshot=False): def render_result(r): """Render one verdict as plain text (renders cleanly in console and Teams).""" - lines = [f"[{r['verdict']:7}] {r['project']}", + lines = [f"[{r['verdict']:9}] {r['project']}", f" source: {r['source_folder']}", f" files: live={r['live']} archived={r['archived']} newer={r['newer']}"] lines += [f" - {reason}" for reason in r["reasons"]] @@ -552,6 +606,11 @@ def main(): ap.add_argument("--match", help="Only include archives whose project, source_folder, or " "fortress_tar contains this substring (e.g. 'landing-zone')") ap.add_argument("--holds", help="Holds file: project names/path substrings to never delete (one per line)") + ap.add_argument("--canonical-root", dest="canonical_roots", action="append", metavar="PATH", + help="A path that is NEVER deletable (repeatable). A source equal to or under it " + "gets verdict CANONICAL, independent of the holds file and with no override — not " + "even --delete --yes --allow-no-snapshot. Also seeded from FORTRESS_CANONICAL_ROOTS " + "(os.pathsep-separated). Use for curated/canonical data the engine must never remove.") ap.add_argument("--delete", action="store_true", help="Delete SAFE sources (guarded). Default: report only") ap.add_argument("--yes", action="store_true", help="Skip the per-folder confirmation prompt") ap.add_argument("--allow-no-snapshot", dest="allow_no_snapshot", action="store_true", @@ -571,6 +630,9 @@ def main(): run_start = datetime.datetime.now() holds = load_holds(args.holds) + canonical_roots = load_canonical_roots(args.canonical_roots) + if canonical_roots: + print(f"Canonical roots (never deletable): {', '.join(canonical_roots)}") # Build the list of (log_dict, config_dict) to verify. items = [] @@ -623,7 +685,8 @@ def matches(log, cfg): # zip_name from the log + tmp_dir from the config. results = [] for log, cfg in items: - r = verify(log, resolve_source_folder(log, cfg), holds) + r = verify(log, resolve_source_folder(log, cfg), holds, + canonical_roots=canonical_roots) r["zip_name"] = log.get("zip_name") r["tmp_dir"] = (cfg or {}).get("tmp_dir") results.append(r) @@ -647,7 +710,7 @@ def matches(log, cfg): if args.email: recipients += [a.strip() for a in args.email.split(",") if a.strip()] - order = [SAFE, DRIFTED, FAILED, RUNNING, HOLD, GONE] + order = [SAFE, DRIFTED, FAILED, RUNNING, HOLD, CANONICAL, GONE] verdict_sub = " ".join(f"{k}={counts[k]}" for k in order if counts.get(k)) # Report-only (e.g. nightly to the Teams channel): email the verdict report, done. @@ -669,7 +732,8 @@ def matches(log, cfg): print(f"\nDeleting {len(safe)} SAFE folder(s) (each re-verified against Fortress " f"and Depot snapshot first):") for r in safe: - if delete_source(r, args.yes, args.allow_no_snapshot): + if delete_source(r, args.yes, args.allow_no_snapshot, + canonical_roots=canonical_roots): deleted.append(r) if args.no_zip_cleanup: print(" cleanup: skipped (--no-zip-cleanup)") diff --git a/tests/test_canonical_root.py b/tests/test_canonical_root.py new file mode 100644 index 0000000..e78f498 --- /dev/null +++ b/tests/test_canonical_root.py @@ -0,0 +1,116 @@ +""" +Tests for the canonical-root invariant — the capability-level backstop to the +advisory holds list. + +A "canonical root" is a path the reclaim engine must NEVER delete (e.g. the +curated FloraSense repository, which is a Fortress backup target, not a reclaim +source). Unlike a holds-list entry, it is enforced at two independent points +and has no override flag: + + 1. verify() returns the CANONICAL verdict first, so such a source never + enters the SAFE/deletable set — regardless of archive, hold, or drift + state. + 2. safe_to_rm() (re-checked at the actual rm site) refuses the path, so even + a direct delete_source() call with assume_yes + allow_no_snapshot cannot + remove it. + +Stdlib unittest; run from the repo root: + python3 -m unittest discover -s tests -v +""" +import os +import sys +import shutil +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import reclaim + + +class CanonicalRootHelpers(unittest.TestCase): + def test_load_merges_cli_and_env_normalizes_and_dedupes(self): + os.environ["FORTRESS_CANONICAL_ROOTS"] = ( + "/depot/florasense/data/repository/" + os.pathsep + "/depot/other") + try: + roots = reclaim.load_canonical_roots( + ["/depot/florasense/data/repository", " /depot/x/ "]) + finally: + del os.environ["FORTRESS_CANONICAL_ROOTS"] + # trailing slash stripped, whitespace trimmed, duplicate collapsed + self.assertIn("/depot/florasense/data/repository", roots) + self.assertIn("/depot/other", roots) + self.assertIn("/depot/x", roots) + self.assertEqual(len(roots), len(set(roots))) + self.assertEqual( + roots.count("/depot/florasense/data/repository"), 1) + + def test_load_empty_is_empty(self): + self.assertEqual(reclaim.load_canonical_roots(None), []) + + def test_is_canonical_matches_root_and_descendants_only(self): + roots = ["/depot/florasense/data/repository"] + # the root itself and anything nested under it + self.assertTrue(reclaim.is_canonical( + "/depot/florasense/data/repository", roots)) + self.assertTrue(reclaim.is_canonical( + "/depot/florasense/data/repository/X0E", roots)) + self.assertTrue(reclaim.is_canonical( + "/depot/florasense/data/repository/X0E/2_spectral-standoff", roots)) + # a sibling with the root as a string PREFIX is NOT under it + self.assertFalse(reclaim.is_canonical( + "/depot/florasense/data/repository-staging", roots)) + # unrelated path, and the empty-roots fast path + self.assertFalse(reclaim.is_canonical("/depot/florasense/data/landing-zone", roots)) + self.assertFalse(reclaim.is_canonical("/depot/florasense/data/repository", [])) + + +class CanonicalRootEnforcement(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + self.canon = os.path.join(self.tmp, "repository") + self.sub = os.path.join(self.canon, "X0E") + os.makedirs(self.sub) + with open(os.path.join(self.sub, "raw.bin"), "w") as fh: + fh.write("irreplaceable raw data") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_verify_marks_canonical_over_everything_else(self): + # A log that would otherwise be evaluated normally; the canonical check + # is first, so the verdict is CANONICAL regardless of tar/status/holds. + log = {"project": "X0E", "fortress_tar": "/does/not/exist", + "status": "success", "source_files": {}} + r = reclaim.verify(log, self.sub, holds=["X0E"], + canonical_roots=[self.canon]) + self.assertEqual(r["verdict"], reclaim.CANONICAL) + self.assertTrue(any("canonical root" in why for why in r["reasons"])) + + def test_verify_without_canonical_root_does_not_mark_canonical(self): + # Control: same source, but no canonical root declared -> not CANONICAL. + log = {"project": "X0E", "fortress_tar": "/does/not/exist", + "status": "success", "source_files": {}} + r = reclaim.verify(log, self.sub, holds=[], canonical_roots=[]) + self.assertNotEqual(r["verdict"], reclaim.CANONICAL) + + def test_safe_to_rm_refuses_canonical_path(self): + ok, why = reclaim.safe_to_rm(self.sub, canonical_roots=[self.canon]) + self.assertFalse(ok) + self.assertIn("canonical root", why) + + def test_delete_source_never_removes_canonical_even_with_all_overrides(self): + # The strongest possible delete invocation: assume_yes + allow_no_snapshot. + # The canonical guard fires in safe_to_rm before any tar/snapshot logic. + result = {"source_folder": self.sub, "fortress_tar": "/does/not/exist"} + removed = reclaim.delete_source( + result, assume_yes=True, allow_no_snapshot=True, + canonical_roots=[self.canon]) + self.assertFalse(removed) + # The directory and its data are untouched. + self.assertTrue(os.path.isdir(self.sub)) + self.assertTrue(os.path.isfile(os.path.join(self.sub, "raw.bin"))) + + +if __name__ == "__main__": + unittest.main()