PR reviewer as a GitHub Action: a typed review, and the PR's tests without secrets

git clone https://github.com/more-itertools/more-itertools && cd more-itertools
git fetch origin pull/1342/head && git checkout FETCH_HEAD
BASE=4aeb433 python path/to/pr_reviewer.py
python -m unittest tests.test_more > test-results.txt 2>&1
BASE=4aeb433 python path/to/pr_reviewer.py

name: Agent tests

# The PR's own copy of this workflow runs: with no secrets and a read-only token.
on: pull_request

permissions:
  contents: read

jobs:
  tests:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    steps:
      - uses: actions/checkout@v7
        with:
          persist-credentials: false
      - uses: actions/setup-python@v7
        with:
          python-version: "3.12"
      # Your project: add your test dependencies here. A failed install is in the output too.
      - run: |
          mkdir -p "$RUNNER_TEMP/out" && echo "$PR_NUMBER" > "$RUNNER_TEMP/out/pr-number"
          pip install pytest -e . > "$RUNNER_TEMP/out/test-results.txt" 2>&1 &&
            python -m pytest -q >> "$RUNNER_TEMP/out/test-results.txt" 2>&1 || true
        env:
          PR_NUMBER: ${{ github.event.pull_request.number }}
      - uses: actions/upload-artifact@v7
        with:
          name: test-results
          path: ${{ runner.temp }}/out

name: Agent review

on:
  # After each run of Agent tests on a pull request. This workflow and pr_reviewer.py come from the
  # default branch, with the repository's secrets; no code from the pull request runs here.
  workflow_run:
    workflows: [Agent tests]
    types: [completed]

permissions: {}

jobs:
  review:
    if: >-
      github.event.workflow_run.event == 'pull_request' && github.event.workflow_run.conclusion == 'success'
    runs-on: ubuntu-latest
    timeout-minutes: 20
    permissions:
      actions: read  # the artifact of the tests run
      contents: read
      pull-requests: write
    env:
      GH_TOKEN: ${{ github.token }}
      HEAD: ${{ github.event.workflow_run.head_sha }}
    steps:
      - uses: actions/checkout@v7  # the default branch
        with:
          fetch-depth: 0
      - uses: actions/download-artifact@v8
        with:
          name: test-results
          path: ${{ runner.temp }}/tests
          run-id: ${{ github.event.workflow_run.id }}
          github-token: ${{ github.token }}
      # The pull request wrote the artifact: its number must name the pull request whose head was tested.
      - run: |
          number=$(head -c 20 "$RUNNER_TEMP/tests/pr-number")
          [[ "$number" =~ ^[0-9]+$ ]] || { echo "The artifact's PR number is not a number."; exit 1; }
          read -r pr_head base < <(gh pr view "$number" --json headRefOid,baseRefOid --jq '.headRefOid + " " + .baseRefOid')
          [ "$pr_head" = "$HEAD" ] || { echo "PR $number's head is not the tested commit: a newer push gets its own run."; exit 1; }
          git fetch --no-tags origin "refs/pull/$number/head"  # the PR's commits, read as text only
          git cat-file -e "$HEAD^{commit}" || { echo "The tested commit is gone: a newer push gets its own run."; exit 1; }
          echo "PR_NUMBER=$number" >> "$GITHUB_ENV"
          echo "BASE=$base" >> "$GITHUB_ENV"
      - uses: actions/setup-python@v7
        with:
          python-version: "3.12"
      - run: pip install "mistralai-agents==1.1.0"
      - run: python .github/scripts/pr_reviewer.py
        env:
          MISTRAL_API_KEY: ${{ secrets.MISTRAL_API_KEY }}
          TEST_RESULTS: ${{ runner.temp }}/tests/test-results.txt

import asyncio
import os
import subprocess
from pathlib import Path
from typing import Literal

from pydantic import BaseModel

from mistralai import agents

class Review(BaseModel):
    verdict: Literal["approve", "request_changes"]
    comments: list[str]

def git(*args: str) -> str:
    return subprocess.run(["git", *args], capture_output=True, text=True, check=True).stdout

def test_results_file() -> Path:
    return Path(os.environ.get("TEST_RESULTS", "test-results.txt"))

@agents.tool(annotations=agents.ToolAnnotations(read_only_hint=True, destructive_hint=False))
def read_file(path: str) -> str:
    """Read a file of the repository at the PR's head."""
    shown = subprocess.run(["git", "show", f"{os.environ.get('HEAD', 'HEAD')}:{path}"], capture_output=True, text=True)
    if shown.returncode:  # such as a path the PR deletes: the model reads git's message
        raise agents.ToolError(shown.stderr.strip())
    return shown.stdout

@agents.tool
def test_results() -> str:
    """The output of the PR's tests, from the run that ran them."""
    return test_results_file().read_text()[-3000:]

@agents.tool
def push_fix(path: str, content: str) -> str:
    """Commit a fix to a file and push it to the PR's branch."""
    with open(path, "w") as file:
        file.write(content)
    git("commit", "--quiet", f"--message=Fix {path}", "--", path)
    git("push", "--quiet", "origin", f"HEAD:refs/heads/{os.environ['GITHUB_HEAD_REF']}")
    return "pushed"

reviewer = agents.Agent(
    model="mistral-medium-latest",
    name="pr-reviewer",
    instructions=(
        "You review pull requests of a Python repository. Read each file the diff changes with "
        "read_file. When the diff changes code or tests, read the test results with test_results. "
        "If you see a typo, fix it with push_fix. Then give your verdict, and one comment per "
        "finding that names the file and the function."
    ),
    tools=[read_file, test_results, push_fix],
    output_type=Review,  # the model is asked for a Review
    permission_mode="ask",  # the default "auto" asks nothing, so no handler runs
)

def ci_policy(
    ctx: agents.ApprovalCallbackContext, request: agents.ApprovalRequest
) -> agents.AllowApproval | agents.DenyApproval | agents.KeepCallbackOpen:
    if request.tool.annotations.read_only_hint:
        return agents.AllowApproval()
    if request.tool.id == "client.test_results":
        if test_results_file().exists():
            return agents.AllowApproval()  # the tests run wrote them
        return agents.KeepCallbackOpen()  # no test results: the turn comes back Blocked
    return agents.DenyApproval()  # CI is read-only: no push

async def review_pr(reviewer: agents.Agent, diff: str) -> str:
    session = await reviewer.start_session(on_approval=ci_policy)
    try:
        state = await session.run(diff, timeout=900)
    finally:
        await session.cancel()  # the CI job ends here: a turn still running or waiting stops with it
    match state:
        case agents.Completed(output=agents.InvalidOutput(error=error)):
            return f"Invalid review: {error}"  # a value: run the review again, in a new job
        case agents.Completed(output=Review() as review):
            return "\n".join([f"Verdict: {review.verdict}", *(f"- {comment}" for comment in review.comments)])
        case agents.Blocked(callbacks=callbacks):
            asked = ", ".join(callback.title for callback in callbacks)
            return f"Stopped: {asked} No test results: write them to {test_results_file()}."
        case agents.Running():
            return "Stopped: the review took more than 15 minutes."
        case agents.Failed(error=error):
            return f"Failed: {error.message}"
        case other:  # Interrupted, or a turn that ended without a review
            return f"No review: the turn ended as {type(other).__name__}."

def post(body: str) -> None:
    if number := os.environ.get("PR_NUMBER"):
        subprocess.run(["gh", "pr", "review", number, "--comment", "--body", body], check=True)
    else:
        print(body)  # a run outside the workflow posts nothing

async def main() -> None:
    diff = git("diff", f"{os.environ.get('BASE', 'origin/main')}...{os.environ.get('HEAD', 'HEAD')}")
    post(await review_pr(reviewer, diff))

if __name__ == "__main__":
    asyncio.run(main())

"""The tests of the PR reviewer example: a scripted model, no API key.

Run them beside pr_reviewer.py: pytest -o asyncio_mode=auto test_pr_reviewer.py
"""

import json
from pathlib import Path

import pr_reviewer
import pytest

from mistralai import agents

def review(verdict: str, comment: str) -> agents.testing.ReplyStep:
    return agents.testing.reply(json.dumps({"verdict": verdict, "comments": [comment]}))

def tool(name: str, arguments: dict[str, str]) -> agents.testing.ToolCallStep:
    # The model calls each tool from code, as a real model does.
    return agents.testing.run_typescript(f"return await tools.client.{name}({json.dumps(arguments)});")

def in_scratch_folder(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> list[tuple[str, str]]:
    """Run in an empty folder, and record each answer of the policy."""
    monkeypatch.chdir(tmp_path)  # push_fix would write here, and the test results are read here
    monkeypatch.delenv("TEST_RESULTS", raising=False)
    answers = []
    policy = pr_reviewer.ci_policy

    def recording_policy(
        ctx: agents.ApprovalCallbackContext, request: agents.ApprovalRequest
    ) -> agents.AllowApproval | agents.DenyApproval | agents.KeepCallbackOpen:
        answer = policy(ctx, request)
        answers.append((request.tool.id, type(answer).__name__))
        return answer

    monkeypatch.setattr(pr_reviewer, "ci_policy", recording_policy)
    return answers

async def test_reads_run_and_a_push_is_denied(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    answers = in_scratch_folder(monkeypatch, tmp_path)
    (tmp_path / "more.py").write_text("original\n")
    model = agents.testing.ScriptedModel(
        [
            tool("read_file", {"path": "more.py"}),
            tool("push_fix", {"path": "more.py", "content": "# fixed"}),
            review("request_changes", "more.py, chunked_even: n=0 loops forever."),
        ]
    )

    result = await pr_reviewer.review_pr(pr_reviewer.reviewer.override(model=model), "diff")

    model.assert_consumed()
    assert result == "Verdict: request_changes\n- more.py, chunked_even: n=0 loops forever."
    assert answers == [("client.read_file", "AllowApproval"), ("client.push_fix", "DenyApproval")]
    assert (tmp_path / "more.py").read_text() == "original\n"  # push_fix never ran

async def test_an_answer_that_is_not_a_review_is_a_value() -> None:
    model = agents.testing.ScriptedModel([agents.testing.reply("Looks good to me!")])

    result = await pr_reviewer.review_pr(pr_reviewer.reviewer.override(model=model), "diff")

    assert result.startswith("Invalid review: ")

async def test_without_test_results_the_run_stops_and_leaves_no_session(
    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
    answers = in_scratch_folder(monkeypatch, tmp_path)
    model = agents.testing.ScriptedModel([tool("test_results", {})])
    cancelled = []
    cancel = agents.Session.cancel

    async def record_cancel(session: agents.Session) -> None:
        cancelled.append(session.id)
        await cancel(session)

    monkeypatch.setattr(agents.Session, "cancel", record_cancel)

    result = await pr_reviewer.review_pr(pr_reviewer.reviewer.override(model=model), "diff")

    assert result.startswith("Stopped: ")
    assert result.endswith("No test results: write them to test-results.txt.")
    assert answers == [("client.test_results", "KeepCallbackOpen")]
    assert len(cancelled) == 1  # the job leaves no session waiting

async def test_with_test_results_the_agent_reads_them(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    answers = in_scratch_folder(monkeypatch, tmp_path)
    (tmp_path / "test-results.txt").write_text("42 passed in 1.20s\n")  # what the tests run wrote
    model = agents.testing.ScriptedModel(
        [
            tool("test_results", {}),
            review("approve", "All tests pass."),
        ]
    )

    result = await pr_reviewer.review_pr(pr_reviewer.reviewer.override(model=model), "diff")

    assert result == "Verdict: approve\n- All tests pass."
    assert answers == [("client.test_results", "AllowApproval")]