fix(source-control): read babysit trust files at the branch tip SHA

.changes/source-control/fix-babysit-trust-reads-at-sha-d99e94a3.md

@@ -0,0 +1,7 @@
+---
+bump: patch
+---
+
+### Fixed
+
+- The babysit merge gate reads the trusted-actor list and the admission rules at the tip commit of their branch, resolved by its full ref through `base_tip`, never by the branch name, so a tag named like the branch cannot supply them. The escalation lift reads the trusted list the same way, and the verdict answer key names both files by `refs/heads/<branch>`. A branch whose tip does not resolve to a commit SHA fails closed: the verdict read is unreadable and the label stays on.

plugins/source-control/skills/babysit-prs/scripts/babysit_merge.py

@@ -3859,14 +3859,18 @@ def _lifted_by_person(
         default = repository_default_branch(repo)
         if not default:
             return False, "default branch unread"
+        # At its tip SHA, so no tag of the same name stands in for it; one
+        # that does not resolve raises, which is no lift.
+        owner, _, name = repo.partition("/")
+        tip = merge_policy.base_tip(owner, name, default)
         # The newest removal decides; a tie keeps the later timeline item.
         at, event = max(enumerate(removals), key=lambda r: (r[1][0], r[0]))[1]
         actor = json_object(event.get("actor"))
         return (
             at > escalated
             and event.get("performed_via_github_app") is None
             and actor.get("type") == "User"
-            and actor.get("id") in merge_policy.trusted_actors(repo, default)[1]
+            and actor.get("id") in merge_policy.trusted_actors(repo, tip)[1]
             and merge_policy.is_human(
                 {"__typename": "User", "login": actor.get("login")},
                 repo=repo,

plugins/source-control/skills/babysit-prs/scripts/babysit_verify_evidence.py

@@ -1386,18 +1386,29 @@ def _trusted_humans(self) -> frozenset[int]:
             run_cache.note(self._humans_at)
         return self._humans or frozenset()
 
+    def _tip(self, branch: str) -> str:
+        """`branch`'s tip SHA (`merge_policy.base_tip`), so no tag of the same
+        name stands in for it; `_Unreadable` when it does not resolve."""
+        owner, _, name = self.repo.partition("/")
+        try:
+            return merge_policy.base_tip(owner, name, branch)
+        except merge_policy.PolicyReadError as exc:
+            raise _Unreadable(str(exc)) from exc
+
     def _read_trusted_list(self) -> None:
-        """Read the trusted list at the branch once: its human ids, and every
-        login's id (`_lane_app_id`), each shared by the run's gate processes."""
+        """Read the trusted list at the branch's tip once: its human ids, and
+        every login's id (`_lane_app_id`), each shared by the run's gate
+        processes."""
         key = f"{self.repo}@{self.branch}"
         at = run_cache.now()
+        tip = self._tip(self.branch)
         listing = json_object(
             _read(
                 [
                     "api",
                     "-H",
                     "Accept: application/vnd.github.raw",
-                    f"repos/{self.repo}/contents/{TRUSTED_ACTORS_PATH}?ref={self.branch}",
+                    f"repos/{self.repo}/contents/{TRUSTED_ACTORS_PATH}?ref={tip}",
                 ],
                 "trusted-actor list",
             )
@@ -1456,13 +1467,14 @@ def _task_list(self) -> str | None:
                 self._task_list_value, at = shared
             else:
                 at = run_cache.now()
+                tip = self._tip(default)
                 rules = json_object(
                     _read(
                         [
                             "api",
                             "-H",
                             "Accept: application/vnd.github.raw",
-                            f"repos/{self.repo}/contents/{ADMISSION_RULES_PATH}?ref={default}",
+                            f"repos/{self.repo}/contents/{ADMISSION_RULES_PATH}?ref={tip}",
                         ],
                         "admission rules",
                     )
@@ -2596,8 +2608,9 @@ def _answer_key(
             "r": name,
             "n": args.number,
             "h": args.head,
-            "ta": f"{args.branch}:{TRUSTED_ACTORS_PATH}",
-            "ar": f"{args.branch}:{ADMISSION_RULES_PATH}",
+            # By full ref: a bare name resolves as git does, a tag first.
+            "ta": f"refs/heads/{args.branch}:{TRUSTED_ACTORS_PATH}",
+            "ar": f"refs/heads/{args.branch}:{ADMISSION_RULES_PATH}",
         },
     )
     pull = json_object(repository.get("pullRequest"))

plugins/source-control/skills/babysit-prs/scripts/tests/test_babysit_merge_unattended.py

@@ -787,6 +787,8 @@ def setUp(self) -> None:
         self.trusted_refs: list[str] = []
         # The repository's default branch as `repository_default_branch` reads it.
         self.repo_default: str | None = "main"
+        # Each branch's tip as `base_tip` resolves it; one absent here fails.
+        self.tips: dict[str, str] = {"main": "e" * 40, "trunk": "f" * 40}
         self.files = ((LOW_FILE,), True)
         self.pr_view = fixtures._pr()
         self.queue = fixtures._queue_read(required=True)
@@ -859,6 +861,11 @@ def _dependabot_eligibility(self, *args: Any, **kwargs: Any) -> Any:
             else self.dependabot_reads[0]
         )
 
+    def _base_tip(self, owner: str, name: str, base: str) -> str:
+        if base not in self.tips:
+            raise policy.PolicyReadError(f"the base branch {base} did not resolve")
+        return self.tips[base]
+
     def _lane_evidence(self, *args: Any) -> Any:
         self.lane_args = args
         if self.lane_error is not None:
@@ -1098,6 +1105,7 @@ def gh_json(args: list[str]) -> Any:
                 "repository_default_branch",
                 side_effect=lambda repo: self.repo_default,
             ),
+            mock.patch.object(policy, "base_tip", side_effect=self._base_tip),
             mock.patch.object(merge, "read_merge_queue", return_value=self.queue),
             mock.patch.object(merge, "async_merge", self.async_merge),
             mock.patch.object(merge, "rerun_rate_limited_reviews", self.rerun),
@@ -3648,8 +3656,8 @@ def test_a_person_removing_the_label_after_the_escalation_keeps_it_off(
             out["unattended"]["writes"]["riskEscalation"],
             {"label": "lifted", "comment": "unchanged"},
         )
-        # The trusted list is read at the default branch, never the head.
-        self.assertEqual(self.trusted_refs, ["main"])
+        # The trusted list is read at the default branch's tip, never the head.
+        self.assertEqual(self.trusted_refs, [self.tips["main"]])
 
     def test_the_trusted_list_is_read_at_the_default_branch(self) -> None:
         # The pull request's base is main; a retarget cannot pick the list.
@@ -3659,11 +3667,27 @@ def test_the_trusted_list_is_read_at_the_default_branch(self) -> None:
             self.unlabeled("owner-person", "User", self.WRITTEN + timedelta(minutes=30))
         ]
         _, out = self.run_main()
-        self.assertEqual(self.trusted_refs, ["trunk"])
+        self.assertEqual(self.trusted_refs, [self.tips["trunk"]])
         self.assertEqual(
             out["unattended"]["writes"]["riskEscalation"]["label"], "lifted"
         )
 
+    def test_an_unresolved_default_branch_tip_is_no_lift(self) -> None:
+        # The list is never read by the branch's name, which a tag of the
+        # same name could answer.
+        self.escalated_earlier()
+        del self.tips["main"]
+        self.timeline = [
+            self.unlabeled("owner-person", "User", self.WRITTEN + timedelta(minutes=30))
+        ]
+        _, out = self.run_main()
+        self.assertEqual(self.log, [self.LABEL])
+        self.assertEqual(self.trusted_refs, [])
+        self.assertEqual(
+            out["unattended"]["writes"]["riskEscalation"]["liftUnread"],
+            "PolicyReadError: the base branch main did not resolve",
+        )
+
     def test_an_unread_default_branch_is_no_lift(self) -> None:
         self.escalated_earlier()
         self.repo_default = None
@@ -5497,6 +5521,7 @@ def person(self) -> contextlib.ExitStack:
         stack.enter_context(
             mock.patch.object(merge, "repository_default_branch", return_value="main")
         )
+        stack.enter_context(mock.patch.object(policy, "base_tip", return_value="e" * 40))
         return stack
 
     def test_only_the_identitys_comment_dates_the_escalation(self) -> None:

plugins/source-control/skills/babysit-prs/scripts/tests/test_babysit_split_child_trust.py

@@ -32,6 +32,12 @@
 REPO = "example-org/example-repo"
 LANES_LOGIN = "agent-lanes[bot]"
 CHILD = 9
+# Each branch's tip object, distinct per branch, so a read names the tip of
+# the branch it meant.
+TIPS = {
+    branch: {"sha": digit * 40, "type": "commit"}
+    for branch, digit in (("main", "a"), ("trunk", "b"), ("feature-x", "c"))
+}
 # The `task-list` of `.github/actions/resolve-config/admission-rules.json`,
 # which the fake serves as the branch's rules.
 TASK_LIST = json.loads(
@@ -122,6 +128,10 @@ def __init__(
         self.fail = False
         self.rules_fail = False
         self.default_fail = False
+        # Each branch's tip, as `git/ref/heads/<branch>` answers it; a branch
+        # absent here answers 404, and a value that is not a SHA is served as
+        # it stands.
+        self.tips: dict[str, Any] = {b: TIPS[b] for b in TIPS}
         self.calls: list[tuple[list[str], Any]] = []
 
     def graphql_calls(self) -> int:
@@ -134,14 +144,20 @@ def rules_calls(self) -> int:
             if any(ve.ADMISSION_RULES_PATH in a for a in args)
         )
 
-    def rules_refs(self) -> list[str]:
+    def refs(self, path: str) -> list[str]:
         return [
             a.rpartition("?ref=")[2]
             for args, _ in self.calls
             for a in args
-            if ve.ADMISSION_RULES_PATH in a
+            if path in a
         ]
 
+    def rules_refs(self) -> list[str]:
+        return self.refs(ve.ADMISSION_RULES_PATH)
+
+    def actors_refs(self) -> list[str]:
+        return self.refs(ve.TRUSTED_ACTORS_PATH)
+
     def __call__(
         self, args: list[str], **kwargs: Any
     ) -> subprocess.CompletedProcess[str]:
@@ -151,6 +167,13 @@ def __call__(
                 return subprocess.CompletedProcess(args, 1, "", "gh: boom (HTTP 502)")
             answer = {"default_branch": self.default_branch}
             return subprocess.CompletedProcess(args, 0, json.dumps(answer), "")
+        heads = f"repos/{REPO}/git/ref/heads/"
+        if len(args) == 2 and args[1].startswith(heads):
+            branch = args[1].removeprefix(heads)
+            if branch not in self.tips:
+                return subprocess.CompletedProcess(args, 1, "", "gh: Not Found (HTTP 404)")
+            answer = {"ref": f"refs/heads/{branch}", "object": self.tips[branch]}
+            return subprocess.CompletedProcess(args, 0, json.dumps(answer), "")
         if args[:2] == ["api", "graphql"]:
             assert any(a == f"query={ve.SPLIT_CHILD_QUERY}" for a in args), args
             if self.fail:
@@ -173,15 +196,19 @@ def __call__(
 
 
 def judge(
-    fake: FakeGitHub, issue: dict[str, Any], identity: str | None, **reader: Any
+    fake: FakeGitHub,
+    issue: dict[str, Any],
+    identity: str | None,
+    branch: str = "main",
+    **reader: Any,
 ) -> bool:
     with (
         mock.patch.dict("os.environ", {}, clear=False) as env,
         mock.patch.object(postmerge, "gh_capture", side_effect=fake),
         mock.patch.object(ve.merge_policy, "gh_capture", side_effect=fake),
     ):
         env.pop("BABYSIT_RUN_CACHE", None)
-        found = ve.Reader(REPO, "main", agent_identity=identity, **reader)
+        found = ve.Reader(REPO, branch, agent_identity=identity, **reader)
         return found._trusted_spec({"linked_issues": [issue]})
 
 
@@ -331,7 +358,69 @@ def test_a_feature_base_reads_the_rules_at_the_default_branch(self) -> None:
             self.assertTrue(
                 reader._trusted_spec({"linked_issues": [linked(case["child"])]})
             )
-        self.assertEqual(fake.rules_refs(), ["trunk"])
+        self.assertEqual(fake.rules_refs(), [TIPS["trunk"]["sha"]])
+        # The trusted list keeps the reader's branch, at its tip.
+        self.assertEqual(fake.actors_refs(), [TIPS["feature-x"]["sha"]])
+
+
+class TheReadsNameACommit(unittest.TestCase):
+    """Both trust reads name the branch's tip SHA, read by its full ref, so a
+    tag of the same name cannot stand in for the branch; a tip that does not
+    resolve to a commit SHA reads nothing and fails closed."""
+
+    def setUp(self) -> None:
+        data = json.loads(VECTORS.read_text(encoding="utf-8"))
+        case = next(c for c in data["cases"] if c["name"] == "the trusted child")
+        self.child = linked(case["child"])
+        self.fake = FakeGitHub(
+            data["humans"], data["app_id"], graphql_issue(case, data["parents"])
+        )
+        self.fake.default_branch = "trunk"
+
+    def judge(self) -> bool:
+        return judge(self.fake, self.child, LANES_LOGIN, branch="feature-x")
+
+    def test_each_read_names_a_sha_never_a_branch(self) -> None:
+        self.assertTrue(self.judge())
+        refs = self.fake.actors_refs() + self.fake.rules_refs()
+        self.assertEqual(
+            refs, [TIPS["feature-x"]["sha"], TIPS["trunk"]["sha"]]
+        )
+        heads = [
+            args[1]
+            for args, _ in self.fake.calls
+            if args[1].startswith(f"repos/{REPO}/git/ref/heads/")
+        ]
+        self.assertEqual(
+            heads,
+            [
+                f"repos/{REPO}/git/ref/heads/feature-x",
+                f"repos/{REPO}/git/ref/heads/trunk",
+            ],
+        )
+
+    def test_an_unreadable_tip_fails_closed(self) -> None:
+        bad = {
+            "absent (404)": None,
+            "a short sha": {"sha": "abc123", "type": "commit"},
+            "an uppercase sha": {"sha": "A" * 40, "type": "commit"},
+            "the branch name": {"sha": "trunk", "type": "commit"},
+            "no sha": {"type": "commit"},
+            "no object": "",
+        }
+        for branch in ("feature-x", "trunk"):
+            for label, tip in bad.items():
+                with self.subTest(branch=branch, tip=label):
+                    self.setUp()
+                    if tip is None:
+                        del self.fake.tips[branch]
+                    else:
+                        self.fake.tips[branch] = tip
+                    with self.assertRaises(ve._Unreadable):
+                        self.judge()
+                    # Nothing reads a trust file at the branch's name.
+                    self.assertNotIn(branch, self.fake.actors_refs())
+                    self.assertNotIn(branch, self.fake.rules_refs())
 
 
 class OneReadPerChild(unittest.TestCase):

plugins/source-control/skills/babysit-prs/scripts/tests/test_babysit_verify_answers.py

@@ -325,6 +325,17 @@ def test_each_input_of_the_read_moves_the_key(self) -> None:
             with self.subTest(row):
                 self.assertTrue(self.moved(change))
 
+    def test_the_trust_files_are_probed_by_full_branch_ref(self) -> None:
+        # A bare name resolves as git does, where a tag of the same name
+        # wins; `refs/heads/` names the branch alone.
+        world = World()
+        key(world, branch="trunk")
+        probe_args = next(
+            c for c in world.calls if f"query={ve.PROBE_QUERY}" in c
+        )
+        self.assertIn(f"ta=refs/heads/trunk:{ve.TRUSTED_ACTORS_PATH}", probe_args)
+        self.assertIn(f"ar=refs/heads/trunk:{ve.ADMISSION_RULES_PATH}", probe_args)
+
     def test_the_config_rung_and_base_move_the_key(self) -> None:
         same = World()
         self.assertNotEqual(

plugins/source-control/skills/babysit-prs/scripts/tests/test_babysit_verify_evidence.py

@@ -437,8 +437,11 @@ def rest(
             if sha not in self.diffs:
                 return missing()
             return subprocess.CompletedProcess(args, 0, self.diffs[sha], "")
+        if path == f"repos/{REPO}/git/ref/heads/main":
+            return ok({"ref": "refs/heads/main", "object": {"sha": MAIN_SHA}})
         if path.startswith(f"repos/{REPO}/contents/"):
-            return ok(self.actors)
+            # Read at main's tip only, never by the branch's name.
+            return ok(self.actors) if path.endswith(f"?ref={MAIN_SHA}") else missing()
         issues = f"repos/{REPO}/issues/"
         if path.startswith(issues) and path.endswith("/timeline?per_page=100"):
             n = int(path[len(issues) :].split("/")[0])
@@ -486,6 +489,7 @@ def read(fake: FakeGitHub, pr: dict[str, Any] | None = None) -> ve.Evidence:
     # this module's setUpModule.
     with (
         mock.patch.object(postmerge, "gh_capture", side_effect=fake),
+        mock.patch.object(ve.merge_policy, "gh_capture", side_effect=fake),
         mock.patch.object(ve, "gh_api_with_headers", side_effect=_page_with_date),
     ):
         return ve.Reader(REPO, "main", now=NOW).evidence(PR, HEAD, pr or pull())
@@ -2724,13 +2728,25 @@ def _page_with_date(args: list[str], **kwargs: Any) -> Any:
 _role_label = mock.patch.object(postmerge, "_role_label", return_value="needs-human")
 
 
+# The branch tip `merge_policy.base_tip` reads goes to whatever fake
+# `postmerge.gh_capture` is patched to; a test that answers it otherwise
+# patches it again.
+_tip_reads = mock.patch.object(
+    ve.merge_policy,
+    "gh_capture",
+    side_effect=lambda args, **kwargs: postmerge.gh_capture(args, **kwargs),
+)
+
+
 def setUpModule() -> None:
     _no_sleep.start()
     _no_pages.start()
     _role_label.start()
+    _tip_reads.start()
 
 
 def tearDownModule() -> None:
+    _tip_reads.stop()
     _role_label.stop()
     _no_pages.stop()
     _no_sleep.stop()