feat(routing): Phase-2 aggregate reclaim — verify_target (slice 3)

reclaim.py

@@ -545,6 +545,164 @@ def verify(log, source_folder, holds, file_pattern=None, canonical_roots=(),
     }
 
 
+def find_manifests(log_dir):
+    """
+    Load every Phase-2 size-routing manifest ({log_dir}/_vault/*.manifest.json).
+    These are the per-target aggregate indexes verify_target() checks; they live in
+    the _vault/ child precisely so find_logs' non-recursive glob never sees them.
+    Unreadable/parse-broken manifests are skipped.
+    """
+    out = []
+    for mp in sorted(glob.glob(os.path.join(log_dir, "_vault", "*.manifest.json"))):
+        try:
+            out.append(load_json(mp))
+        except (ValueError, OSError):
+            pass
+    return out
+
+
+def verify_target(manifest, holds, canonical_roots=(), allowlist=None, opt_in=False):
+    """
+    Per-target AGGREGATE structural double-check for a size-routed target (Phase 2,
+    RFC §2.6). A routed target ships as MANY objects (solos + shards); reclaim's
+    per-log verify() would see only one object's few files and perpetually false-DRIFT.
+    verify_target instead reconstructs the whole target from its manifest:
+
+      expected member set = ⋃ objects' arcnames
+      archived count      = Σ objects' n_files
+      archived bytes      = Σ objects' source_bytes
+      every object tar must be present on tape (hsi_exists per object)
+      newer-than-run is per-object (a live member is drift iff newer than the run
+        of the object that currently owns it)
+
+    Verdict vocabulary + INV-E exclusion re-gate match verify() exactly. Returns the
+    same result-dict shape (plus `all_tars`), so the report and guarded --delete paths
+    work unchanged. The manifest carries `source_folder`/`file_pattern`/`exclusion`, so
+    no config or original log is needed.
+    """
+    project = manifest.get("project")
+    source_folder = manifest.get("source_folder")
+    pattern = manifest.get("file_pattern")
+    objects = manifest.get("objects") or []
+    reasons = []
+
+    expected = set()
+    arc_run = {}          # arcname -> owning object's run datetime
+    all_tars = []
+    archived = 0
+    archived_bytes = 0
+    for o in objects:
+        rdt = run_datetime(o.get("run_timestamp"))
+        for a in (o.get("arcnames") or []):
+            expected.add(a)
+            arc_run[a] = rdt
+        archived += o.get("n_files") or 0
+        archived_bytes += o.get("source_bytes") or 0
+        if o.get("fortress_tar"):
+            all_tars.append(o["fortress_tar"])
+
+    held = any((h == project) or (source_folder and h in source_folder) for h in holds)
+
+    # INV-E: re-prove the archive's recorded exclusions are still git-COVERED. The
+    # manifest carries the same `exclusion` block a log does, so recheck_exclusions
+    # works on it directly.
+    excl_check = recheck_exclusions(manifest, source_folder)
+    still_covered = excl_check["still_covered"]
+
+    # Every object tar must be present on tape (bounded O(#objects) metadata calls).
+    missing_tars = [t for t in all_tars if not hsi_exists(t)]
+    tars_ok = bool(all_tars) and not missing_tars
+
+    # One pattern-scoped live walk: count, bytes, and per-object newer-than-run.
+    live = None
+    live_bytes = None
+    newer_count = 0
+    newer_samples = []
+    if source_folder and os.path.isdir(source_folder):
+        rx = re.compile(pattern) if pattern else None
+        live = 0
+        live_bytes = 0
+        for root, _dirs, files in os.walk(source_folder):
+            for f in files:
+                p = os.path.join(root, f)
+                rel = os.path.relpath(p, source_folder)
+                if rx is not None and not rx.search(rel):
+                    continue
+                if rel in still_covered:
+                    continue   # INV-E re-proven-covered exclusion — not on tape by design
+                live += 1
+                try:
+                    st = os.stat(p)
+                except OSError:
+                    continue
+                live_bytes += st.st_size
+                rdt = arc_run.get(rel)
+                # A member newer than the run of the object that owns it changed since
+                # it was archived. A live file in NO object is new-since-archive; the
+                # count check (live != archived) catches it — leave the newer signal to
+                # genuine modifications of archived members.
+                if rdt is not None and st.st_mtime > rdt.timestamp():
+                    newer_count += 1
+                    if len(newer_samples) < 3:
+                        newer_samples.append(rel)
+
+    # Verdict — most-blocking first, mirroring verify().
+    if is_canonical(source_folder, canonical_roots):
+        verdict = CANONICAL
+        reasons.append("under canonical root — never deletable (hard invariant)")
+    elif not objects or source_folder is None:
+        verdict = FAILED
+        reasons.append("manifest has no objects" if not objects
+                       else "source_folder unknown")
+    elif not tars_ok:
+        verdict = FAILED
+        if not all_tars:
+            reasons.append("no object tars recorded in manifest")
+        else:
+            reasons.append(f"{len(missing_tars)} of {len(all_tars)} object tar(s) NOT "
+                           f"found on Fortress (e.g. {missing_tars[0]})")
+    elif live is None:
+        verdict = GONE
+        reasons.append("source already removed from Depot")
+    elif excl_check.get("applies") and excl_check.get("no_longer_covered"):
+        verdict = DRIFTED
+        gone = excl_check["no_longer_covered"]
+        sample = ", ".join(f"{a} ({v})" for a, v in gone[:3])
+        reasons.append(f"{len(gone)} previously-excluded file(s) no longer git-COVERED "
+                       f"(e.g. {sample}) — current bytes are not on tape")
+    elif newer_count > 0:
+        verdict = DRIFTED
+        reasons.append(f"{newer_count} member(s) newer than the object run that "
+                       f"archived them (e.g. {', '.join(newer_samples)})")
+    elif live != archived:
+        verdict = DRIFTED
+        reasons.append(f"live files ({live}) != archived files ({archived}) across "
+                       f"{len(objects)} object(s)")
+    elif live_bytes != archived_bytes:
+        verdict = DRIFTED
+        reasons.append(f"live bytes ({live_bytes}) != archived bytes ({archived_bytes}) "
+                       f"— same count, content changed")
+    elif held:
+        verdict = HOLD
+        reasons.append("on holds list")
+    elif opt_in and not on_allowlist(project, source_folder, allowlist):
+        verdict = KEEP
+        reasons.append("not on reclaim allowlist (opt-in mode) — kept")
+    else:
+        verdict = SAFE
+        reasons.append(f"{live} files across {len(objects)} object(s) match; nothing "
+                       f"newer than run; all tars present")
+
+    return {
+        "project": project, "source_folder": source_folder,
+        "fortress_tar": (all_tars[0] if all_tars else None), "all_tars": all_tars,
+        "archived": archived, "live": live, "newer": newer_count,
+        "archived_bytes": archived_bytes, "live_bytes": live_bytes,
+        "pattern": pattern, "run": manifest.get("badge"),
+        "verdict": verdict, "reasons": reasons,
+    }
+
+
 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
