feat(orchestrator): W3 artifacts tree, ancestry, gates and token-free cleanup

Artifacts show the branch tree with ancestry and gate marks. The cleanup runner calls pkgmgr branch cleanup with its real argv and parses its repos[*].items output, refuses any plan with a remote-branch action, marks an item done only when pkgmgr reports it done, and offers no delete for a branch a worktree holds. Every text states what the pkgmgr command does to volumes. Committed below the bar on the operator's word 2026-11-12 (no further per-bundle tests, one gate after the merge): the W3 e2e never ran; the client switches to --local once pkgmgr ships it.

🎯 Fix: 90 %
💥 Break: 6 %
💸 Cost: 30 %
⏱️ Time: 40 %
🛡️ Security: 100 %
🏆 Score: 100
🚦 Gate: ⛔ break 6 % > 5

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_hub_cleanup.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import ast
 import importlib.util
 import json
+import os
 import secrets
+import shutil
 import subprocess
 import sys
 import tempfile
@@ -11,16 +14,18 @@ import unittest
 from pathlib import Path
 
 from hub_case import HubCase
-from hub_support import HUB, ORIGIN, key, signing, verifier
+from hub_support import GIT_ENV, HUB, ORIGIN, key, signing, verifier
 
 CLEANUP = HUB / "client" / "cleanup.py"
 CLEANUP_SPEC = importlib.util.spec_from_file_location("kvb_orchestrator_cleanup_runner", CLEANUP)
 host = importlib.util.module_from_spec(CLEANUP_SPEC)
 CLEANUP_SPEC.loader.exec_module(host)
+INSTALLED_PKGMGR = shutil.which("pkgmgr")
 DONE_TIP = "5b1e6fd3a7c0e2d94f8a61b3c2d7e9f0a1b2c3d4"
 OTHER_TIP = "1111111111111111111111111111111111111111"
-# Param: format with python=<interpreter>; plan.json beside it is the dry-run plan, next-tree.json the rewritten tree.
+# Param: format with python=<interpreter>; parses argv as pkgmgr's branch_cmd.py does (argparse exits 2 on any other flag) and prints the plan as gc.py _render does, {"host_pid_namespace", "repos": [{"repo", "default", "merged_into", "warnings", "items"}]}; plan.json beside it is the plan's items, next-tree.json the rewritten tree, outcome.txt the result --apply reports per item (done without it), exit 0 also for a skipped item.
 FAKE_PKGMGR = """#!{python}
+import argparse
 import json
 import os
 import sys
@@ -29,22 +34,48 @@ from pathlib import Path
 here = Path(__file__).parent
 with (here / "calls.jsonl").open("a", encoding="utf-8") as log:
     log.write(json.dumps({{"argv": sys.argv[1:], "cwd": os.getcwd()}}) + "\\n")
-if sys.argv[1:3] == ["branch", "tree"]:
+parser = argparse.ArgumentParser(prog="pkgmgr")
+branch = parser.add_subparsers(dest="command", required=True).add_parser("branch").add_subparsers(dest="subcommand", required=True)
+tree = branch.add_parser("tree")
+tree.add_argument("--format", choices=("json", "yaml"), default="json")
+tree.add_argument("--output")
+cleanup = branch.add_parser("cleanup")
+for flag in ("--apply", "--json", "--all", "--sizes"):
+    cleanup.add_argument(flag, action="store_true")
+for option in ("--branch", "--exclude", "--merged-into"):
+    cleanup.add_argument(option, action="append", default=[])
+cleanup.add_argument("--min-age", type=float)
+args = parser.parse_args()
+if args.subcommand == "tree":
     following = here / "next-tree.json"
     rewritten = following.read_text(encoding="utf-8") if following.exists() else json.dumps({{"rewritten": True}})
-    Path(sys.argv[sys.argv.index("--output") + 1]).write_text(rewritten, encoding="utf-8")
+    Path(args.output).write_text(rewritten, encoding="utf-8")
     raise SystemExit(0)
-print(json.dumps({{"items": json.loads((here / "plan.json").read_text(encoding="utf-8"))}}))
+items = json.loads((here / "plan.json").read_text(encoding="utf-8"))
+if args.apply:
+    outcome = here / "outcome.txt"
+    result = outcome.read_text(encoding="utf-8") if outcome.exists() else "done"
+    items = [{{**item, "result": result}} for item in items]
+plan = {{"host_pid_namespace": True, "repos": [{{"repo": os.getcwd(), "default": "origin/main", "merged_into": [], "warnings": [], "items": items}}]}}
+print(json.dumps(plan) if args.json else "\\n".join(str(item["branch"]) for item in items))
 """
 
 
 class TestHubCleanup(HubCase):
