From f657aade253cbcf4b4ce8b2385ab8c557f3b08cf Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Thu, 16 Jul 2026 13:08:30 -0400 Subject: [PATCH] feat(reclaim): add manifest_for_project lookup for size-routed targets A deployment wrapper driving both routed and non-routed targets needs a single lookup that finds a target's manifest by project name without re-deriving its badge (badge_of lives in archive.py, not reclaim.py). Mirrors latest_log_for_project's role for the whole-target log path. Co-Authored-By: Claude Sonnet 5 --- reclaim.py | 16 ++++++++++++++++ tests/test_routing_reclaim.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/reclaim.py b/reclaim.py index d8f1a00..b6031de 100644 --- a/reclaim.py +++ b/reclaim.py @@ -721,6 +721,22 @@ def find_manifests(log_dir): return out +def manifest_for_project(log_dir, project): + """ + The size-routing manifest for `project`, or None if it has never routed (no + manifest yet). Mirrors latest_log_for_project()'s role for the whole-target + log path — a deployment wrapper driving BOTH routed and non-routed targets + (e.g. a per-target size_routing config flag) needs a single lookup that + works either way without re-deriving the badge (badge_of lives in archive.py, + not reclaim.py; matching on the manifest's own recorded `project` field avoids + that cross-module dependency). + """ + for m in find_manifests(log_dir): + if m.get("project") == project: + return m + return None + + 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, diff --git a/tests/test_routing_reclaim.py b/tests/test_routing_reclaim.py index 63ffd71..7000b6c 100644 --- a/tests/test_routing_reclaim.py +++ b/tests/test_routing_reclaim.py @@ -174,6 +174,34 @@ def test_empty_when_no_vault(self): self.assertEqual(reclaim.find_manifests(d), []) +class TestManifestForProject(unittest.TestCase): + def test_finds_matching_project(self): + with tempfile.TemporaryDirectory() as d: + vault = os.path.join(d, "_vault") + os.makedirs(vault) + for badge, proj in (("aaa", "repository_X1D_2_spectral-standoff"), + ("bbb", "repository_X1D_3_transcriptomics")): + with open(os.path.join(vault, "%s.manifest.json" % badge), "w") as fh: + json.dump({"badge": badge, "project": proj, "objects": []}, fh) + m = reclaim.manifest_for_project(d, "repository_X1D_3_transcriptomics") + self.assertEqual(m["badge"], "bbb") + + def test_none_when_no_match(self): + with tempfile.TemporaryDirectory() as d: + self.assertIsNone(reclaim.manifest_for_project(d, "repository_X1D_3_transcriptomics")) + + def test_none_when_project_field_absent(self): + # A manifest that predates the 'project' field (shouldn't happen in + # practice — schema_version 1 always writes it — but a lookup must not + # crash or false-match on a bare .get(None) == None comparison). + with tempfile.TemporaryDirectory() as d: + vault = os.path.join(d, "_vault") + os.makedirs(vault) + with open(os.path.join(vault, "aaa.manifest.json"), "w") as fh: + json.dump({"badge": "aaa", "objects": []}, fh) + self.assertIsNone(reclaim.manifest_for_project(d, "repository_X1D_3_transcriptomics")) + + class TestDeleteMultiTar(unittest.TestCase): def setUp(self): self._hsi = reclaim.hsi_exists