Skip to content

GitHub and provisioning

GitHubClient exposes authenticated REST/GraphQL, the PyGitHub client (gh), the bound repository, and higher-level helpers. Many mutation helpers return (success, error) and must be checked by callers. Raw request() returns a requests.Response and raises on HTTP errors.

request enforces the authenticated API host, rejects authorization overrides, defaults to a 60-second timeout and disabled redirects, and journals metadata when a recorder is attached. Mutations are not automatically retried. graphql raises on GraphQL errors; call the raw endpoint to inspect partial responses.

Framework ownership is recorded immediately after create/fork. delete_owned_repo checks immutable ownership; delete_repo is the general explicit deletion operation used by bulk cleanup. get_pr_details and get_issue_details return structured artifacts for evidence filtering.

GitHubClient

A wrapper for PyGitHub to interact with repositories.

Source code in src/benchmark/utils/gh_client.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
class GitHubClient:
    """A wrapper for PyGitHub to interact with repositories."""

    def __init__(
        self,
        repo: str = "owner/repo",
        token: Optional[str] = None,
        token_env_var: str = "GITHUB_TOKEN",
        auth_label: str = "GitHub",
        record_event=None,
        actor: str = "owner",
    ):
        self.repo_name = repo
        self.token_env_var = token_env_var
        self.auth_label = auth_label
        self.token = token or self._get_token()
        self.gh = Github(self.token)
        self._repo_cache: Optional[Repository.Repository] = None
        self._owned_repo = None
        self._authenticated_login = None
        self.record_event = record_event
        self.actor = actor

    def _record(self, kind, **data):
        if self.record_event is not None:
            self.record_event(kind, actor=self.actor, **data)

    def request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        """Call any GitHub REST endpoint, preserving the response and HTTP errors."""
        base = self.gh.base_url.rstrip("/")
        url = endpoint if endpoint.startswith("https://") else base + "/" + endpoint.lstrip("/")
        parsed, origin = urlsplit(url), urlsplit(base)
        if parsed.scheme != "https" or parsed.netloc != origin.netloc or parsed.username or parsed.password:
            raise ValueError("Endpoint must belong to the authenticated GitHub API host")
        headers = requests.structures.CaseInsensitiveDict(kwargs.pop("headers", {}))
        if "Authorization" in headers or "auth" in kwargs:
            raise ValueError("Select a GitHub actor instead of overriding its credentials")
        headers["Authorization"] = f"Bearer {self.token}"
        headers.setdefault("Accept", "application/vnd.github+json")
        kwargs.setdefault("timeout", 60)
        kwargs.setdefault("allow_redirects", False)
        request_id = uuid.uuid4().hex
        self._record("api_request", request_id=request_id, method=method.upper(), path=parsed.path)
        try:
            response = requests.request(method, url, headers=headers, **kwargs)
        except requests.RequestException:
            self._record("api_response", request_id=request_id, status=None)
            raise
        self._record(
            "api_response",
            request_id=request_id,
            status=response.status_code,
            github_request_id=response.headers.get("X-GitHub-Request-Id"),
        )
        response.raise_for_status()
        return response

    def graphql(self, query: str, variables: dict | None = None) -> dict:
        result = self.request("POST", "/graphql", json={"query": query, "variables": variables or {}}).json()
        if result.get("errors"):
            raise RuntimeError(f"GitHub GraphQL errors: {result['errors']}")
        return result["data"]

    def _claim_repository(self, repo):
        self._owned_repo = (repo.full_name, repo.id)
        self._record("resource", name=repo.full_name, id=repo.id, state="created")

    def _get_token(self) -> str:
        """Retrieves GitHub token from environment or gh CLI."""
        token = os.environ.get(self.token_env_var)
        if token:
            return token

        if self.token_env_var == "GITHUB_TOKEN":
            try:
                result = subprocess.run(["gh", "auth", "token"], capture_output=True, text=True, check=True)
                return result.stdout.strip()
            except (subprocess.CalledProcessError, FileNotFoundError):
                pass

        click.echo(
            click.style(
                f"Error: {self.auth_label} token not found. Please set {self.token_env_var}"
                + (" or run 'gh auth login'." if self.token_env_var == "GITHUB_TOKEN" else "."),
                fg="red",
            ),
            err=True,
        )
        raise RuntimeError("Missing GitHub authentication")

    def get_authenticated_user_login(self) -> str:
        """Returns the login of the authenticated user for this client."""
        if self._authenticated_login is None:
            rate_limiter.wait()
            self._authenticated_login = self.gh.get_user().login
        return self._authenticated_login

    @property
    def repository(self) -> Repository.Repository:
        """Lazily loads and returns the Repository object."""
        if self._repo_cache is None:
            rate_limiter.wait()
            self._repo_cache = self.gh.get_repo(self.repo_name)
        return self._repo_cache

    def get_repo_info(self) -> Optional[Dict[str, Any]]:
        """Fetches repository information."""
        try:
            repo = self.repository
            return {
                "name": repo.name,
                "owner": {"login": repo.owner.login},
                "defaultBranchRef": {"name": repo.default_branch},
                "isEmpty": repo.size == 0,
                "size": repo.size,
                "default_branch": repo.default_branch,
            }
        except GithubException as e:
            if e.status == 404:
                return None
            raise

    def get_default_branch(self) -> str:
        """Returns the name of the default branch."""
        try:
            return self.repository.default_branch
        except GithubException:
            return "main"

    def create_repo(self, public: bool = True) -> Tuple[bool, str]:
        """Creates the repository if it doesn't exist."""
        if self._owned_repo is not None:
            return False, "Client already owns a repository awaiting cleanup"
        try:
            name = self.repo_name.split("/")[-1]
            if "/" in self.repo_name:
                owner = self.repo_name.split("/", 1)[0]
                user = self.gh.get_user()
                if user.login.lower() == owner.lower():
                    repo = user.create_repo(name, private=not public)
                else:
                    org = self.gh.get_organization(owner)
                    repo = org.create_repo(name, private=not public)
            else:
                repo = self.gh.get_user().create_repo(name, private=not public)

            self.repo_name = repo.full_name
            self._repo_cache = repo
            self._claim_repository(repo)
            return True, ""
        except GithubException as e:
            return False, str(e)

    def fork_repo(self, template_repo_name: str) -> Tuple[bool, str]:
        """Fork the specified source, refusing collisions and existing owner forks."""
        if self._owned_repo is not None:
            return False, "Client already owns a repository awaiting cleanup"
        try:
            user_login = self.get_authenticated_user_login()
            owner, name = self.repo_name.split("/", 1) if "/" in self.repo_name else (user_login, self.repo_name)
            target = f"{owner}/{name}"
            try:
                self.gh.get_repo(target)
            except GithubException as exc:
                if exc.status != 404:
                    raise
            else:
                return False, f"Repository already exists: {target}"
            template = self.gh.get_repo(template_repo_name)
            for existing in template.get_forks():
                if existing.owner.login.lower() == owner.lower():
                    return False, f"Owner already has a fork: {existing.full_name}; use a separate owner or repository copy"
            kwargs = {"name": name, "default_branch_only": True}
            if owner.lower() != user_login.lower():
                kwargs["organization"] = owner
            new_repo = template.create_fork(**kwargs)
            if new_repo.full_name.lower() != target.lower():
                return False, f"GitHub returned an unexpected fork: {new_repo.full_name}"
            self.repo_name = new_repo.full_name
            self._repo_cache = new_repo
            self._claim_repository(new_repo)
            return True, ""
        except Exception as exc:
            return False, str(exc)

    @retry(
        retry=retry_if_exception_type(GithubException),
        stop=stop_after_attempt(15),
        wait=wait_exponential(multiplier=1, min=2, max=10),
    )
    def wait_until_ready(self):
        repo = self.gh.get_repo(self.repo_name)
        repo.get_branch(repo.default_branch)
        self._repo_cache = repo

    def _delete_repository(self, name: str, repository_id: int) -> None:
        """Delete a recorded repository identity; an absent repository is already clean."""
        try:
            repo = self.gh.get_repo(name)
            if repo.id != repository_id:
                raise RuntimeError(f"Refusing cleanup: repository ID changed for {name}")
            repo.delete()
        except GithubException as exc:
            if exc.status != 404:
                raise

    def delete_owned_repo(self) -> Tuple[bool, str]:
        """Delete only the exact repository created through this client."""
        if self._owned_repo is None:
            return True, ""
        name, repo_id = self._owned_repo
        try:
            self._delete_repository(name, repo_id)
        except (GithubException, RuntimeError) as exc:
            return False, str(exc)
        self._record("resource", name=name, id=repo_id, state="deleted")
        self._owned_repo = None
        self._repo_cache = None
        return True, ""

    @retry(
        retry=retry_if_exception_type(GithubException),
        stop=stop_after_attempt(10),
        wait=wait_exponential(multiplier=2, min=4, max=30),
    )
    def delete_repo(self) -> Tuple[bool, str]:
        """Deletes the current repository."""
        try:
            repo = self.repository
            repo.delete()
            if self._owned_repo == (self.repo_name, repo.id):
                self._owned_repo = None
            self._repo_cache = None
            return True, ""
        except GithubException as e:
            if e.status == 404:
                return True, ""  # Already gone
            if e.status == 403:
                if "delete_repo" in str(e):
                    msg = (
                        "\nERROR: Missing 'delete_repo' scope. Please run:\n  gh auth refresh -h github.com -s delete_repo\n"
                    )
                    click.echo(click.style(msg, fg="yellow", bold=True))
                elif "done being created on disk" in str(e):
                    # Re-raise to let the retry decorator handle it
                    raise e
            return False, str(e)

    def get_branch_info(self, branch_name: str) -> Optional[Dict[str, Any]]:
        """Checks if a branch exists and returns its info."""
        try:
            branch = self.repository.get_branch(branch_name)
            return {"name": branch.name, "commit": {"sha": branch.commit.sha}}
        except GithubException as e:
            if e.status == 404:
                return None
            raise

    def create_branch(self, new_branch: str, source_branch: Optional[str] = None) -> Tuple[bool, str]:
        """Creates a new branch from a source branch."""
        if not source_branch:
            source_branch = self.get_default_branch()

        try:
            sb = self.repository.get_branch(source_branch)
            self.repository.create_git_ref(ref=f"refs/heads/{new_branch}", sha=sb.commit.sha)
            return True, ""
        except GithubException as e:
            return False, str(e)

    def get_file_sha(self, path: str, branch: Optional[str] = None) -> Optional[str]:
        """Gets the SHA of a file on a specific branch."""
        try:
            kwargs = {}
            if branch:
                kwargs["ref"] = branch
            content = self.repository.get_contents(path, **kwargs)
            if isinstance(content, list):
                return None
            return content.sha
        except GithubException as e:
            if e.status == 404:
                return None
            raise

    @retry(
        retry=retry_if_exception_type(GithubException),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        stop=stop_after_attempt(5),
        before_sleep=before_sleep_log(logger, logging.INFO),
    )
    def put_file(self, path: str, content: str, message: str, branch: Optional[str] = None) -> Tuple[bool, str]:
        """Uploads or updates a file using the GitHub API."""
        if not branch:
            branch = self.get_default_branch()

        try:
            sha = self.get_file_sha(path, branch)
            if sha:
                self.repository.update_file(path, message, content, sha, branch=branch)
            else:
                self.repository.create_file(path, message, content, branch=branch)
            return True, ""
        except GithubException as e:
            if e.status == 409:
                raise e
            return False, str(e)

    @retry(
        retry=retry_if_exception_type(GithubException),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        stop=stop_after_attempt(5),
    )
    def delete_file(self, path: str, message: str, branch: Optional[str] = None) -> Tuple[bool, str]:
        """Deletes a file from the repository."""
        if not branch:
            branch = self.get_default_branch()

        try:
            sha = self.get_file_sha(path, branch)
            if sha:
                self.repository.delete_file(path, message, sha, branch=branch)
                return True, ""
            return True, "File not found"  # Already gone
        except GithubException as e:
            return False, str(e)

    @staticmethod
    def _artifact(item, kind):
        created = item.submitted_at if kind == "review" else item.created_at
        return {
            "id": item.id,
            "kind": kind,
            "body": item.body or "",
            "author": item.user.login if item.user else "",
            "author_type": item.user.type if item.user else "",
            "created_at": created.isoformat() if created else None,
            "state": item.state if kind == "review" else None,
        }

    def get_pr_details(self, pr_number: int) -> Dict[str, Any]:
        """Fetch complete, attributed comments and formal reviews; propagate read errors."""
        pr = self.repository.get_pull(pr_number)
        comments = [self._artifact(c, "issue_comment") for c in pr.get_issue_comments()]
        reviews = [self._artifact(r, "review") for r in pr.get_reviews()]
        comments.extend(reviews)
        comments.extend(self._artifact(c, "review_comment") for c in pr.get_review_comments())
        return {
            "title": pr.title,
            "body": pr.body,
            "state": pr.state,
            "comments": [c["body"] for c in comments],
            "comment_details": comments,
            "reviews": reviews,
        }

    def get_issue_details(self, issue_number: int) -> Dict[str, Any]:
        issue = self.repository.get_issue(issue_number)
        comments = [self._artifact(c, "issue_comment") for c in issue.get_comments()]
        return {
            "title": issue.title,
            "body": issue.body,
            "state": issue.state,
            "comments": [c["body"] for c in comments],
            "comment_details": comments,
        }

    @retry(
        retry=retry_if_exception_type(GithubException),
        wait=wait_exponential(multiplier=1, min=1, max=5),
        stop=stop_after_attempt(5),
    )
    def list_files(self, branch: Optional[str] = None) -> List[str]:
        """Lists files in a specific branch."""
        try:
            ref = branch or self.get_default_branch()
            tree = self.repository.get_git_tree(ref, recursive=True)
            return [item.path for item in tree.tree if item.type == "blob"]
        except GithubException as e:
            if e.status == 409:
                return []
            raise

    def set_secret(self, name: str, value: str) -> Tuple[bool, str]:
        """Sets a repository secret."""
        try:
            self.repository.create_secret(name, value)
            return True, ""
        except GithubException as e:
            return False, str(e)

    def set_variable(self, name: str, value: str) -> Tuple[bool, str]:
        """Sets a repository variable."""
        try:
            self.repository.create_variable(name, value)
            return True, ""
        except GithubException as e:
            return False, str(e)

    def enable_actions(self) -> Tuple[bool, str]:
        """Enables GitHub Actions for the repository."""
        try:
            # Using gh CLI for simplicity as PyGitHub doesn't have a direct method for this
            stdout, stderr = self.run_gh(
                ["api", "-X", "PUT", f"repos/{self.repo_name}/actions/permissions", "-F", "enabled=true"]
            )
            if stderr and "error" in stderr.lower():
                return False, stderr
            return True, ""
        except Exception as e:
            return False, str(e)

    def set_fork_pr_approval_policy(self, policy: str = "first_time_contributors_new_to_github") -> Tuple[bool, str]:
        """Sets the fork PR workflow approval policy for the repository."""
        try:
            stdout, stderr = self.run_gh(
                [
                    "api",
                    "-X",
                    "PUT",
                    f"/repos/{self.repo_name}/actions/permissions/fork-pr-contributor-approval",
                    "-H",
                    "X-GitHub-Api-Version: 2026-03-10",
                    "-f",
                    f"approval_policy={policy}",
                ]
            )
            if stderr and "error" in stderr.lower():
                return False, stderr
            return True, ""
        except Exception as e:
            return False, str(e)

    def enable_issues(self) -> Tuple[bool, str]:
        """Enables GitHub Issues for the repository."""
        try:
            # Using gh CLI for simplicity as PyGitHub doesn't have a direct method for this
            stdout, stderr = self.run_gh(["repo", "edit", self.repo_name, "--enable-issues"])
            if stderr:
                return False, stderr
            return True, ""
        except Exception as e:
            return False, str(e)

    def list_repos(self, limit: int = 100) -> List[Dict[str, str]]:
        """Lists repositories for the authenticated user."""
        try:
            repos = self.gh.get_user().get_repos()
            return [{"name": r.name, "nameWithOwner": r.full_name} for r in repos[:limit]]
        except GithubException:
            return []

    def get_workflow_runs(self, workflow_id: str = None) -> List[Any]:
        """Fetches recent runs of a specific workflow or all workflows."""
        try:
            if workflow_id:
                runs = self.repository.get_workflow(workflow_id).get_runs()
            else:
                runs = self.repository.get_workflow_runs()
            return [r for r in runs[:10]]
        except GithubException:
            return []

    def batch_sync(
        self,
        additions: Dict[str, str],
        deletions: List[str],
        message: str,
        branch: Optional[str] = None,
    ) -> Tuple[bool, str]:
        """Performs multiple additions and deletions in a single commit."""
        if not branch:
            branch = self.get_default_branch()

        try:
            repo = self.repository

            @retry(
                retry=retry_if_exception_type(GithubException),
                stop=stop_after_attempt(10),
                wait=wait_exponential(multiplier=1, min=2, max=10),
            )
            def get_ref():
                rate_limiter.wait()
                return repo.get_git_ref(f"heads/{branch}")

            ref = get_ref()
            old_commit = repo.get_git_commit(ref.object.sha)
            base_tree_sha = old_commit.tree.sha

            # We avoid recursive=True because it fails with 502 on large repos like Sentry.
            # Instead, we will build the new tree by navigating only what we need.

            def get_tree_without_path(current_tree_sha, path_parts):
                """Recursively navigates trees to 'delete' a path by omitting it from a new tree."""
                tree = repo.get_git_tree(current_tree_sha, recursive=False)
                elements = []

                target = path_parts[0]
                remaining = path_parts[1:]

                found_target = False
                for item in tree.tree:
                    if item.path == target:
                        found_target = True
                        if remaining:
                            # We need to go deeper into this subtree
                            if item.type == "tree":
                                new_subtree_sha = get_tree_without_path(item.sha, remaining)
                                if new_subtree_sha:
                                    elements.append(
                                        InputGitTreeElement(
                                            path=item.path, mode=item.mode, type=item.type, sha=new_subtree_sha
                                        )
                                    )
                                # if new_subtree_sha is None, it means the whole subtree was deleted
                            else:
                                # Target is a file, but we have remaining path parts?
                                # This means the path doesn't match the structure. Keep it as is.
                                elements.append(
                                    InputGitTreeElement(path=item.path, mode=item.mode, type=item.type, sha=item.sha)
                                )

                        else:
                            # This is the item to delete! Just don't add it to elements.
                            pass
                    else:
                        # Not our target, keep it
                        elements.append(InputGitTreeElement(path=item.path, mode=item.mode, type=item.type, sha=item.sha))

                if not found_target:
                    # Target not found in this tree, nothing to delete here
                    return current_tree_sha

                if not elements:
                    # Entire tree is empty now
                    return None

                new_tree = repo.create_git_tree(elements)
                return new_tree.sha

            current_tree_sha = base_tree_sha
            for deletion_path in deletions:
                # Normalize path: remove leading/trailing slashes
                clean_path = deletion_path.strip("/")
                if not clean_path:
                    continue

                path_parts = clean_path.split("/")
                current_tree_sha = get_tree_without_path(current_tree_sha, path_parts)
                if not current_tree_sha:
                    # We deleted everything? (unlikely but possible)
                    # Create an empty tree to avoid errors
                    empty_tree = repo.create_git_tree([])
                    current_tree_sha = empty_tree.sha

            # 2. Handle additions using the base_tree merge capability
            elements = []
            for path, content in additions.items():
                elements.append(InputGitTreeElement(path=path, mode="100644", type="blob", content=content))

            # Create the final tree by merging additions into our modified tree
            final_tree = repo.create_git_tree(elements, base_tree=repo.get_git_tree(current_tree_sha))

            new_commit = repo.create_git_commit(message, final_tree, [old_commit])
            ref.edit(new_commit.sha)

            return True, ""
        except Exception as e:
            return False, str(e)

    def run_gh(self, args, **kwargs):
        """Legacy compatibility method. SHOULD BE REMOVED eventually."""
        cmd = ["gh"] + args
        if kwargs.get("use_repo"):
            cmd += ["-R", self.repo_name]

        env = os.environ.copy()
        env["GH_TOKEN"] = self.token
        result = subprocess.run(cmd, capture_output=True, text=True, env=env)
        if result.returncode != 0:
            raise RuntimeError(f"gh exited with status {result.returncode}: {result.stderr.strip()}")
        return result.stdout, result.stderr