-    def ready(self, ready: bool) -> dict:
+    def ready(self, ready: bool, held: bool) -> dict:
+        """Args:
+            ready: delete_ready of the tree file.
+            held: keep feature/done in its worktree; otherwise it is a local
+                branch without one.
+        """
         root = self.write_tree()
         data = json.loads(self.tree_file.read_text(encoding="utf-8"))
         data["delete_ready"], data["delete_reason"] = ready, "" if ready else "Secret Service locked"
+        done = next(item for item in data["repos"][0]["items"] if item["branch"] == "feature/done")
+        if not held:
+            done.update(worktree=None, worktree_state=None, actions=["local-branch", "remote-branch"], compose_projects=[], images=[])
         self.tree_file.write_text(json.dumps(data), encoding="utf-8")
-        return signing.cleanup_item(f"{root}/repos/github.com/me/demo", "feature/done", f"{root}/worktrees/github.com/me/demo/feature-done", DONE_TIP)
+        return signing.cleanup_item(f"{root}/repos/github.com/me/demo", "feature/done", done["worktree"] or "", DONE_TIP)
 
     def branches(self) -> dict:
         status, _, body = self.get("/api/tree")
@@ -57,27 +88,24 @@ class TestHubCleanup(HubCase):
         return status, body
 
     def test_the_tree_offers_a_delete_only_for_merged_clean_branches_without_skip_reason(self):
-        item = self.ready(False)
+        item = self.ready(False, False)
         branches = self.branches()
         done = branches["feature/done"]
-        self.assertEqual(done["cleanup"], item)
+        self.assertEqual((done["cleanup"], done["held"]), (item, ""))
         self.assertEqual(done["plan"], [
             "feature/done @ 5b1e6fd3a7c0: merged-clean, merged by #12",
-            f"removes the worktree {item['worktree']}",
             "deletes the local branch feature/done",
-            "stays: remote-branch (local only)",
-            "stops the Compose stack demo-feature-done without -v",
-            "removes the image demo:feature-done",
+            "stays: remote-branch (local only; the host refuses a plan that deletes it)",
         ])
         self.assertEqual((done["merged_by"], done["merged_into"], done["gone"]), ("#12", "main", True))
         self.assertEqual({name: entry["cleanup"] for name, entry in branches.items() if name != "feature/done"}, {"feature": None, "feature/kept": None, "main": None})
         self.assertEqual(branches["feature/kept"]["skip"], "ignored files beyond known caches: notes.txt")
 
     def test_signing_a_delete_needs_delete_ready_and_exactly_an_offered_item(self):
-        item = self.ready(False)
+        item = self.ready(False, False)
         status, body = self.sign([item])
         self.assertEqual((status, body.decode()), (400, "delete blocked: Secret Service locked\n"))
-        item = self.ready(True)
+        item = self.ready(True, False)
         kept = {**item, "branch": "feature/kept", "worktree": "", "tip": "0c9d8e7f6a5b4c3d2e1f0a9b8c7d6e5f4a3b2c1d"}
         for name, items in (("changed tip", [{**item, "tip": OTHER_TIP}]), ("skip reason", [kept]), ("pattern", [{**item, "branch": "feature/*"}]), ("twice", [item, item]), ("extra key", [{**item, "command": "rm -rf /"}])):
             with self.subTest(name=name):
