perf(source-control): batch the demotion record reads

.changes/source-control/perf-batch-demotion-reads-7e3a91c4.md

@@ -0,0 +1,7 @@
+---
+bump: patch
+---
+
+### Changed
+
+- The demotion record read (`read_demotion`), behind the post-merge check's floor, the merge gate and the lane room, reads its records 20 issues to a GraphQL query instead of one each, and reads on its own any record a batch fails or leaves out, so its answer is unchanged. On this repository's 39 candidate issues it took 5 calls instead of 42.

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

@@ -1413,14 +1413,18 @@ def is_human(
     "repository(owner:$o,name:$r){issues(first:100,after:$c,filterBy:{createdBy:$a}){"
     "pageInfo{hasNextPage endCursor} nodes{number body lastEditedAt}}}}"
 )
-DEMOTION_RECORD = (
-    "query PolicyDemotionRecord($o:String!,$r:String!,$n:Int!){"
-    "repository(owner:$o,name:$r){issue(number:$n){number state body createdAt "
-    "author{__typename login} "
+DEMOTION_FIELDS = (
+    "number state body createdAt author{__typename login} "
     "userContentEdits(first:100){totalCount nodes{editedAt deletedAt editor{__typename login} diff}} "
     "timelineItems(itemTypes:[CLOSED_EVENT],last:1){nodes{... on ClosedEvent{"
-    "createdAt actor{__typename login} closer{__typename}}}}}}}"
+    "createdAt actor{__typename login} closer{__typename}}}}"
+)
+DEMOTION_RECORD = (
+    "query PolicyDemotionRecord($o:String!,$r:String!,$n:Int!){"
+    f"repository(owner:$o,name:$r){{issue(number:$n){{{DEMOTION_FIELDS}}}}}}}"
 )
+# `read_demotion`: records per query, as `babysit_postmerge.read_pins` batches pins.
+DEMOTION_BATCH = 20
 
 
 @dataclass(frozen=True)
@@ -1464,6 +1468,28 @@ def _candidate_issues(owner: str, name: str, filer: str) -> list[int]:
     raise PolicyReadError(f"issues by {filer}: more than {ISSUES_PAGE_LIMIT} pages")
 
 
+def _demotion_records(owner: str, name: str, numbers: list[int]) -> dict[int, Any]:
+    """`DEMOTION_RECORD`'s issue for each of `numbers` that one aliased query
+    returned; an issue it left out or answered with no number is absent, and
+    a failed query answers none, so the caller reads those one at a time."""
+    aliases = " ".join(f"i{n}:issue(number:{n}){{...d}}" for n in numbers)
+    query = (
+        "query PolicyDemotionRecords($o:String!,$r:String!){"
+        f"repository(owner:$o,name:$r){{{aliases}}}}} "
+        f"fragment d on Issue{{{DEMOTION_FIELDS}}}"
+    )
+    try:
+        data = _graphql(query, f"issues #{numbers[0]}+", o=owner, r=name)
+    except PolicyReadError:
+        return {}
+    repository = json_object(data.get("repository"))
+    return {
+        n: issue
+        for n in numbers
+        if is_json_object(issue := repository.get(f"i{n}")) and issue.get("number") == n
+    }
+
+
 def _record_hold(issue: dict[str, Any], human: Callable[[Any], bool]) -> str | None:
     """Why a demotion record still holds the rung off, or None once a human
     closed it directly and no non-human ever edited it."""
@@ -1529,7 +1555,9 @@ def read_demotion(
     body carries the demotion marker now or in any revision, so stripping the
     marker or the label hides none. A record holds the rung off unless its
     latest `ClosedEvent` has no `closer` and a human actor, and no non-human
-    ever edited it. Any read failure holds it off.
+    ever edited it. Any read failure holds it off. Records are read
+    `DEMOTION_BATCH` to a query (`_demotion_records`), and one the batch did
+    not return is read on its own, so the answer is the per-issue read's.
     """
     owner, name = _split_repo(repo)
     filers = [DEMOTION_FILER, *([agent_identity.strip()] if agent_identity else [])]
@@ -1548,11 +1576,19 @@ def read_demotion(
         candidates = sorted(
             {n for f in filers for n in _candidate_issues(owner, name, f)}
         )
-        for number in candidates:
-            data = _graphql(
-                DEMOTION_RECORD, f"issue #{number}", o=owner, r=name, n=number
-            )
-            issue = json_object(json_object(data.get("repository")).get("issue"))
+        batch: dict[int, Any] = {}
+        for i, number in enumerate(candidates):
+            if i % DEMOTION_BATCH == 0:
+                batch = _demotion_records(
+                    owner, name, candidates[i : i + DEMOTION_BATCH]
+                )
+            if number in batch:
+                issue = batch[number]
+            else:
+                data = _graphql(
+                    DEMOTION_RECORD, f"issue #{number}", o=owner, r=name, n=number
+                )
+                issue = json_object(json_object(data.get("repository")).get("issue"))
             if not issue:
                 raise PolicyReadError(f"issue #{number} returned nothing")
             revisions = [

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

@@ -18,6 +18,7 @@
 import json
 import os
 import pathlib
+import re
 import shutil
 import subprocess
 import sys
@@ -725,6 +726,8 @@ def by_author(variables: dict[str, str]) -> dict[str, Any]:
         return {"repository": {"issues": page}}
 
     gh.graphql["PolicyIssuesByAuthor"] = by_author
+    # Before the single read's name, which the batch query's name extends.
+    gh.graphql["PolicyDemotionRecords"] = {"repository": {"i9": record}}
     gh.graphql["PolicyDemotionRecord"] = {"repository": {"issue": record}}
     return gh
 
@@ -1000,6 +1003,138 @@ def test_a_timestamp_without_an_offset_is_off(self) -> None:
         self.assertTrue(result.reason.startswith("demotion-unreadable"), result.reason)
 
 
+class RecordsFake:
+    """`read_demotion`'s reads over many records: the identity's listing names
+    each of `records`; the batch query answers each alias it asks for, less
+    `dropped`, or fails when it asks for any of `failing`; a single read
+    answers its record, or nothing for a number with none."""
+
+    ALIAS = re.compile(r"i(\d+):issue\(number:(\d+)\)")
+
+    def __init__(self, records: dict[int, dict[str, Any]]) -> None:
+        self.records = records
+        self.dropped: set[int] = set()
+        self.failing: set[int] = set()
+        self.queries: list[str] = []
+
+    def __call__(self, args: list[str], **_: Any) -> subprocess.CompletedProcess[str]:
+        query = next(a for a in args if a.startswith("query="))
+        variables = dict(a.split("=", 1) for a in args if "=" in a and a[0] != "q")
+        self.queries.append(query.split("(", 1)[0].removeprefix("query=query "))
+        if "PolicyIssuesByAuthor" in query:
+            nodes = [
+                {"number": n, "body": DEMOTION_BODY, "lastEditedAt": None}
+                for n in self.records
+                if variables["a"] == IDENTITY
+            ]
+            page = {"pageInfo": {"hasNextPage": False, "endCursor": None}}
+            return ok({"data": {"repository": {"issues": page | {"nodes": nodes}}}})
+        if "PolicyDemotionRecords" in query:
+            asked = [int(n) for _, n in self.ALIAS.findall(query)]
+            if self.failing & set(asked):
+                return ok({"errors": [{"type": "NOT_FOUND"}], "data": None})
+            repository = {
+                f"i{n}": self.records.get(n) for n in asked if n not in self.dropped
+            }
+            return ok({"data": {"repository": repository}})
+        number = int(variables["n"])
+        return ok({"data": {"repository": {"issue": self.records.get(number)}}})
+
+
+def demotion_record(number: int, **issue: Any) -> dict[str, Any]:
+    return {
+        "number": number,
+        "state": "CLOSED",
+        "body": DEMOTION_BODY,
+        "createdAt": f"2026-10-12T10:{number:02d}:00Z",
+        "author": bot("agent-lanes"),
+        "userContentEdits": {"totalCount": 0, "nodes": []},
+        "timelineItems": {
+            "nodes": [
+                {
+                    "createdAt": f"2026-10-12T12:{number:02d}:00Z",
+                    "actor": user(OPERATOR),
+                    "closer": None,
+                }
+            ]
+        },
+    } | issue
+
+
+class ReadDemotionBatched(unittest.TestCase):
+    """`read_demotion` reads records `DEMOTION_BATCH` to a query and answers
+    exactly as reading each one on its own does."""
+
+    def read(self, fake: RecordsFake, *, per_issue: bool = False) -> Any:
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(mock.patch.object(policy, "gh_capture", fake))
+            if per_issue:
+                # The reference: every record read on its own.
+                stack.enter_context(
+                    mock.patch.object(policy, "_demotion_records", lambda *a: {})
+                )
+            return read_demotion()
+
+    def test_45_records_take_two_listings_and_three_batches(self) -> None:
+        fake = RecordsFake({n: demotion_record(n) for n in range(1, 46)})
+        result = self.read(fake)
+        self.assertFalse(result.active, result.reason)
+        self.assertEqual(result.records, tuple(range(1, 46)))
+        self.assertEqual(
+            fake.queries,
+            ["PolicyIssuesByAuthor"] * 2 + ["PolicyDemotionRecords"] * 3,
+        )
+
+    def test_batched_reads_equal_per_issue_reads(self) -> None:
+        open_hold = {"state": "OPEN", "timelineItems": {"nodes": []}}
+        edit = {"editedAt": "2026-10-12T11:00:00Z", "deletedAt": None}
+        edit |= {"editor": bot("agent-lanes"), "diff": "scrubbed"}
+        truncated = {
+            "body": "scrubbed",
+            "userContentEdits": {"totalCount": 150, "nodes": [edit] * 100},
+        }
+        cases: dict[str, tuple[dict[int, dict[str, Any]], set[int], set[int]]] = {
+            "all clear": ({}, set(), set()),
+            "a hold before an unreadable record": ({5: open_hold}, set(), set()),
+            "a truncated edit history": ({12: truncated}, set(), set()),
+            "a batch error falls back": ({3: open_hold}, {7}, set()),
+            "an issue missing from the batch answer": ({}, set(), {24}),
+            "a missing issue with no single read either": ({}, set(), {30}),
+        }
+        for name, (overrides, failing, dropped) in cases.items():
+            with self.subTest(name):
+                records = {n: demotion_record(n) for n in range(1, 31)}
+                for n, issue in overrides.items():
+                    records[n] |= issue
+                if name.startswith("a missing issue"):
+                    records[30] = None  # type: ignore[assignment]
+                if name.startswith("a hold before"):
+                    records[22] = None  # type: ignore[assignment]
+                batched_fake = RecordsFake(records)
+                batched_fake.failing, batched_fake.dropped = failing, dropped
+                batched = self.read(batched_fake)
+                expected = self.read(RecordsFake(records), per_issue=True)
+                self.assertEqual(batched, expected)
+                singles = batched_fake.queries.count("PolicyDemotionRecord")
+                if failing:
+                    # The failed chunk's records, each read on its own up to the hold.
+                    self.assertEqual(singles, 3)
+                if dropped:
+                    self.assertEqual(singles, len(dropped))
+
+    def test_the_cases_reach_their_answers(self) -> None:
+        # The equality cases above are not all one answer.
+        records = {n: demotion_record(n) for n in range(1, 31)}
+        records[12] |= {"userContentEdits": {"totalCount": 101, "nodes": []}}
+        self.assertEqual(
+            self.read(RecordsFake(records)).reason, "#12 edit-history-incomplete"
+        )
+        records = {n: demotion_record(n) for n in range(1, 31)}
+        records[30] = None  # type: ignore[assignment]
+        reason = self.read(RecordsFake(records)).reason
+        self.assertEqual(reason, "demotion-unreadable: issue #30 returned nothing")
+
+
 class ResolveIdentity(unittest.TestCase):
     def test_equal_logins_resolve(self) -> None:
         result = policy.resolve_identity(IDENTITY, "Agent-Lanes[bot]")