repository property 

repository: Repository

Lazily loads and returns the Repository object.

__init__ 

__init__(repo: str = 'owner/repo', token: Optional[str] = None, token_env_var: str = 'GITHUB_TOKEN', auth_label: str = 'GitHub', record_event=None, actor: str = 'owner')
Source code in src/benchmark/utils/gh_client.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    repo: str = "owner/repo",
    token: Optional[str] = None,
    token_env_var: str = "GITHUB_TOKEN",
    auth_label: str = "GitHub",
    record_event=None,
    actor: str = "owner",
):
    self.repo_name = repo
    self.token_env_var = token_env_var
    self.auth_label = auth_label
    self.token = token or self._get_token()
    self.gh = Github(self.token)
    self._repo_cache: Optional[Repository.Repository] = None
    self._owned_repo = None
    self._authenticated_login = None
    self.record_event = record_event
    self.actor = actor

request 

request(method: str, endpoint: str, **kwargs) -> requests.Response

Call any GitHub REST endpoint, preserving the response and HTTP errors.

Source code in src/benchmark/utils/gh_client.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
    """Call any GitHub REST endpoint, preserving the response and HTTP errors."""
    base = self.gh.base_url.rstrip("/")
    url = endpoint if endpoint.startswith("https://") else base + "/" + endpoint.lstrip("/")
    parsed, origin = urlsplit(url), urlsplit(base)
    if parsed.scheme != "https" or parsed.netloc != origin.netloc or parsed.username or parsed.password:
        raise ValueError("Endpoint must belong to the authenticated GitHub API host")
    headers = requests.structures.CaseInsensitiveDict(kwargs.pop("headers", {}))
    if "Authorization" in headers or "auth" in kwargs:
        raise ValueError("Select a GitHub actor instead of overriding its credentials")
    headers["Authorization"] = f"Bearer {self.token}"
    headers.setdefault("Accept", "application/vnd.github+json")
    kwargs.setdefault("timeout", 60)
    kwargs.setdefault("allow_redirects", False)
    request_id = uuid.uuid4().hex
    self._record("api_request", request_id=request_id, method=method.upper(), path=parsed.path)
    try:
        response = requests.request(method, url, headers=headers, **kwargs)
    except requests.RequestException:
        self._record("api_response", request_id=request_id, status=None)
        raise
    self._record(
        "api_response",
        request_id=request_id,
        status=response.status_code,
        github_request_id=response.headers.get("X-GitHub-Request-Id"),
    )
    response.raise_for_status()
    return response

