diff --git a/docs/HOLDS_POLICY.md b/docs/HOLDS_POLICY.md new file mode 100644 index 0000000..0166e85 --- /dev/null +++ b/docs/HOLDS_POLICY.md @@ -0,0 +1,115 @@ +# Reclaim holds policy — typed holds + promote receipts + +**Status:** engine side implemented (`reclaim.py`). The receipt *writer* and the Teams +review card are deployment glue (florasense-tools) and land separately. + +`reclaim.py` decides which archived Depot sources are **held** vs **reclaimable**. The +flat `reclaim_holds.txt` conflated distinct reasons for holding, so holds are now **typed** +by *why* a source is held, which yields three review tiers. + +## The three tiers + +| Tier | `reason` (+ flag) | Verdict | Meaning | +|---|---|---|---| +| **Hold** | `business` | `HOLD` | A human/downstream needs it live (e.g. GeneWiz_X1C until Amanda relocates it). Never auto-reclaimed; `review_date` can drive a nightly nag. | +| **Held (pending receipt)** | `promote-pending`, no valid receipt | `HOLD` | A staging copy of data being promoted into `repository/`. Held until a content-coverage receipt proves the promote landed. | +| **Needs approval** | `promote-pending`, valid receipt | `AWAITING-APPROVAL` | Promotion is **proven**; the source is *eligible*, but a human approves the actual delete. This is what the Teams review card surfaces. | +| **Held (pending record)** | `superseded`, no record | `HOLD` | A corrected/reprocessed/deduped copy (DECISIONS #25). Held until a supersession provenance record exists. | +| **Needs approval** | `superseded`, valid record | `AWAITING-APPROVAL` | The supersession record exists (the "why" is preserved); *eligible*, but a human approves the delete. | + +> **There is no auto-reclaim.** Reclaim never deletes on its own: it is **report-only** +> unless `--delete` is passed, and `--delete` confirms **per folder** (unless `--yes`). So +> "don't reclaim by default" is structural, not a flag that could be flipped. A valid-receipt +> promote-pending source becomes `AWAITING-APPROVAL` — eligible, but a human runs the guarded +> delete. (No `auto_reclaim` field; if unattended sweeping is ever wanted it is a deliberate +> future feature with its own review, not a dormant default.) + +**Nothing un-promoted is ever reclaimed.** The receipt *is* the "has it landed in +`repository/`?" proof — no valid receipt → `HOLD`, full stop. And even an approved delete +still clears the existing hard gates below. + +## `reclaim_holds.json` + +```json +{ + "_comment": "Typed reclaim holds. reason = business | promote-pending | superseded.", + "holds": [ + { "match": "GeneWiz_X1C", "reason": "business", + "owner": "Amanda", "review_date": "2026-12-01", + "note": "keep on Depot until relocated" }, + + { "match": "landing-zone/AAPF", "reason": "promote-pending", + "receipt_dir": "/depot/florasense/etc/logs_fortress/promote_receipts", + "note": "eligible once fs-verify-promoted confirms the copy in repository/" }, + + { "match": "X0H_Soy/2_spectral-standoff/hsi/mishap_original", "reason": "superseded", + "record": "/depot/florasense/data/repository/X0H_Soy/2_spectral-standoff/hsi/provenance/X0H_Soy_AAPF_mishap_PROVENANCE.md", + "note": "AAPF barcode-swap; eligible once the supersession record exists (DECISIONS #25)" } + ] +} +``` + +- `match` — exact `project_name` **or** a substring of `source_folder` (same matching as the + old flat list). +- Legacy **`reclaim_holds.txt`** still works: each line is treated as a `business` hold, so + existing deployments are unchanged until they move to JSON. + +## The receipt (content-coverage) + +Written by the deployment's promote-verify tool (florasense-tools `fs-verify-promoted`) into +`receipt_dir` as JSON: + +```json +{ "source_folder": "/depot/.../landing-zone/AAPF", + "file_count": 68382, "all_present": true, + "verified_at": "2026-07-09T14:30:00", "tool": "fs-verify-promoted" } +``` + +`fs-verify-promoted` proves `set(md5 of every landing-zone file) ⊆ set(md5 under +repository/)` (a real byte-coverage check; it must **reject symlinks** as a "promoted copy", +or a `repository/` symlink back into the landing-zone would falsely pass). `reclaim.py` is +content-agnostic — it trusts the receipt but re-checks it is **current**: +`receipt_valid()` requires `all_present` **AND** `file_count == live count` **AND** no live +file newer than `verified_at`. Any problem (no receipt, stale, count mismatch) → held. + +## Hard gates that still apply to every delete + +The tiering only decides eligibility; a `--delete` additionally requires **all** of: + +1. the source is **archived + round-trip-verified on Fortress** (recoverable from tape); +2. not under a **canonical root** (`/depot/florasense/data/repository` — the curated tree, + never deletable, no override); +3. a **snapshot** proves recoverability — **every live file present in the newest NetApp + snapshot with matching size and mtime** (not merely a file-count ≥ check; that missed a + file *replaced* since the snapshot, same count → old bytes, and renames). Override only + with `--allow-no-snapshot`; +4. a path-depth guard (under `/depot/`, ≥ 4 levels). + +## Related deployment gates (florasense-tools) + +Two deployed gates wrap the engine's reclaim and this policy should be read alongside them: + +- **Supersession-record gate** (florasense `DECISIONS #25` + `docs/data-supersession.md` + + `docs/templates/supersession-record-template.md`). Deleting a **superseded** copy + (corrected / re-delivered / reprocessed / deduped data — *distinct* from a routine + single-copy reclaim of a stale-but-unchanged source) requires a human-authored + **supersession record**, written in **both** the Fortress archive **and** the modality's + repository `provenance/` dir, carrying the integrity/MD5 verdict that justifies the delete. + **The engine now enforces this** via the `superseded` tier above: a matching source stays + `HOLD` until its `record` (the repository `provenance/` copy — it lives under the canonical + root, so it persists) exists, is non-empty, and is actually filled in (not a raw template). + The record's *content* (the MD5 verdict) is human judgment the engine trusts; its + *existence* is the machine-checkable gate #25 requires. The Fortress-archive copy is ensured + by the archive step (it is on tape, not re-checked here). +- **`fs-coverage-audit`** (weekly; `docs/negishi-cron-setup.md`) covers reclaim's blind spot: + reclaim only sees sources that have an archive log, so it cannot flag **never-archived / + single-copy / FAILED** experiments. The trio: coverage-audit = "never backed up", + `fs-reclaim-report` = "backup is stale", reclaim = consumes both. + +## Division of labor + +- **Engine (`reclaim.py`, this repo):** typed-hold parsing, the tiering + `AWAITING-APPROVAL` + verdict, the content-agnostic receipt hook, the hardened snapshot guard. +- **Deployment (florasense-tools):** `fs-verify-promoted` (writes receipts), and the + review-only Teams card (extends `fs-reclaim-report`, posts the `AWAITING-APPROVAL` + + `HOLD` tiers nightly-when-non-empty). Approve out-of-band via the guarded `--delete`. diff --git a/reclaim.py b/reclaim.py index c6c4088..9622134 100644 --- a/reclaim.py +++ b/reclaim.py @@ -97,6 +97,10 @@ # no flag to override — see load_canonical_roots) KEEP = "KEEP" # opt-in mode: archived & verified but NOT on the allowlist # -> kept (fail-safe default; only allowlisted sources are SAFE) +AWAITING_APPROVAL = "AWAITING-APPROVAL" # promote-pending hold whose content-coverage + # receipt PROVES the copy landed in repository/ -> ELIGIBLE, but a + # human runs the guarded delete (there is no auto-reclaim path; see + # docs/HOLDS_POLICY.md). This is the "needs approval" tier. def load_json(path): @@ -106,19 +110,173 @@ def load_json(path): def load_holds(holds_path): """ - Load a holds list: one project_name (or path substring) per line, '#' comments - allowed. Projects/paths matching any entry are never deleted (verdict HOLD). + Load the holds list as NORMALIZED TYPED entries (docs/HOLDS_POLICY.md). Returns a + list of dicts, each at least {match, reason, ...}. Two input formats, auto-detected: + + *.json — {"holds": [ {match, reason, owner, review_date, receipt_dir, record, + note}, ... ]}. reason is 'business' (a human owns it — + never reclaimed), 'promote-pending' (a staging copy; eligible only once + a content-coverage receipt proves the promote landed — receipt_dir), or + 'superseded' (DECISIONS #25: a corrected/reprocessed copy; eligible only + once a supersession provenance record exists — record). See classify_hold. + Unknown fields are carried through harmlessly. + *.txt / legacy — one project_name or path-substring per line ('#' comments). + Each line becomes {match: , reason: 'business'} — i.e. the old + flat holds behaved exactly like a business hold (never auto), so + existing deployments are unchanged until they move to JSON. + + Unknown/missing path -> []. A malformed JSON hold list raises (fail-loud: a + broken holds file must not be silently treated as "no holds"). """ - holds = [] - if holds_path and os.path.exists(holds_path): + if not holds_path or not os.path.exists(holds_path): + return [] + if holds_path.endswith(".json"): with open(holds_path) as fh: - for line in fh: - line = line.split("#", 1)[0].strip() - if line: - holds.append(line) + data = json.load(fh) + out = [] + for h in (data.get("holds") or []): + if not h.get("match"): + continue + e = dict(h) + e.setdefault("reason", "business") + out.append(e) + return out + holds = [] + with open(holds_path) as fh: + for line in fh: + line = line.split("#", 1)[0].strip() + if line: + holds.append({"match": line, "reason": "business"}) return holds +def hold_entry_for(holds, project, source_folder): + """First holds entry matching this source (match == project_name, OR match is a + substring of source_folder) — mirrors the legacy holds/allowlist matching. Returns + the NORMALIZED entry dict, or None if unheld. Tolerates both typed dict entries + (from load_holds) and bare strings (a str is a legacy business hold), so any caller + passing a plain string list still works.""" + for h in (holds or []): + entry = {"match": h, "reason": "business"} if isinstance(h, str) else h + m = entry.get("match") + if m and ((m == project) or (source_folder and m in source_folder)): + return entry + return None + + +def receipt_valid(entry, source_folder, live_count): + """ + Is there a current content-coverage receipt proving this promote-pending source's + bytes all landed in repository/ (docs/HOLDS_POLICY.md)? Returns (ok, detail). + + The receipt is written by the deployment's promote-verify tool (florasense-tools + fs-verify-promoted) into entry['receipt_dir'] as JSON: + {source_folder, file_count, all_present (bool), verified_at (YYYY-MM-DDTHH:MM:SS), tool} + It is VALID iff: all_present is true AND file_count == live_count AND NO live file + is newer than verified_at (the source hasn't changed since it was verified). This + is content-AGNOSTIC here (reclaim trusts the receipt); the actual md5 ⊆ check lives + in the deployment tool. FAIL-SAFE: any problem (no receipt_dir, no receipt for this + source, unreadable, stale) -> (False, reason) -> the source stays HELD. + """ + rdir = entry.get("receipt_dir") + if not rdir or not os.path.isdir(rdir): + return (False, "no receipt_dir") + best = None + for name in os.listdir(rdir): + if not name.endswith(".json"): + continue + try: + r = load_json(os.path.join(rdir, name)) + except (ValueError, OSError): + continue + if os.path.normpath(r.get("source_folder") or "") != os.path.normpath(source_folder): + continue + va = r.get("verified_at") or "" + try: + vdt = datetime.datetime.strptime(va, "%Y-%m-%dT%H:%M:%S") + except ValueError: + continue + if best is None or vdt > best[0]: + best = (vdt, r) + if best is None: + return (False, "no valid receipt for this source") + vdt, r = best + if not r.get("all_present"): + return (False, f"receipt {vdt:%Y-%m-%d} says all_present=false") + if r.get("file_count") != live_count: + return (False, f"receipt file_count ({r.get('file_count')}) != live ({live_count})") + newer, _samples = files_newer_than(source_folder, vdt) + if newer: + return (False, f"{newer} file(s) changed since the receipt ({vdt:%Y-%m-%d %H:%M})") + return (True, f"promote verified {vdt:%Y-%m-%d %H:%M} ({r.get('file_count')} files)") + + +def supersession_record_valid(entry, source_folder): + """ + DECISIONS #25 (florasense docs/data-supersession.md): a SUPERSEDED copy (corrected / + re-delivered / reprocessed / deduplicated data) may be deleted only once a supersession + provenance record (a Markdown file) EXISTS — the deletion is gated on the record. The + record is human-authored: its integrity/MD5 verdict is human judgment, so the engine + enforces the part #25 makes machine-checkable — the record is present, non-empty, and + actually filled in (not a raw template copy). Returns (ok, detail). + + entry['record'] is the path to the record, normally the repository `provenance/` copy + (e.g. {EXP}/2_spectral-standoff/{hsi,rgb}/provenance/*.md) — which lives under the + canonical root and so persists. The Fortress-archive copy is ensured by the archive + step, not re-checked here (it is on tape). FAIL-SAFE: any problem -> held. + """ + rec = entry.get("record") + if not rec: + return (False, "no 'record' path in the hold entry") + if not os.path.isfile(rec): + return (False, f"supersession record not found at {rec}") + try: + with open(rec) as fh: + text = fh.read() + except OSError as e: + return (False, f"cannot read supersession record: {e}") + if not text.strip(): + return (False, "supersession record is empty") + # Footgun guard: an unfilled template copy still carries the template's own + # instruction to delete its comments once filled in. + if "Delete these instructional comments once the record is filled in" in text: + return (False, "supersession record is an unfilled template") + return (True, os.path.basename(rec)) + + +def classify_hold(entry, source_folder, live_count): + """Given a matching holds entry, return (verdict, reason) for the source. Applied + ONLY when the source is otherwise SAFE (archived, verified, not drifted). + + There is NO auto-reclaim path: reclaim never deletes on its own — it is report-only + unless --delete, and --delete confirms per folder (unless --yes). So a source that + clears its tier's evidence gate becomes AWAITING_APPROVAL (eligible, a human runs the + guarded delete), never a silent SAFE. "Don't reclaim by default" is structural. + + Tiers: business (a human owns it -> HOLD); promote-pending (a staging copy; held until + a content-coverage receipt proves the promote landed); superseded (DECISIONS #25: a + corrected/reprocessed copy; held until a supersession provenance record exists).""" + reason = entry.get("reason", "business") + owner = entry.get("owner") + if reason == "business": + who = f" (owner {owner})" if owner else "" + return (HOLD, f"business hold{who} — a human owns this; never auto-reclaimed") + if reason == "promote-pending": + ok, detail = receipt_valid(entry, source_folder, live_count) + if not ok: + return (HOLD, f"promote-pending: held until promote is verified — {detail}") + return (AWAITING_APPROVAL, f"{detail} — eligible; awaiting your approval to free Depot") + if reason == "superseded": + ok, detail = supersession_record_valid(entry, source_folder) + if not ok: + return (HOLD, f"superseded: held until a supersession record exists " + f"(DECISIONS #25) — {detail}") + return (AWAITING_APPROVAL, f"supersession record present ({detail}) — eligible; " + f"awaiting your approval to delete the superseded copy") + # Unknown reason -> fail-safe HOLD (never silently deletable). + return (HOLD, f"hold reason {reason!r} unrecognized — kept (fail-safe)") + + def load_allowlist(allowlist_path): """Load an opt-in reclaim allowlist (or None if no path given). @@ -455,7 +613,7 @@ def verify(log, source_folder, holds, file_pattern=None, canonical_roots=(), # walk so it surfaces as drift below — never silently SAFE, never --delete'd. excl_check = recheck_exclusions(log, source_folder) - held = any((h == project) or (source_folder and h in source_folder) for h in holds) + hold = hold_entry_for(holds, project, source_folder) # Fast structural signals status = log.get("status") @@ -523,9 +681,11 @@ def verify(log, source_folder, holds, file_pattern=None, canonical_roots=(), verdict = DRIFTED reasons.append(f"live bytes ({live_bytes}) != archived bytes " f"({archived_bytes}) — same count/mtime, content changed") - elif held: - verdict = HOLD - reasons.append("on holds list") + elif hold is not None: + # Typed hold tiering (docs/HOLDS_POLICY.md): business -> HOLD; promote-pending + # -> HOLD until a valid receipt, then AWAITING_APPROVAL (auto-reclaim disabled). + verdict, hreason = classify_hold(hold, source_folder, live) + reasons.append(hreason) elif opt_in and not on_allowlist(project, source_folder, allowlist): # Opt-in (allowlist) mode: archived + verified, but not explicitly # allowed for reclaim → keep. Fail-safe: forgetting to allowlist @@ -560,19 +720,37 @@ def safe_to_rm(folder, canonical_roots=()): return (True, "") +def _attr_catalog(root, skip=None): + """Metadata-only {relpath: (size, floored_mtime_s)} under root. Skips a named + child dir (used to keep the live walk out of .snapshots). Files that vanish + mid-walk are dropped.""" + cat = {} + for dp, dns, fns in os.walk(root): + if skip and skip in dns: + dns.remove(skip) + for f in fns: + p = os.path.join(dp, f) + try: + st = os.stat(p) + except OSError: + continue + cat[os.path.relpath(p, root)] = (st.st_size, int(st.st_mtime)) + return cat + + def snapshot_present(folder, live_count): """ - Confirm `folder` is captured COMPLETELY in the newest Depot (NetApp) snapshot, - so deleting it is fully recoverable. NetApp exposes `/.snapshots//` - as the contents of as of that snapshot (accessed by path directly — the - dir is hidden and may not appear in a listing of the parent). - - Rather than just checking a snapshot is non-empty, compare the newest - snapshot copy's file count against the live count and require - snapshot >= live: every file currently on disk must be in the snapshot. - This catches the dangerous case where files were added since the last - snapshot (newest snapshot would be short -> not yet recoverable). - Returns (ok, detail). + Confirm `folder` is captured COMPLETELY in the newest Depot (NetApp) snapshot, so + deleting it is fully recoverable. NetApp exposes `/.snapshots//` as the + contents of as of that snapshot (accessed by path directly — the dir is + hidden and may not appear in a listing of the parent). + + Every LIVE file must be present in the newest snapshot with MATCHING size and + (whole-second) mtime — not merely a file-count comparison. Count-only (the earlier + check) passed on a snapshot that had *at least as many* files, so it silently + missed a file REPLACED since the snapshot (same count, snapshot holds the old + bytes) or RENAMES (same count, different paths). The set+attr check proves the + exact current bytes are recoverable. Returns (ok, detail). Metadata-only walk. """ snapdir = os.path.join(folder, SNAPSHOT_DIRNAME) if not os.path.isdir(snapdir): @@ -589,11 +767,20 @@ def snapshot_present(folder, live_count): # Snapshot names are @GMT-YYYY.MM.DD-HH.MM.SS, so lexical sort == chronological. newest = snaps[-1] - snap_count = count_live_files(os.path.join(snapdir, newest)) or 0 - if snap_count >= live_count: - return (True, f"newest snapshot {newest} holds {snap_count} files (live {live_count})") - return (False, f"newest snapshot {newest} holds {snap_count} files < live {live_count} " - f"— files added since last snapshot are not captured yet") + snap_cat = _attr_catalog(os.path.join(snapdir, newest)) + live_cat = _attr_catalog(folder, skip=SNAPSHOT_DIRNAME) + + missing = [] # live files absent OR size/mtime-mismatched in the snapshot + for rel, attr in live_cat.items(): + if snap_cat.get(rel) != attr: + missing.append(rel) + if not missing: + return (True, f"newest snapshot {newest} fully captures all {len(live_cat)} " + f"live files (size+mtime match)") + sample = ", ".join(sorted(missing)[:3]) + return (False, f"newest snapshot {newest} is missing or stale for {len(missing)} of " + f"{len(live_cat)} live files (e.g. {sample}) — those current bytes are " + f"NOT recoverable from snapshot") def delete_source(result, assume_yes, allow_no_snapshot=False, canonical_roots=()): @@ -871,7 +1058,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, KEEP, CANONICAL, GONE] + order = [SAFE, AWAITING_APPROVAL, DRIFTED, FAILED, RUNNING, HOLD, KEEP, 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. diff --git a/tests/test_typed_holds.py b/tests/test_typed_holds.py new file mode 100644 index 0000000..531832d --- /dev/null +++ b/tests/test_typed_holds.py @@ -0,0 +1,256 @@ +""" +Tests for typed reclaim holds + the hardened snapshot guard (docs/HOLDS_POLICY.md). + +Tiers: business -> HOLD (never auto); promote-pending -> HOLD until a valid +content-coverage receipt, then AWAITING-APPROVAL (auto-reclaim disabled). Legacy +flat .txt holds behave as business holds (unchanged). The snapshot guard now +requires every live file present in the newest snapshot with matching size+mtime, +not merely a file-count >= comparison. +""" +import datetime +import json +import os +import shutil +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import reclaim # noqa: E402 + + +def write_receipt(rdir, source_folder, file_count, all_present=True, verified_at="2020-01-01T00:00:00"): + os.makedirs(rdir, exist_ok=True) + p = os.path.join(rdir, "receipt.json") + with open(p, "w") as fh: + json.dump({"source_folder": source_folder, "file_count": file_count, + "all_present": all_present, "verified_at": verified_at, + "tool": "test"}, fh) + return p + + +class TestLoadHolds(unittest.TestCase): + def test_txt_is_business(self): + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "holds.txt") + with open(p, "w") as fh: + fh.write("# c\nGeneWiz_X1C\n/depot/x/landing-zone/AAPF\n") + hs = reclaim.load_holds(p) + self.assertEqual([h["reason"] for h in hs], ["business", "business"]) + self.assertEqual(hs[0]["match"], "GeneWiz_X1C") + + def test_json_typed(self): + with tempfile.TemporaryDirectory() as d: + p = os.path.join(d, "reclaim_holds.json") + _payload = {"holds": [ + {"match": "GeneWiz_X1C", "reason": "business", "owner": "Amanda"}, + {"match": "landing-zone/AAPF", "reason": "promote-pending", + "receipt_dir": "/r"}, + {"reason": "business"}, # no match -> dropped + ]} + with open(p, "w") as fh: + json.dump(_payload, fh) + hs = reclaim.load_holds(p) + self.assertEqual(len(hs), 2) + self.assertEqual(hs[1]["reason"], "promote-pending") + + def test_missing(self): + self.assertEqual(reclaim.load_holds("/no/such.json"), []) + self.assertEqual(reclaim.load_holds(None), []) + + +class TestHoldEntryFor(unittest.TestCase): + def test_string_and_dict_and_match_modes(self): + # legacy bare string -> business + e = reclaim.hold_entry_for(["X0E"], "X0E", "/depot/x/X0E") + self.assertEqual(e, {"match": "X0E", "reason": "business"}) + # dict, substring of source_folder + typed = [{"match": "landing-zone/AAPF", "reason": "promote-pending"}] + self.assertIsNotNone(reclaim.hold_entry_for(typed, "p", "/depot/landing-zone/AAPF/x")) + # no match + self.assertIsNone(reclaim.hold_entry_for(typed, "other", "/depot/other")) + + +class TestReceiptValid(unittest.TestCase): + def test_no_receipt_dir(self): + ok, _ = reclaim.receipt_valid({"receipt_dir": "/no/such"}, "/s", 3) + self.assertFalse(ok) + + def test_valid(self): + with tempfile.TemporaryDirectory() as d: + write_receipt(d, "/no/such/src", 3) # nonexistent src -> nothing "newer" + ok, _ = reclaim.receipt_valid({"receipt_dir": d}, "/no/such/src", 3) + self.assertTrue(ok) + + def test_count_mismatch(self): + with tempfile.TemporaryDirectory() as d: + write_receipt(d, "/no/such/src", 2) + ok, _ = reclaim.receipt_valid({"receipt_dir": d}, "/no/such/src", 3) + self.assertFalse(ok) + + def test_stale_when_file_newer_than_receipt(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as rd: + with open(os.path.join(src, "a.bin"), "w") as fh: + fh.write("x") # mtime ~now > 2020 + write_receipt(rd, src, 1, verified_at="2020-01-01T00:00:00") + ok, detail = reclaim.receipt_valid({"receipt_dir": rd}, src, 1) + self.assertFalse(ok) + self.assertIn("since the receipt", detail) + + +class TestClassifyHold(unittest.TestCase): + def test_business(self): + v, _ = reclaim.classify_hold({"reason": "business", "owner": "Amanda"}, "/s", 3) + self.assertEqual(v, reclaim.HOLD) + + def test_promote_pending_no_receipt_holds(self): + v, _ = reclaim.classify_hold({"reason": "promote-pending", "receipt_dir": "/no"}, "/s", 3) + self.assertEqual(v, reclaim.HOLD) + + def test_promote_pending_valid_receipt_awaits_approval(self): + with tempfile.TemporaryDirectory() as d: + write_receipt(d, "/no/such/src", 3) + v, _ = reclaim.classify_hold( + {"reason": "promote-pending", "receipt_dir": d}, "/no/such/src", 3) + self.assertEqual(v, reclaim.AWAITING_APPROVAL) + + def test_stray_auto_reclaim_field_ignored(self): + # There is no auto-reclaim path; a leftover auto_reclaim field has no effect — + # a valid-receipt promote-pending source is always AWAITING_APPROVAL, never SAFE. + with tempfile.TemporaryDirectory() as d: + write_receipt(d, "/no/such/src", 3) + v, _ = reclaim.classify_hold( + {"reason": "promote-pending", "receipt_dir": d, "auto_reclaim": True}, + "/no/such/src", 3) + self.assertEqual(v, reclaim.AWAITING_APPROVAL) + + def test_unknown_reason_fails_safe_to_hold(self): + v, _ = reclaim.classify_hold({"reason": "weird"}, "/s", 3) + self.assertEqual(v, reclaim.HOLD) + + def test_superseded_no_record_holds(self): + v, r = reclaim.classify_hold({"reason": "superseded", "record": "/no/rec.md"}, "/s", 3) + self.assertEqual(v, reclaim.HOLD) + self.assertIn("DECISIONS #25", r) + + def test_superseded_valid_record_awaits_approval(self): + with tempfile.TemporaryDirectory() as d: + rec = os.path.join(d, "X0H_mishap_PROVENANCE.md") + with open(rec, "w") as fh: + fh.write("# Supersession record\n## Integrity evidence\nMD5 verdict: kept " + "copy intact; superseded copy redundant (job 40434526).\n") + v, _ = reclaim.classify_hold( + {"reason": "superseded", "record": rec}, "/s", 3) + self.assertEqual(v, reclaim.AWAITING_APPROVAL) + + def test_superseded_unfilled_template_holds(self): + with tempfile.TemporaryDirectory() as d: + rec = os.path.join(d, "template.md") + with open(rec, "w") as fh: + fh.write("# Supersession record\n\n") + v, _ = reclaim.classify_hold({"reason": "superseded", "record": rec}, "/s", 3) + self.assertEqual(v, reclaim.HOLD) + + +class TestSnapshotHardening(unittest.TestCase): + def _make(self, d): + """source with 3 files + a .snapshots/@GMT snapshot mirroring them (size+mtime).""" + src = os.path.join(d, "src") + os.makedirs(src) + for i in (1, 2, 3): + with open(os.path.join(src, "f%d.bin" % i), "wb") as fh: + fh.write(b"x" * (10 + i)) + snap = os.path.join(src, reclaim.SNAPSHOT_DIRNAME, "@GMT-2026.01.01-00.00.00") + os.makedirs(snap) + for i in (1, 2, 3): + shutil.copy2(os.path.join(src, "f%d.bin" % i), os.path.join(snap, "f%d.bin" % i)) + return src + + def test_exact_match_ok(self): + with tempfile.TemporaryDirectory() as d: + src = self._make(d) + ok, _ = reclaim.snapshot_present(src, 3) + self.assertTrue(ok) + + def test_replaced_file_same_count_fails(self): + # THE key improvement: a file replaced since the snapshot (count unchanged) + # must fail, where the old count>=live check passed. + with tempfile.TemporaryDirectory() as d: + src = self._make(d) + with open(os.path.join(src, "f2.bin"), "wb") as fh: + fh.write(b"y" * 999) # different size -> not the snapshot's bytes + ok, detail = reclaim.snapshot_present(src, 3) + self.assertFalse(ok) + self.assertIn("f2.bin", detail) + + def test_added_file_fails(self): + with tempfile.TemporaryDirectory() as d: + src = self._make(d) + with open(os.path.join(src, "f4_new.bin"), "wb") as fh: + fh.write(b"z") + ok, _ = reclaim.snapshot_present(src, 4) + self.assertFalse(ok) + + def test_mtime_change_fails(self): + with tempfile.TemporaryDirectory() as d: + src = self._make(d) + os.utime(os.path.join(src, "f1.bin"), (10 ** 9, 10 ** 9)) # far-past mtime + ok, _ = reclaim.snapshot_present(src, 3) + self.assertFalse(ok) + + def test_extra_in_snapshot_still_ok(self): + # deleted-since-snapshot (snapshot has MORE) is fine: live is a subset. + with tempfile.TemporaryDirectory() as d: + src = self._make(d) + snap = os.path.join(src, reclaim.SNAPSHOT_DIRNAME, "@GMT-2026.01.01-00.00.00") + with open(os.path.join(snap, "old_deleted.bin"), "wb") as fh: + fh.write(b"q") + ok, _ = reclaim.snapshot_present(src, 3) + self.assertTrue(ok) + + def test_no_snapshot_dir(self): + with tempfile.TemporaryDirectory() as d: + os.makedirs(os.path.join(d, "src")) + ok, _ = reclaim.snapshot_present(os.path.join(d, "src"), 1) + self.assertFalse(ok) + + +class TestVerifyTiering(unittest.TestCase): + """End-to-end through verify(): an otherwise-SAFE source + a typed hold.""" + def setUp(self): + self._hsi, self._scan = reclaim.hsi_exists, reclaim.scan_source + reclaim.hsi_exists = lambda tar: True + reclaim.scan_source = lambda folder, pattern, run_dt, **kw: (3, 0, [], None) + + def tearDown(self): + reclaim.hsi_exists, reclaim.scan_source = self._hsi, self._scan + + def _log(self): + return {"project": "AAPF", "fortress_tar": "/t.tar", "status": "success", + "source_files": {"a": "1", "b": "2", "c": "3"}, # archived=3 == live 3 + "run_timestamp": "AAPF_20990101_000000"} # nothing newer + + def _verdict(self, holds): + return reclaim.verify(self._log(), "/depot/x/landing-zone/AAPF", holds)["verdict"] + + def test_business_hold(self): + self.assertEqual(self._verdict([{"match": "AAPF", "reason": "business"}]), reclaim.HOLD) + + def test_promote_pending_no_receipt_holds(self): + self.assertEqual( + self._verdict([{"match": "AAPF", "reason": "promote-pending", "receipt_dir": "/no"}]), + reclaim.HOLD) + + def test_promote_pending_valid_receipt_awaits_approval(self): + with tempfile.TemporaryDirectory() as rd: + write_receipt(rd, "/depot/x/landing-zone/AAPF", 3) + v = self._verdict([{"match": "AAPF", "reason": "promote-pending", "receipt_dir": rd}]) + self.assertEqual(v, reclaim.AWAITING_APPROVAL) + + def test_no_hold_is_safe(self): + self.assertEqual(self._verdict([]), reclaim.SAFE) + + +if __name__ == "__main__": + unittest.main()