Canonical-root invariant: paths reclaim can never delete

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)")

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()