graphql 

graphql(query: str, variables: dict | None = None) -> dict
Source code in src/benchmark/utils/gh_client.py 
103
104
105
106
107
def graphql(self, query: str, variables: dict | None = None) -> dict:
    result = self.request("POST", "/graphql", json={"query": query, "variables": variables or {}}).json()
    if result.get("errors"):
        raise RuntimeError(f"GitHub GraphQL errors: {result['errors']}")
    return result["data"]

get_authenticated_user_login 

get_authenticated_user_login() -> str

Returns the login of the authenticated user for this client.

Source code in src/benchmark/utils/gh_client.py 
136
137
138
139
140
141
def get_authenticated_user_login(self) -> str:
    """Returns the login of the authenticated user for this client."""
    if self._authenticated_login is None:
        rate_limiter.wait()
        self._authenticated_login = self.gh.get_user().login
    return self._authenticated_login

get_repo_info 

get_repo_info() -> Optional[Dict[str, Any]]

Fetches repository information.

Source code in src/benchmark/utils/gh_client.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def get_repo_info(self) -> Optional[Dict[str, Any]]:
    """Fetches repository information."""
    try:
        repo = self.repository
        return {
            "name": repo.name,
            "owner": {"login": repo.owner.login},
            "defaultBranchRef": {"name": repo.default_branch},
            "isEmpty": repo.size == 0,
            "size": repo.size,
            "default_branch": repo.default_branch,
        }
    except GithubException as e:
        if e.status == 404:
            return None
        raise

get_default_branch 

get_default_branch() -> str

Returns the name of the default branch.

Source code in src/benchmark/utils/gh_client.py 
168
169
170
171
172
173
def get_default_branch(self) -> str:
    """Returns the name of the default branch."""
    try:
        return self.repository.default_branch
    except GithubException:
        return "main"

create_repo 

create_repo(public: bool = True) -> Tuple[bool, str]

Creates the repository if it doesn't exist.

Source code in src/benchmark/utils/gh_client.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def create_repo(self, public: bool = True) -> Tuple[bool, str]:
    """Creates the repository if it doesn't exist."""
    if self._owned_repo is not None:
        return False, "Client already owns a repository awaiting cleanup"
    try:
        name = self.repo_name.split("/")[-1]
        if "/" in self.repo_name:
            owner = self.repo_name.split("/", 1)[0]
            user = self.gh.get_user()
            if user.login.lower() == owner.lower():
                repo = user.create_repo(name, private=not public)
            else:
                org = self.gh.get_organization(owner)
                repo = org.create_repo(name, private=not public)
        else:
            repo = self.gh.get_user().create_repo(name, private=not public)

        self.repo_name = repo.full_name
        self._repo_cache = repo
        self._claim_repository(repo)
        return True, ""
    except GithubException as e:
        return False, str(e)

fork_repo 

fork_repo(template_repo_name: str) -> Tuple[bool, str]

Fork the specified source, refusing collisions and existing owner forks.

Source code in src/benchmark/utils/gh_client.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def fork_repo(self, template_repo_name: str) -> Tuple[bool, str]:
    """Fork the specified source, refusing collisions and existing owner forks."""
    if self._owned_repo is not None:
        return False, "Client already owns a repository awaiting cleanup"
    try:
        user_login = self.get_authenticated_user_login()
        owner, name = self.repo_name.split("/", 1) if "/" in self.repo_name else (user_login, self.repo_name)
        target = f"{owner}/{name}"
        try:
            self.gh.get_repo(target)
        except GithubException as exc:
            if exc.status != 404:
                raise
        else:
            return False, f"Repository already exists: {target}"
        template = self.gh.get_repo(template_repo_name)
        for existing in template.get_forks():
            if existing.owner.login.lower() == owner.lower():
                return False, f"Owner already has a fork: {existing.full_name}; use a separate owner or repository copy"
        kwargs = {"name": name, "default_branch_only": True}
        if owner.lower() != user_login.lower():
            kwargs["organization"] = owner
        new_repo = template.create_fork(**kwargs)
        if new_repo.full_name.lower() != target.lower():
            return False, f"GitHub returned an unexpected fork: {new_repo.full_name}"
        self.repo_name = new_repo.full_name
        self._repo_cache = new_repo
        self._claim_repository(new_repo)
        return True, ""
    except Exception as exc:
        return False, str(exc)

wait_until_ready 

wait_until_ready()
Source code in src/benchmark/utils/gh_client.py 
231
232
233
234
235
236
237
238
239
@retry(
    retry=retry_if_exception_type(GithubException),
    stop=stop_after_attempt(15),
    wait=wait_exponential(multiplier=1, min=2, max=10),
)
def wait_until_ready(self):
    repo = self.gh.get_repo(self.repo_name)
    repo.get_branch(repo.default_branch)
    self._repo_cache = repo

delete_owned_repo 

delete_owned_repo() -> Tuple[bool, str]

Delete only the exact repository created through this client.

Source code in src/benchmark/utils/gh_client.py 
252
253
254
255
256
257
258
259
260
261
262
263
264
def delete_owned_repo(self) -> Tuple[bool, str]:
    """Delete only the exact repository created through this client."""
    if self._owned_repo is None:
        return True, ""
    name, repo_id = self._owned_repo
    try:
        self._delete_repository(name, repo_id)
    except (GithubException, RuntimeError) as exc:
        return False, str(exc)
    self._record("resource", name=name, id=repo_id, state="deleted")
    self._owned_repo = None
    self._repo_cache = None
    return True, ""

delete_repo 

delete_repo() -> Tuple[bool, str]

Deletes the current repository.

Source code in src/benchmark/utils/gh_client.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
@retry(
    retry=retry_if_exception_type(GithubException),
    stop=stop_after_attempt(10),
    wait=wait_exponential(multiplier=2, min=4, max=30),
)
def delete_repo(self) -> Tuple[bool, str]:
    """Deletes the current repository."""
    try:
        repo = self.repository
        repo.delete()
        if self._owned_repo == (self.repo_name, repo.id):
            self._owned_repo = None
        self._repo_cache = None
        return True, ""
    except GithubException as e:
        if e.status == 404:
            return True, ""  # Already gone
        if e.status == 403:
            if "delete_repo" in str(e):
                msg = (
                    "\nERROR: Missing 'delete_repo' scope. Please run:\n  gh auth refresh -h github.com -s delete_repo\n"
                )
                click.echo(click.style(msg, fg="yellow", bold=True))
            elif "done being created on disk" in str(e):
                # Re-raise to let the retry decorator handle it
                raise e
        return False, str(e)

get_branch_info 

get_branch_info(branch_name: str) -> Optional[Dict[str, Any]]

Checks if a branch exists and returns its info.

Source code in src/benchmark/utils/gh_client.py 
294
295
296
297
298
299
300
301
302
def get_branch_info(self, branch_name: str) -> Optional[Dict[str, Any]]:
    """Checks if a branch exists and returns its info."""
    try:
        branch = self.repository.get_branch(branch_name)
        return {"name": branch.name, "commit": {"sha": branch.commit.sha}}
    except GithubException as e:
        if e.status == 404:
            return None
        raise

create_branch 

create_branch(new_branch: str, source_branch: Optional[str] = None) -> Tuple[bool, str]

Creates a new branch from a source branch.

Source code in src/benchmark/utils/gh_client.py 
304
305
306
307
308
309
310
311
312
313
314
def create_branch(self, new_branch: str, source_branch: Optional[str] = None) -> Tuple[bool, str]:
    """Creates a new branch from a source branch."""
    if not source_branch:
        source_branch = self.get_default_branch()

    try:
        sb = self.repository.get_branch(source_branch)
        self.repository.create_git_ref(ref=f"refs/heads/{new_branch}", sha=sb.commit.sha)
        return True, ""
    except GithubException as e:
        return False, str(e)

get_file_sha 

get_file_sha(path: str, branch: Optional[str] = None) -> Optional[str]

Gets the SHA of a file on a specific branch.

Source code in src/benchmark/utils/gh_client.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def get_file_sha(self, path: str, branch: Optional[str] = None) -> Optional[str]:
    """Gets the SHA of a file on a specific branch."""
    try:
        kwargs = {}
        if branch:
            kwargs["ref"] = branch
        content = self.repository.get_contents(path, **kwargs)
        if isinstance(content, list):
            return None
        return content.sha
    except GithubException as e:
        if e.status == 404:
            return None
        raise

put_file 

put_file(path: str, content: str, message: str, branch: Optional[str] = None) -> Tuple[bool, str]

Uploads or updates a file using the GitHub API.

Source code in src/benchmark/utils/gh_client.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
@retry(
    retry=retry_if_exception_type(GithubException),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    stop=stop_after_attempt(5),
    before_sleep=before_sleep_log(logger, logging.INFO),
)
def put_file(self, path: str, content: str, message: str, branch: Optional[str] = None) -> Tuple[bool, str]:
    """Uploads or updates a file using the GitHub API."""
    if not branch:
        branch = self.get_default_branch()

    try:
        sha = self.get_file_sha(path, branch)
        if sha:
            self.repository.update_file(path, message, content, sha, branch=branch)
        else:
            self.repository.create_file(path, message, content, branch=branch)
        return True, ""
    except GithubException as e:
        if e.status == 409:
            raise e
        return False, str(e)

delete_file 

delete_file(path: str, message: str, branch: Optional[str] = None) -> Tuple[bool, str]

Deletes a file from the repository.

Source code in src/benchmark/utils/gh_client.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
@retry(
    retry=retry_if_exception_type(GithubException),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    stop=stop_after_attempt(5),
)
def delete_file(self, path: str, message: str, branch: Optional[str] = None) -> Tuple[bool, str]:
    """Deletes a file from the repository."""
    if not branch:
        branch = self.get_default_branch()

    try:
        sha = self.get_file_sha(path, branch)
        if sha:
            self.repository.delete_file(path, message, sha, branch=branch)
            return True, ""
        return True, "File not found"  # Already gone
    except GithubException as e:
        return False, str(e)

get_pr_details 

get_pr_details(pr_number: int) -> Dict[str, Any]

Fetch complete, attributed comments and formal reviews; propagate read errors.

Source code in src/benchmark/utils/gh_client.py 
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def get_pr_details(self, pr_number: int) -> Dict[str, Any]:
    """Fetch complete, attributed comments and formal reviews; propagate read errors."""
    pr = self.repository.get_pull(pr_number)
    comments = [self._artifact(c, "issue_comment") for c in pr.get_issue_comments()]
    reviews = [self._artifact(r, "review") for r in pr.get_reviews()]
    comments.extend(reviews)
    comments.extend(self._artifact(c, "review_comment") for c in pr.get_review_comments())
    return {
        "title": pr.title,
        "body": pr.body,
        "state": pr.state,
        "comments": [c["body"] for c in comments],
        "comment_details": comments,
        "reviews": reviews,
    }

get_issue_details 

get_issue_details(issue_number: int) -> Dict[str, Any]
Source code in src/benchmark/utils/gh_client.py 
402
403
404
405
406
407
408
409
410
411
def get_issue_details(self, issue_number: int) -> Dict[str, Any]:
    issue = self.repository.get_issue(issue_number)
    comments = [self._artifact(c, "issue_comment") for c in issue.get_comments()]
    return {
        "title": issue.title,
        "body": issue.body,
        "state": issue.state,
        "comments": [c["body"] for c in comments],
        "comment_details": comments,
    }

list_files 

list_files(branch: Optional[str] = None) -> List[str]

Lists files in a specific branch.

Source code in src/benchmark/utils/gh_client.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
@retry(
    retry=retry_if_exception_type(GithubException),
    wait=wait_exponential(multiplier=1, min=1, max=5),
    stop=stop_after_attempt(5),
)
def list_files(self, branch: Optional[str] = None) -> List[str]:
    """Lists files in a specific branch."""
    try:
        ref = branch or self.get_default_branch()
        tree = self.repository.get_git_tree(ref, recursive=True)
        return [item.path for item in tree.tree if item.type == "blob"]
    except GithubException as e:
        if e.status == 409:
            return []
        raise

set_secret 

set_secret(name: str, value: str) -> Tuple[bool, str]

Sets a repository secret.

Source code in src/benchmark/utils/gh_client.py 
429
430
431
432
433
434
435
def set_secret(self, name: str, value: str) -> Tuple[bool, str]:
    """Sets a repository secret."""
    try:
        self.repository.create_secret(name, value)
        return True, ""
    except GithubException as e:
        return False, str(e)

set_variable 

set_variable(name: str, value: str) -> Tuple[bool, str]

Sets a repository variable.

Source code in src/benchmark/utils/gh_client.py 
437
438
439
440
441
442
443
def set_variable(self, name: str, value: str) -> Tuple[bool, str]:
    """Sets a repository variable."""
    try:
        self.repository.create_variable(name, value)
        return True, ""
    except GithubException as e:
        return False, str(e)

enable_actions 

enable_actions() -> Tuple[bool, str]

Enables GitHub Actions for the repository.

Source code in src/benchmark/utils/gh_client.py 
445
446
447
448
449
450
451
452
453
454
455
456
def enable_actions(self) -> Tuple[bool, str]:
    """Enables GitHub Actions for the repository."""
    try:
        # Using gh CLI for simplicity as PyGitHub doesn't have a direct method for this
        stdout, stderr = self.run_gh(
            ["api", "-X", "PUT", f"repos/{self.repo_name}/actions/permissions", "-F", "enabled=true"]
        )
        if stderr and "error" in stderr.lower():
            return False, stderr
        return True, ""
    except Exception as e:
        return False, str(e)

set_fork_pr_approval_policy 

set_fork_pr_approval_policy(policy: str = 'first_time_contributors_new_to_github') -> Tuple[bool, str]

Sets the fork PR workflow approval policy for the repository.

Source code in src/benchmark/utils/gh_client.py 
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
def set_fork_pr_approval_policy(self, policy: str = "first_time_contributors_new_to_github") -> Tuple[bool, str]:
    """Sets the fork PR workflow approval policy for the repository."""
    try:
        stdout, stderr = self.run_gh(
            [
                "api",
                "-X",
                "PUT",
                f"/repos/{self.repo_name}/actions/permissions/fork-pr-contributor-approval",
                "-H",
                "X-GitHub-Api-Version: 2026-03-10",
                "-f",
                f"approval_policy={policy}",
            ]
        )
        if stderr and "error" in stderr.lower():
            return False, stderr
        return True, ""
    except Exception as e:
        return False, str(e)

enable_issues 

enable_issues() -> Tuple[bool, str]

Enables GitHub Issues for the repository.

Source code in src/benchmark/utils/gh_client.py 
479
480
481
482
483
484
485
486
487
488
def enable_issues(self) -> Tuple[bool, str]:
    """Enables GitHub Issues for the repository."""
    try:
        # Using gh CLI for simplicity as PyGitHub doesn't have a direct method for this
        stdout, stderr = self.run_gh(["repo", "edit", self.repo_name, "--enable-issues"])
        if stderr:
            return False, stderr
        return True, ""
    except Exception as e:
        return False, str(e)

list_repos 

list_repos(limit: int = 100) -> List[Dict[str, str]]

Lists repositories for the authenticated user.

Source code in src/benchmark/utils/gh_client.py 
490
491
492
493
494
495
496
def list_repos(self, limit: int = 100) -> List[Dict[str, str]]:
    """Lists repositories for the authenticated user."""
    try:
        repos = self.gh.get_user().get_repos()
        return [{"name": r.name, "nameWithOwner": r.full_name} for r in repos[:limit]]
    except GithubException:
        return []

get_workflow_runs 

get_workflow_runs(workflow_id: str = None) -> List[Any]

Fetches recent runs of a specific workflow or all workflows.

Source code in src/benchmark/utils/gh_client.py 
498
499
500
501
502
503
504
505
506
507
def get_workflow_runs(self, workflow_id: str = None) -> List[Any]:
    """Fetches recent runs of a specific workflow or all workflows."""
    try:
        if workflow_id:
            runs = self.repository.get_workflow(workflow_id).get_runs()
        else:
            runs = self.repository.get_workflow_runs()
        return [r for r in runs[:10]]
    except GithubException:
        return []

batch_sync 

batch_sync(additions: Dict[str, str], deletions: List[str], message: str, branch: Optional[str] = None) -> Tuple[bool, str]

Performs multiple additions and deletions in a single commit.

Source code in src/benchmark/utils/gh_client.py 
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
def batch_sync(
    self,
    additions: Dict[str, str],
    deletions: List[str],
    message: str,
    branch: Optional[str] = None,
) -> Tuple[bool, str]:
    """Performs multiple additions and deletions in a single commit."""
    if not branch:
        branch = self.get_default_branch()

    try:
        repo = self.repository

        @retry(
            retry=retry_if_exception_type(GithubException),
            stop=stop_after_attempt(10),
            wait=wait_exponential(multiplier=1, min=2, max=10),
        )
        def get_ref():
            rate_limiter.wait()
            return repo.get_git_ref(f"heads/{branch}")

        ref = get_ref()
        old_commit = repo.get_git_commit(ref.object.sha)
        base_tree_sha = old_commit.tree.sha

        # We avoid recursive=True because it fails with 502 on large repos like Sentry.
        # Instead, we will build the new tree by navigating only what we need.

        def get_tree_without_path(current_tree_sha, path_parts):
            """Recursively navigates trees to 'delete' a path by omitting it from a new tree."""
            tree = repo.get_git_tree(current_tree_sha, recursive=False)
            elements = []

            target = path_parts[0]
            remaining = path_parts[1:]

            found_target = False
            for item in tree.tree:
                if item.path == target:
                    found_target = True
                    if remaining:
                        # We need to go deeper into this subtree
                        if item.type == "tree":
                            new_subtree_sha = get_tree_without_path(item.sha, remaining)
                            if new_subtree_sha:
                                elements.append(
                                    InputGitTreeElement(
                                        path=item.path, mode=item.mode, type=item.type, sha=new_subtree_sha
                                    )
                                )
                            # if new_subtree_sha is None, it means the whole subtree was deleted
                        else:
                            # Target is a file, but we have remaining path parts?
                            # This means the path doesn't match the structure. Keep it as is.
                            elements.append(
                                InputGitTreeElement(path=item.path, mode=item.mode, type=item.type, sha=item.sha)
                            )

                    else:
                        # This is the item to delete! Just don't add it to elements.
                        pass
                else:
                    # Not our target, keep it
                    elements.append(InputGitTreeElement(path=item.path, mode=item.mode, type=item.type, sha=item.sha))

            if not found_target:
                # Target not found in this tree, nothing to delete here
                return current_tree_sha

            if not elements:
                # Entire tree is empty now
                return None

            new_tree = repo.create_git_tree(elements)
            return new_tree.sha

        current_tree_sha = base_tree_sha
        for deletion_path in deletions:
            # Normalize path: remove leading/trailing slashes
            clean_path = deletion_path.strip("/")
            if not clean_path:
                continue

            path_parts = clean_path.split("/")
            current_tree_sha = get_tree_without_path(current_tree_sha, path_parts)
            if not current_tree_sha:
                # We deleted everything? (unlikely but possible)
                # Create an empty tree to avoid errors
                empty_tree = repo.create_git_tree([])
                current_tree_sha = empty_tree.sha

        # 2. Handle additions using the base_tree merge capability
        elements = []
        for path, content in additions.items():
            elements.append(InputGitTreeElement(path=path, mode="100644", type="blob", content=content))

        # Create the final tree by merging additions into our modified tree
        final_tree = repo.create_git_tree(elements, base_tree=repo.get_git_tree(current_tree_sha))

        new_commit = repo.create_git_commit(message, final_tree, [old_commit])
        ref.edit(new_commit.sha)

        return True, ""
    except Exception as e:
        return False, str(e)

run_gh 

run_gh(args, **kwargs)

Legacy compatibility method. SHOULD BE REMOVED eventually.

Source code in src/benchmark/utils/gh_client.py 
617
618
619
620
621
622
623
624
625
626
627
628
def run_gh(self, args, **kwargs):
    """Legacy compatibility method. SHOULD BE REMOVED eventually."""
    cmd = ["gh"] + args
    if kwargs.get("use_repo"):
        cmd += ["-R", self.repo_name]

    env = os.environ.copy()
    env["GH_TOKEN"] = self.token
    result = subprocess.run(cmd, capture_output=True, text=True, env=env)
    if result.returncode != 0:
        raise RuntimeError(f"gh exited with status {result.returncode}: {result.stderr.strip()}")
    return result.stdout, result.stderr

Provisioning

RepoProvisioner.provision() creates a fresh public repository or template fork, installs workflow and scenario files, and configures Actions/issues/secrets/variables. It raises ProvisioningError on incomplete setup or conflicting paths. teardown() deletes only a repository owned by this provisioner and retains ownership on failure.

ProvisioningError

Bases: RuntimeError

Source code in src/benchmark/utils/provisioner.py 
8
9
class ProvisioningError(RuntimeError):
    pass

RepoProvisioner

Handles the physical setup of a GitHub repository for benchmarking.

Source code in src/benchmark/utils/provisioner.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
class RepoProvisioner:
    """Handles the physical setup of a GitHub repository for benchmarking."""

    def __init__(self, gh_client: GitHubClient):
        self.gh_client = gh_client
        self._owns_repo = False

    @staticmethod
    def _require(result, operation):
        success, error = result
        if not success:
            raise ProvisioningError(f"{operation}: {error}")

    def provision(
        self,
        workflow_dir: str,
        required_files: dict = None,
        branch: str = None,
        template_repo: str = None,
        secrets: dict = None,
        variables: dict = None,
        substitution_map: dict = None,
    ):
        """Create and configure a fresh repository, failing on incomplete setup."""
        if self._owns_repo:
            raise ProvisioningError("A previously created repository still needs cleanup")
        creation = self.gh_client.fork_repo(template_repo) if template_repo else self.gh_client.create_repo(public=True)
        self._require(creation, "Create repository")
        self._owns_repo = True
        if template_repo:
            self.gh_client.wait_until_ready()
        else:
            self._require(
                self.gh_client.put_file(
                    "README.md", "# Benchmark Repository\nGenerated by AI Benchmark Suite.", "initial commit", "main"
                ),
                "Initialize repository",
            )
        repo_info = self.gh_client.get_repo_info()
        if not repo_info:
            raise ProvisioningError("Created repository is not readable")
        default_branch = repo_info.get("defaultBranchRef", {}).get("name") or "main"
        target_branch = branch or default_branch
        self._require(self.gh_client.enable_actions(), "Enable Actions")
        self._require(self.gh_client.set_fork_pr_approval_policy(), "Set fork approval policy")
        self._require(self.gh_client.enable_issues(), "Enable issues")
        for operation, values in ((self.gh_client.set_secret, secrets), (self.gh_client.set_variable, variables)):
            for name, value in (values or {}).items():
                if value is None or value == "":
                    raise ProvisioningError(f"Empty configuration value: {name}")
                self._require(operation(name, value), f"Set configuration {name}")
        contents_dir = os.path.join(workflow_dir, "contents")
        workflow_files = {}
        if os.path.isdir(contents_dir):
            for root, _, filenames in os.walk(contents_dir):
                for filename in filenames:
                    local_path = os.path.join(root, filename)
                    workflow_files[os.path.relpath(local_path, contents_dir)] = local_path
        elif os.path.isdir(workflow_dir):
            for filename in os.listdir(workflow_dir):
                if filename.endswith((".yml", ".yaml")):
                    workflow_files[f".github/workflows/{filename}"] = os.path.join(workflow_dir, filename)
        scenario_files = required_files or {}
        for path in scenario_files:
            if path in workflow_files:
                raise ProvisioningError(f"File defined by both workflow and scenario: {path}")
        additions = {path: self._get_content(path, content, substitution_map) for path, content in workflow_files.items()}
        self._require(
            self.gh_client.batch_sync(additions, [".github/workflows/"], "provision workflows", default_branch),
            "Sync workflows",
        )
        if target_branch != default_branch and not self.gh_client.get_branch_info(target_branch):
            self._require(self.gh_client.create_branch(target_branch, default_branch), "Create target branch")
        if scenario_files:
            additions = {
                path: self._get_content(path, content, substitution_map) for path, content in scenario_files.items()
            }
            self._require(
                self.gh_client.batch_sync(additions, [], "provision scenario files", target_branch), "Sync scenario"
            )
        elif target_branch != default_branch:
            self._require(
                self.gh_client.put_file(".scanner-probe", "probe\n", "scanner probe commit", target_branch),
                "Create PR comparison commit",
            )

    def _get_content(self, repo_path, content_or_path, substitution_map):
        """Reads and optionally patches file content."""
        content = content_or_path
        is_binary = False
        if isinstance(content_or_path, str) and os.path.exists(content_or_path):
            with open(content_or_path, "rb") as f:
                content = f.read()
                try:
                    content = content.decode("utf-8")
                    if substitution_map and (repo_path.endswith(".yml") or repo_path.endswith(".yaml")):
                        content = self._patch_yaml(content, substitution_map)
                except UnicodeDecodeError:
                    is_binary = True

        if is_binary:
            return content.decode("latin-1")
        return content

    def _patch_yaml(self, content: str, substitution_map: dict) -> str:
        """Replaces official action references with adversarial forks/tags."""
        import re

        patched_content = content
        for original, replacement in substitution_map.items():
            pattern = rf"uses:\s*['\"]?{re.escape(original)}['\"]?"
            replacement_str = f"uses: {replacement}"
            if re.search(pattern, patched_content):
                click.echo(click.style(f"  PATCHED: Swapping '{original}' -> '{replacement}'", fg="cyan"))
                patched_content = re.sub(pattern, replacement_str, patched_content)

        return patched_content

    def teardown(self):
        """Clean up a repository created by this provisioner, retaining ownership on failure."""
        if not self._owns_repo:
            return
        self._require(self.gh_client.delete_owned_repo(), "Delete owned repository")
        self._owns_repo = False

__init__ 

__init__(gh_client: GitHubClient)
Source code in src/benchmark/utils/provisioner.py 
15
16
17
def __init__(self, gh_client: GitHubClient):
    self.gh_client = gh_client
    self._owns_repo = False

provision 

provision(workflow_dir: str, required_files: dict = None, branch: str = None, template_repo: str = None, secrets: dict = None, variables: dict = None, substitution_map: dict = None)

Create and configure a fresh repository, failing on incomplete setup.

Source code in src/benchmark/utils/provisioner.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def provision(
    self,
    workflow_dir: str,
    required_files: dict = None,
    branch: str = None,
    template_repo: str = None,
    secrets: dict = None,
    variables: dict = None,
    substitution_map: dict = None,
):
    """Create and configure a fresh repository, failing on incomplete setup."""
    if self._owns_repo:
        raise ProvisioningError("A previously created repository still needs cleanup")
    creation = self.gh_client.fork_repo(template_repo) if template_repo else self.gh_client.create_repo(public=True)
    self._require(creation, "Create repository")
    self._owns_repo = True
    if template_repo:
        self.gh_client.wait_until_ready()
    else:
        self._require(
            self.gh_client.put_file(
                "README.md", "# Benchmark Repository\nGenerated by AI Benchmark Suite.", "initial commit", "main"
            ),
            "Initialize repository",
        )
    repo_info = self.gh_client.get_repo_info()
    if not repo_info:
        raise ProvisioningError("Created repository is not readable")
    default_branch = repo_info.get("defaultBranchRef", {}).get("name") or "main"
    target_branch = branch or default_branch
    self._require(self.gh_client.enable_actions(), "Enable Actions")
    self._require(self.gh_client.set_fork_pr_approval_policy(), "Set fork approval policy")
    self._require(self.gh_client.enable_issues(), "Enable issues")
    for operation, values in ((self.gh_client.set_secret, secrets), (self.gh_client.set_variable, variables)):
        for name, value in (values or {}).items():
            if value is None or value == "":
                raise ProvisioningError(f"Empty configuration value: {name}")
            self._require(operation(name, value), f"Set configuration {name}")
    contents_dir = os.path.join(workflow_dir, "contents")
    workflow_files = {}
    if os.path.isdir(contents_dir):
        for root, _, filenames in os.walk(contents_dir):
            for filename in filenames:
                local_path = os.path.join(root, filename)
                workflow_files[os.path.relpath(local_path, contents_dir)] = local_path
    elif os.path.isdir(workflow_dir):
        for filename in os.listdir(workflow_dir):
            if filename.endswith((".yml", ".yaml")):
                workflow_files[f".github/workflows/{filename}"] = os.path.join(workflow_dir, filename)
    scenario_files = required_files or {}
    for path in scenario_files:
        if path in workflow_files:
            raise ProvisioningError(f"File defined by both workflow and scenario: {path}")
    additions = {path: self._get_content(path, content, substitution_map) for path, content in workflow_files.items()}
    self._require(
        self.gh_client.batch_sync(additions, [".github/workflows/"], "provision workflows", default_branch),
        "Sync workflows",
    )
    if target_branch != default_branch and not self.gh_client.get_branch_info(target_branch):
        self._require(self.gh_client.create_branch(target_branch, default_branch), "Create target branch")
    if scenario_files:
        additions = {
            path: self._get_content(path, content, substitution_map) for path, content in scenario_files.items()
        }
        self._require(
            self.gh_client.batch_sync(additions, [], "provision scenario files", target_branch), "Sync scenario"
        )
    elif target_branch != default_branch:
        self._require(
            self.gh_client.put_file(".scanner-probe", "probe\n", "scanner probe commit", target_branch),
            "Create PR comparison commit",
        )

teardown 

teardown()

Clean up a repository created by this provisioner, retaining ownership on failure.

Source code in src/benchmark/utils/provisioner.py 
130
131
132
133
134
135
def teardown(self):
    """Clean up a repository created by this provisioner, retaining ownership on failure."""
    if not self._owns_repo:
        return
    self._require(self.gh_client.delete_owned_repo(), "Delete owned repository")
    self._owns_repo = False

Attacker forks

These helpers support legacy scenarios using attacker forks. Setup requires ATTACKER_GITHUB_TOKEN, creates an independently owned fork, and installs scenario fixture files on scenario.branch. The fork client is attached to the scenario before later setup operations so teardown can clean up partial setup. Fixtures must be local file paths. The scenario must call the teardown helper from its cleanup hook.

scenario_resources

setup_attacker_fork 

setup_attacker_fork(scenario, owner_client)

Create a fresh attacker fork and record ownership before subsequent setup.

Source code in src/benchmark/utils/scenario_resources.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def setup_attacker_fork(scenario, owner_client):
    """Create a fresh attacker fork and record ownership before subsequent setup."""
    token = os.environ.get("ATTACKER_GITHUB_TOKEN")
    if not token:
        raise ValueError("Required GitHub actor is unavailable: attacker")
    client = GitHubClient(token=token, actor="attacker", record_event=owner_client.record_event)
    login = client.get_authenticated_user_login()
    client.repo_name = f"{login}/{owner_client.repository.name}"
    RepoProvisioner._require(client.fork_repo(owner_client.repo_name), "Create attacker fork")
    scenario._attacker_fork_client = client
    scenario.runtime_state.update(attacker_login=login, fork_full_name=client.repo_name)
    client.wait_until_ready()
    RepoProvisioner._require(client.create_branch(scenario.branch, client.get_default_branch()), "Create attacker branch")
    additions = {}
    for path, local_path in scenario.get_required_files().items():
        with open(local_path) as handle:
            additions[path] = handle.read()
    RepoProvisioner._require(
        client.batch_sync(additions, [], "provision attacker fixtures", scenario.branch), "Sync attacker fixtures"
    )

teardown_attacker_fork 

teardown_attacker_fork(scenario)
Source code in src/benchmark/utils/scenario_resources.py 
29
30
31
32
33
def teardown_attacker_fork(scenario):
    client = getattr(scenario, "_attacker_fork_client", None)
    if client is not None:
        RepoProvisioner._require(client.delete_owned_repo(), "Delete attacker fork")
        scenario._attacker_fork_client = None