@@ -86,8 +114,37 @@ class TestHubCleanup(HubCase):
         self.assertEqual(status, 200, body)
         self.assertEqual(json.loads(body)["payload"]["items"], [signing.item_hash(item)])
 
+    def test_a_branch_held_by_a_worktree_offers_no_delete_and_names_branch_gc(self):
+        item = self.ready(True, True)
+        done = self.branches()["feature/done"]
+        held = f"held by worktree {item['worktree']}: pkgmgr branch cleanup skips it, use branch gc (kvb-git-branch-gc, pkgmgr gc)"
+        self.assertEqual((done["cleanup"], done["held"], done["plan"][1]), (None, held, held))
+        status, body = self.sign([item])
+        self.assertEqual((status, body.decode()), (400, f"feature/done in {item['repository']} is no merged-clean branch without skip reason at that tip in the tree file\n"))
+        page = (HUB / "templates" / "page-tree.js").read_text(encoding="utf-8")
+        self.assertIn('if (entry.held) {\n      return h("span", {class: "hint held", title: entry.held}, chip("bi-folder-symlink", entry.held, "neutral"));', page)
+        self.assertIn("deleteDialog(items, locked, held)", page)
+        self.assertIn("held by a worktree: not deleted here", (HUB / "templates" / "page-cleanup.js").read_text(encoding="utf-8"))
+
+    @unittest.skipUnless(INSTALLED_PKGMGR, "no pkgmgr installed")
+    def test_every_volume_text_states_the_compose_down_argv_of_the_installed_pkgmgr(self):
+        found = subprocess.run([str(Path(INSTALLED_PKGMGR).parent / "python"), "-I", "-c", "import pkgmgr.actions.branch.gc as gc; print(gc.__file__)"], capture_output=True, text=True, timeout=60, check=True)
+        calls = [node for node in ast.walk(ast.parse(Path(found.stdout.strip()).read_text(encoding="utf-8"))) if isinstance(node, ast.Call) and getattr(node.func, "id", None) == "_docker" and node.args and isinstance(node.args[0], ast.List)]
+        downs = [["docker", *(part.value if isinstance(part, ast.Constant) else "{project}" for part in call.args[0].elts)] for call in calls if any(isinstance(part, ast.Constant) and part.value == "down" for part in call.args[0].elts)]
+        self.assertEqual(len(downs), 1, downs)
+        command = " ".join(downs[0]).format(project="demo-feature-done")
+        self.ready(True, True)
+        plan = self.branches()["feature/done"]["plan"]
+        self.assertIn(f"branch gc would run {command}, which deletes the volumes of demo-feature-done", plan)
+        self.assertIn("-v", downs[0])
+        dialog = (HUB / "templates" / "page-cleanup.js").read_text(encoding="utf-8")
+        self.assertIn("never touches a worktree, Compose stack or volume", dialog)
+        self.assertIn("its docker compose down -v deletes the stack's volumes", dialog)
+        texts = [*plan, dialog, (HUB / "kvb_orchestrator" / "tree.py").read_text(encoding="utf-8")]
+        self.assertFalse([text for text in texts if "without -v" in text])
+
     def test_a_signed_delete_becomes_a_job_without_any_command(self):
-        item = self.ready(True)
+        item = self.ready(True, False)
         signer = self.pin()
         proof = signer.proof(self.prepare([item])["payload"], self.origin)
         self.assertEqual(self.post_json("/api/cleanup", {"items": [item], "proof": None}, self.guard(self.host))[0], 400)
@@ -127,7 +184,7 @@ class TestHostCleanup(unittest.TestCase):
         self.calls = self.fake.parent / "calls.jsonl"
         self.tree = base / "tree" / "branch-tree.json"
         self.tree.parent.mkdir()