@@ -603,16 +761,23 @@ def delete_source(result, assume_yes, allow_no_snapshot=False, canonical_roots=(
     never removed, even with --yes and --allow-no-snapshot.
     """
     folder = result["source_folder"]
-    tar = result["fortress_tar"]
+    # A routed target has many object tars; a whole-target archive has one. Re-confirm
+    # EVERY tar (routed: all_tars) fresh, immediately before deleting.
+    tars = result.get("all_tars") or ([result["fortress_tar"]] if result.get("fortress_tar") else [])
 
     ok, why = safe_to_rm(folder, canonical_roots)
     if not ok:
         print(f"  SKIP {folder}: {why}")
         return False
 
     # Fresh re-verify immediately before deleting — never trust the earlier check.
-    if not hsi_exists(tar):
-        print(f"  ABORT {folder}: Fortress tar not confirmed at delete time")
+    if not tars:
+        print(f"  ABORT {folder}: no Fortress tar to confirm at delete time")
+        return False
+    missing = [t for t in tars if not hsi_exists(t)]
+    if missing:
+        print(f"  ABORT {folder}: {len(missing)}/{len(tars)} Fortress tar(s) not "
+              f"confirmed at delete time (e.g. {missing[0]})")
         return False
 
     # Recoverability guard: only delete what a current Depot snapshot can restore.
@@ -806,27 +971,44 @@ def main():
             except (ValueError, OSError):
                 pass
 
+    manifests = []   # Phase-2 size-routed targets, verified as aggregates (verify_target)
     if args.scan:
         seen_projects = set()
         for lpath in find_logs(args.log_dir, days=args.days):
             try:
                 log = load_json(lpath)
             except (ValueError, OSError):
                 continue
+            # Phase-2 routed OBJECT logs are covered by their target's manifest below;
+            # never feed them to the per-log verify() (they'd each false-DRIFT).
+            if log.get("routed_object"):
+                continue
             proj = log.get("project")
             # newest-first: only verify the most recent successful-or-not log per project
             if proj in seen_projects:
                 continue
             seen_projects.add(proj)
             items.append((log, configs_by_project.get(proj)))
+        manifests = find_manifests(args.log_dir)
     elif args.config:
         config = load_json(args.config)
         proj = config["project_name"]
-        lpath = latest_log_for_project(args.log_dir, proj)
-        if not lpath:
-            print(f"No archive log found for project {proj!r} in {args.log_dir}")
-            sys.exit(1)
-        items.append((load_json(lpath), config))
+        if config.get("size_routing"):
+            # Routed target: verify its manifest aggregate, not a per-object log.
+            badge = archive.badge_of(config["project_name"], config["source_folder"],
+                                     config["file_pattern"])
+            mpath = os.path.join(config.get("log_dir") or args.log_dir,
+                                 "_vault", f"{badge}.manifest.json")
+            if not os.path.exists(mpath):
+                print(f"No size-routing manifest for {proj!r} at {mpath}")
+                sys.exit(1)
+            manifests = [load_json(mpath)]
+        else:
+            lpath = latest_log_for_project(args.log_dir, proj)
+            if not lpath:
+                print(f"No archive log found for project {proj!r} in {args.log_dir}")
+                sys.exit(1)
+            items.append((load_json(lpath), config))
     else:
         ap.error("give a config file, or use --scan")
 
@@ -841,6 +1023,12 @@ def matches(log, cfg):
             return args.match in hay
         items = [(log, cfg) for (log, cfg) in items if matches(log, cfg)]
 
+        def manifest_matches(m):
+            hay = " ".join([m.get("project") or "", m.get("source_folder") or ""]
+                           + [o.get("fortress_tar") or "" for o in (m.get("objects") or [])])
+            return args.match in hay
+        manifests = [m for m in manifests if manifest_matches(m)]
+
     # Verify and report. Attach scratch-zip locators (for post-delete cleanup):
     # zip_name from the log + tmp_dir from the config.
     results = []
@@ -851,6 +1039,14 @@ def matches(log, cfg):
         r["zip_name"] = log.get("zip_name")
         r["tmp_dir"] = (cfg or {}).get("tmp_dir")
         results.append(r)
+    # Phase-2 routed targets: one aggregate verdict per manifest (verify_target). No
+    # staging zip to clean up (objects are tars of source), so zip_name/tmp_dir are None.
+    for m in manifests:
+        r = verify_target(m, holds, canonical_roots=canonical_roots,
+                          allowlist=allowlist, opt_in=opt_in)
+        r["zip_name"] = None
+        r["tmp_dir"] = None
+        results.append(r)
     results.sort(key=lambda r: (r["verdict"] != SAFE, r["project"] or ""))
 
     display = [r for r in results if not (args.skip_gone and r["verdict"] == GONE)]