-        self.tree.write_text(json.dumps({"repos": [{"path": str(self.repo), "known": True}, {"path": str(base / "stranger"), "known": False}]}), encoding="utf-8")
+        self.tree.write_text(json.dumps({"delete_ready": True, "repos": [{"path": str(self.repo), "known": True}, {"path": str(base / "stranger"), "known": False}]}), encoding="utf-8")
         self.approvers = base / "config" / "approvers.json"
         self.trust = verifier.Trust(approvers=self.approvers, replay=base / "replay" / "cleanup.json", origin=ORIGIN)
         self.signer = key(-7, "main")
@@ -154,7 +211,7 @@ class TestHostCleanup(unittest.TestCase):
     def test_a_signed_merged_clean_item_is_dry_run_then_deleted_by_exact_name(self):
         result, calls = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
         self.assertEqual([(entry["status"], entry["reason"]) for entry in result["items"]], [("done", "")])
-        selector = ["--json", "--local", "--branch", "feature/done"]
+        selector = ["--json", "--branch", "feature/done"]
         self.assertEqual(calls, [
             {"argv": ["branch", "cleanup", *selector], "cwd": str(self.repo)},
             {"argv": ["branch", "cleanup", "--apply", *selector], "cwd": str(self.repo)},
@@ -163,6 +220,22 @@ class TestHostCleanup(unittest.TestCase):
         self.assertEqual(json.loads(self.tree.read_text(encoding="utf-8")), {"rewritten": True})
         self.assertEqual(result["tree"], {"code": 0, "error": ""})
 
+    def test_an_apply_that_exits_0_but_skips_or_reports_no_result_is_failed_with_pkgmgrs_reason_and_keeps_the_tree(self):
+        outcome = self.fake.parent / "outcome.txt"
+        before = self.tree.read_text(encoding="utf-8")
+        cases = (("skipped", "skipped: changed since plan (dirty: 1 paths)", "skipped: changed since plan (dirty: 1 paths)"), ("no result", "", "pkgmgr reported no result for feature/done"))
+        for name, result, reason in cases:
+            with self.subTest(name=name):
+                outcome.write_text(result, encoding="utf-8")
+                done, calls = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
+                self.assertEqual([(entry["status"], entry["reason"], entry["code"]) for entry in done["items"]], [("failed", reason, 0)])
+                self.assertEqual([call["argv"][:3] for call in calls], [["branch", "cleanup", "--json"], ["branch", "cleanup", "--apply"]])
+                self.assertNotIn("tree", done)
+                self.assertEqual(self.tree.read_text(encoding="utf-8"), before)
+        outcome.write_text("done; image kept: demo:feature-done", encoding="utf-8")
+        done, _ = self.run_job([self.item], self.proof([self.item], self.signer), [self.entry()])
+        self.assertEqual([(entry["status"], entry["reason"]) for entry in done["items"]], [("done", "")])
+
     def test_every_doubtful_item_is_refused_and_changes_nothing(self):
         pattern = {**self.item, "branch": "feature/*"}
         option = {**self.item, "branch": "--all"}
@@ -181,13 +254,14 @@ class TestHostCleanup(unittest.TestCase):
             ("skip reason", [self.item], self.proof([self.item], self.signer), [self.entry(skip="ignored files")], "gc skip reason", 1),
             ("other worktree", [self.item], self.proof([self.item], self.signer), [self.entry(worktree="/elsewhere")], "worktree changed", 1),
             ("pattern match in pkgmgr", [self.item], self.proof([self.item], self.signer), [self.entry(), self.entry(branch="feature/done-too")], "other branches too", 1),
+            ("remote branch delete", [self.item], self.proof([self.item], self.signer), [self.entry(actions=["local-branch", "remote-branch:origin"])], "a signed delete is local only", 1),
         )
         for name, items, proof, plan, reason, dry_runs in cases:
             with self.subTest(name=name):
                 result, calls = self.run_job(items, proof, plan)
                 self.assertEqual([entry["status"] for entry in result["items"]], ["refused"])
                 self.assertIn(reason, result["items"][0]["reason"])
-                self.assertEqual([call["argv"] for call in calls], [["branch", "cleanup", "--json", "--local", "--branch", items[0]["branch"]]] * dry_runs)
+                self.assertEqual([call["argv"] for call in calls], [["branch", "cleanup", "--json", "--branch", items[0]["branch"]]] * dry_runs)
                 self.assertNotIn("tree", result)
 
     def test_a_signature_deletes_once_and_a_malformed_job_nothing(self):
@@ -217,6 +291,44 @@ class TestHostCleanup(unittest.TestCase):
         self.assertEqual((self.results / f"{job_id}.json").stat().st_mode & 0o777, 0o600)
 
 
+@unittest.skipUnless(INSTALLED_PKGMGR, "no pkgmgr installed")
+class TestInstalledPkgmgrContract(unittest.TestCase):
+    def git(self, cwd: Path, *args: str) -> str:
+        return subprocess.run(["git", *args], cwd=cwd, env={**os.environ, **GIT_ENV}, check=True, capture_output=True, text=True).stdout.strip()
+
+    def test_the_installed_pkgmgr_takes_the_runners_dry_run_argv_and_prints_a_plan_the_runner_parses_and_accepts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp)
+            origin, repo = base / "origin.git", base / "repo"
+            self.git(base, "init", "-q", "--bare", "-b", "main", str(origin))
+            self.git(base, "init", "-q", "-b", "main", str(repo))
+            self.git(repo, "remote", "add", "origin", origin.as_uri())
+            self.git(repo, "commit", "-q", "--allow-empty", "-m", "one")
+            self.git(repo, "checkout", "-q", "-b", "side")
+            self.git(repo, "commit", "-q", "--allow-empty", "-m", "side")
+            tip = self.git(repo, "rev-parse", "HEAD")
+            self.git(repo, "checkout", "-q", "main")
+            self.git(repo, "merge", "-q", "--no-ff", "-m", "merge side", "side")
+            self.git(origin, "fetch", "-q", str(repo), "main:main")
+            self.git(repo, "fetch", "-q", "origin")
+            self.git(repo, "remote", "set-head", "origin", "main")
+            code, stdout, stderr = host.run(Path(INSTALLED_PKGMGR), host.plan_argv("side", False), str(repo))
+            self.assertEqual(code, 0, stderr)
+            plan = host.plan_items(stdout)
+            self.assertIsNotNone(plan, stdout[:500])
+            self.assertEqual([(entry["branch"], entry["state"], entry["merged_by"], entry["merged_into"], entry["skip"], entry["actions"]) for entry in plan], [("side", "merged-clean", "ancestor", "origin/main", None, ["local-branch"])])
+            self.assertIsNone(host.mismatch(plan[0], {"tip": tip, "worktree": ""}))
+            self.assertTrue(signing.token_free(plan[0], "main"))
+            self.assertEqual(self.git(repo, "rev-parse", "refs/heads/side"), tip)
+            held = base / "held"
+            self.git(repo, "worktree", "add", "-q", str(held), "side")
+            code, stdout, stderr = host.run(Path(INSTALLED_PKGMGR), host.plan_argv("side", False), str(repo))
+            self.assertEqual(code, 0, stderr)
+            plan = host.plan_items(stdout)
+            self.assertEqual([(entry["worktree"], entry["actions"], bool(entry["skip"])) for entry in plan], [(os.path.realpath(held), [], True)], plan)
+            self.assertIn("gc skip reason", host.mismatch(plan[0], {"tip": tip, "worktree": os.path.realpath(held)}))
+
+
 class TestHostCleanupGitConfig(unittest.TestCase):
     def test_pkgmgr_runs_no_hook_the_repository_config_names(self):
         with tempfile.TemporaryDirectory() as tmp: