adam/open-swe
1
0
Fork 0
mirror of https://github.com/Sea-Haven-Industries/open-swe.git synced 2026-10-07 16:19:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Restore CI auto-fix on the modular dispatch model

Browse source 
Bring back ci_autofix.py and the ci_monitor graph that #1621 deleted,
re-wiring the fork's security-reviewed PR-babysitting onto the new
structure: the CI-event, autofix-toggle, and review-feedback handlers
move into webhooks/github.py and the github_webhook router re-gains the
check_run/check_suite/workflow_run/status routing plus the autofix
command and actionable-review branches. Auto-fix runs now dispatch
through dispatch_agent_run (durability + completion webhook) while
keeping the deliberate batch-while-busy skip-rule via
get_thread_active_status. Restore langgraph.json's ci_monitor entry and
the fork autofix tests (dispatch mock + import paths re-pointed).

Refs: #80
This commit is contained in:
amoussa1229 2026-06-30 21:22:59 +00:00
parent 3f78c3ab15
commit cc3f97df4a
7 changed files with 1313 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

610
agent/ci_autofix.py Normal file
View file

@ -0,0 +1,610 @@
"""Auto-fix CI failures and review feedback on agent-authored pull requests.
This is the shared core for "PR babysitting": when a CI check fails (or a
reviewer leaves actionable feedback) on a PR that Open SWE opened, locate the
originating agent thread and dispatch a confidence-gated fix run on it.
Both the GitHub webhook path (:mod:`agent.webapp`) and the polling fallback
(:mod:`agent.ci_monitor`) call into here, so all the skip-rules, dedupe, and
loop-capping live in one place. Skip-rules mirror Cursor/Claude Code:
* Only PRs Open SWE authored (an agent thread with this ``pr_url`` exists).
* Skip failures inherited from the base branch.
* Skip when the latest commit was authored by a human (don't fight pushes).
* Dedupe per head SHA; cap total attempts.
* Honor the per-user ``auto_fix_ci`` profile flag and the per-PR opt-out.
"""
from __future__ import annotations
import logging
from typing import Any
from langgraph_sdk import get_client
from .dashboard.agent_overrides import load_profile, resolve_login_from_email_async
from .dashboard.autofix_state import is_pr_autofix_disabled
from .dashboard.enabled_repos import is_review_repo_enabled
from .dispatch import dispatch_agent_run
from .reviewer_findings import REVIEWER_THREAD_KIND
from .utils.dashboard_links import dashboard_thread_url
from .utils.github_app import get_github_app_installation_token
from .utils.github_checks import post_autofix_status_check
from .utils.github_ci import (
fetch_open_pr_for_branch,
fetch_pr,
has_repo_write_permission,
head_commit_author_login,
list_failing_check_runs,
list_failing_statuses,
names_failing_on_base,
)
from .utils.github_org_membership import INTERNAL_BOT_LOGINS
from .utils.thread_ops import (
get_thread_active_status,
langgraph_client,
)
logger = logging.getLogger(__name__)
# Hard cap on auto-fix follow-ups per PR so a failure the agent can't resolve
# doesn't loop forever (Cursor caps at 10).
MAX_AUTOFIX_ATTEMPTS = 10
# Keep the dedupe list bounded on thread metadata.
_MAX_HANDLED_KEYS = 30
# Store location for batched auto-fix events, consumed by the message-queue middleware.
_PENDING_AUTOFIX_NS = "autofix"
_PENDING_AUTOFIX_KEY = "pending_event"
def _dedupe_key(head_sha: str) -> str:
return head_sha
async def _user_autofix_enabled(github_login: str, user_email: str = "") -> bool:
"""Check the per-user ``auto_fix_ci`` profile flag (defaults to True)."""
login = github_login.strip() if isinstance(github_login, str) else ""
if not login and user_email:
login = await resolve_login_from_email_async(user_email)
if not login:
return True
profile = await load_profile(login)
if not isinstance(profile, dict):
return True
value = profile.get("auto_fix_ci")
return value if isinstance(value, bool) else True
async def find_agent_thread_for_pr(pr_url: str) -> tuple[str, dict[str, Any]] | None:
"""Return ``(thread_id, metadata)`` of the agent thread that opened ``pr_url``.
Reviewer threads are skipped — only the coding-agent thread can push fixes.
"""
if not pr_url:
return None
client = get_client()
try:
threads = await client.threads.search(metadata={"pr_url": pr_url}, limit=10)
except Exception: # noqa: BLE001
logger.debug("Could not search threads for PR %s", pr_url, exc_info=True)
return None
for thread in threads or []:
metadata = thread.get("metadata") if isinstance(thread, dict) else None
if not isinstance(metadata, dict):
continue
if metadata.get("kind") == REVIEWER_THREAD_KIND:
continue
if metadata.get("agent_kind") != "agent":
continue
thread_id = thread.get("thread_id") or thread.get("id")
if isinstance(thread_id, str) and thread_id:
return thread_id, metadata
return None
def _build_ci_fix_prompt(
*,
owner: str,
repo: str,
pr_number: int,
pr_url: str,
branch: str,
head_sha: str,
failing_checks: list[dict[str, Any]],
) -> str:
lines = []
for check in failing_checks:
name = check.get("name", "check")
conclusion = check.get("conclusion", "failure")
details = check.get("details_url") or ""
suffix = f" — {details}" if details else ""
lines.append(f"- {name} ({conclusion}){suffix}")
failing_block = "\n".join(lines)
return (
"An automated CI check failed on a pull request you opened. Please "
"investigate and fix it.\n\n"
f"## Repository: {owner}/{repo}\n\n"
f"## Pull Request: {pr_url} (#{pr_number})\n\n"
f"## Branch: {branch}\n\n"
f"## Head commit: {head_sha}\n\n"
f"## Failing checks:\n{failing_block}\n\n"
"Instructions:\n"
"1. Make sure you are on the PR branch, then read the failing logs "
"(e.g. `GH_TOKEN=dummy gh pr checks` and `GH_TOKEN=dummy gh run view "
"<run-id> --log-failed`).\n"
"2. Confidence gating — fix autonomously ONLY when the cause is clear "
"and deterministic (lint/format, type errors, missing imports, failed "
"assertions, snapshot updates, build errors). Commit and push to the "
"existing branch; do NOT open a new PR.\n"
"3. If the failure is ambiguous, flaky, infrastructure-related, appears "
"pre-existing, or needs an architectural/design decision, do NOT guess. "
"Post a short PR comment explaining what you found and what input you "
"need, then stop.\n"
"4. Never force-push. Never weaken or delete test assertions just to go "
"green unless the behavior change is intentional and correct.\n"
"5. Before finishing, re-check the PR's latest CI status and review "
"comments. Address any newly failed checks or unhandled actionable "
"comments that arrived while you were working.\n"
"6. After you push, CI re-runs automatically — you don't need to merge."
)
def _build_review_feedback_prompt(
*,
owner: str,
repo: str,
pr_number: int,
pr_url: str,
reviewer: str,
body: str,
) -> str:
return (
"A reviewer left feedback on a pull request you opened. Please respond.\n\n"
f"## Repository: {owner}/{repo}\n\n"
f"## Pull Request: {pr_url} (#{pr_number})\n\n"
f"## Reviewer: {reviewer}\n\n"
f"## Feedback:\n{body}\n\n"
"Instructions:\n"
"1. If the requested change is unambiguous (rename, typo, missing null "
"check, small refactor, add a test), make it, commit, and push to the "
"existing branch.\n"
"2. If the comment is ambiguous, opinion-based, or needs a design "
"decision, reply on the PR asking for clarification instead of guessing.\n"
"3. Before finishing, re-check the PR's latest review comments and CI "
"status. Address any newly arrived actionable comments or failed checks "
"that are clear and deterministic.\n"
"4. Never force-push. Reply to the reviewer on GitHub to explain what "
"you changed."
)
async def _thread_autofix_state(metadata: dict[str, Any]) -> tuple[int, list[str], str, str]:
attempts = metadata.get("autofix_attempts")
attempts = attempts if isinstance(attempts, int) and attempts >= 0 else 0
handled = metadata.get("autofix_handled")
handled = [h for h in handled if isinstance(h, str)] if isinstance(handled, list) else []
github_login = metadata.get("github_login")
github_login = github_login if isinstance(github_login, str) else ""
user_email = metadata.get("triggering_user_email")
user_email = user_email if isinstance(user_email, str) else ""
return attempts, handled, github_login, user_email
async def _record_attempt(
thread_id: str, *, attempts: int, handled: list[str], dedupe_key: str, head_sha: str
) -> None:
new_handled = [*handled, dedupe_key][-_MAX_HANDLED_KEYS:]
try:
await get_client().threads.update(
thread_id=thread_id,
metadata={
"autofix_attempts": attempts + 1,
"autofix_handled": new_handled,
"autofix_last_head_sha": head_sha,
},
)
except Exception: # noqa: BLE001
logger.debug("Failed to record auto-fix attempt for thread %s", thread_id, exc_info=True)
# Run sources the agent's GitHub-token resolver knows how to authenticate.
_AUTH_RESOLVABLE_SOURCES = frozenset(["github", "slack", "dashboard", "linear", "schedule"])
def _run_configurable(
metadata: dict[str, Any], *, repo_config: dict[str, str], pr_number: int
) -> dict[str, Any]:
"""Build the run config for a fix run by reusing the PR thread's identity.
The agent's GitHub-token resolver only authenticates known sources, so a
bespoke ``github_ci`` source would fail in non-bot-token deployments. Reuse
the originating thread's ``source`` + login/email so auth resolves exactly
as it did for the run that opened the PR.
"""
source = metadata.get("source")
if source not in _AUTH_RESOLVABLE_SOURCES:
source = "github"
configurable: dict[str, Any] = {
"source": source,
"repo": repo_config,
"pr_number": pr_number,
}
login = metadata.get("github_login")
if isinstance(login, str) and login:
configurable["github_login"] = login
email = metadata.get("triggering_user_email")
if isinstance(email, str) and email:
configurable["user_email"] = email
return configurable
async def _mark_pending_autofix_event(thread_id: str, reason: str, detail: str = "") -> None:
"""Record a batched auto-fix event in the store the message-queue middleware reads.
Uses the same store namespace mechanism as ``queue_message_for_thread`` so the
in-flight run picks it up in-process at its next ``before_model`` step — no
per-step thread fetch. ``detail`` (e.g. a reviewer's comment) is accumulated so
specifics aren't lost when several events batch against one busy run.
"""
client = langgraph_client()
namespace = (_PENDING_AUTOFIX_NS, thread_id)
try:
details: list[str] = []
try:
existing = await client.store.get_item(namespace, _PENDING_AUTOFIX_KEY)
if existing and existing.get("value"):
prior = existing["value"].get("details")
if isinstance(prior, list):
details = [d for d in prior if isinstance(d, str)]
except Exception: # noqa: BLE001
logger.debug("No existing pending auto-fix event for thread %s", thread_id)
if detail and detail not in details:
details.append(detail)
await client.store.put_item(
namespace, _PENDING_AUTOFIX_KEY, {"reason": reason, "details": details}
)
except Exception: # noqa: BLE001
logger.debug(
"Failed to record pending auto-fix event for thread %s", thread_id, exc_info=True
)
async def _dispatch_or_batch(
thread_id: str, prompt: str, *, configurable: dict[str, Any], reason: str, detail: str = ""
) -> str:
# Deliberate skip-rule: batch auto-fix events while the agent thread is
# actively running so we don't interrupt an in-progress fix. ``interrupt``
# is fine for human follow-ups but undesirable for autofix, so we keep the
# busy-check here even though the webhook hot-path no longer needs one.
if await get_thread_active_status(thread_id) is True:
logger.info("Agent thread %s busy; batching auto-fix event %s", thread_id, reason)
await _mark_pending_autofix_event(thread_id, reason, detail)
return "batched"
await dispatch_agent_run(
thread_id,
prompt,
configurable,
source=str(configurable.get("source") or "github_autofix"),
)
logger.info(
"Created auto-fix run for thread %s (source=%s)", thread_id, configurable.get("source")
)
return "dispatched"
async def handle_ci_failure(
*,
repo_config: dict[str, str],
branch: str,
head_sha: str,
token: str | None = None,
source: str = "github_ci",
failing_checks: list[dict[str, Any]] | None = None,
pr: dict[str, Any] | None = None,
) -> str:
"""Auto-fix failing CI on an agent-authored PR. Returns a status string."""
owner = repo_config.get("owner", "")
repo = repo_config.get("name", "")
if not owner or not repo:
return "missing_repo"
if not await is_review_repo_enabled(owner, repo):
return "repo_not_enabled"
if token is None:
token = await get_github_app_installation_token()
if not token:
logger.warning("No GitHub App token for CI auto-fix on %s/%s", owner, repo)
return "no_token"
if pr is None:
if not branch:
return "no_branch"
pr = await fetch_open_pr_for_branch(owner=owner, repo=repo, branch=branch, token=token)
if not pr:
return "no_open_pr"
pr_number = pr.get("number")
if not isinstance(pr_number, int):
return "no_pr_number"
pr_url = pr.get("html_url") or pr.get("url") or ""
base_sha = (pr.get("base") or {}).get("sha", "")
branch = branch or (pr.get("head") or {}).get("ref", "")
head_sha = head_sha or (pr.get("head") or {}).get("sha", "")
if not head_sha:
return "no_head_sha"
if await is_pr_autofix_disabled(owner, repo, pr_number):
return "pr_disabled"
found = await find_agent_thread_for_pr(pr_url)
if found is None:
return "no_agent_thread"
thread_id, metadata = found
attempts, handled, github_login, user_email = await _thread_autofix_state(metadata)
if not await _user_autofix_enabled(github_login, user_email):
return "autofix_disabled_user"
if attempts >= MAX_AUTOFIX_ATTEMPTS:
await post_autofix_status_check(
owner=owner,
repo=repo,
head_sha=head_sha,
token=token,
title="Auto-fix limit reached",
summary=(
f"Open SWE has attempted {attempts} auto-fixes on this PR and "
"stopped to avoid a loop. Push a commit or comment to continue."
),
details_url=dashboard_thread_url(thread_id),
)
return "max_attempts"
if failing_checks is None:
runs = await list_failing_check_runs(owner=owner, repo=repo, ref=head_sha, token=token)
statuses = await list_failing_statuses(owner=owner, repo=repo, ref=head_sha, token=token)
if runs is None and statuses is None:
return "ci_read_failed"
failing_checks = (runs or []) + (statuses or [])
if not failing_checks:
return "no_failing_checks"
base_failing = await names_failing_on_base(
owner=owner, repo=repo, base_sha=base_sha, token=token
)
actionable = [c for c in failing_checks if c.get("name") not in base_failing]
if not actionable:
return "all_failing_on_base"
dedupe_key = _dedupe_key(head_sha)
if dedupe_key in handled:
return "already_handled"
author_login = await head_commit_author_login(owner=owner, repo=repo, sha=head_sha, token=token)
if (
author_login is not None
and author_login not in INTERNAL_BOT_LOGINS
and (not github_login or author_login.lower() != github_login.lower())
):
logger.info(
"Skipping CI auto-fix on %s/%s#%s: head commit %s authored by human %s",
owner,
repo,
pr_number,
head_sha,
author_login,
)
return "human_commit"
prompt = _build_ci_fix_prompt(
owner=owner,
repo=repo,
pr_number=pr_number,
pr_url=pr_url,
branch=branch,
head_sha=head_sha,
failing_checks=actionable,
)
result = await _dispatch_or_batch(
thread_id,
prompt,
configurable=_run_configurable(
metadata, repo_config={"owner": owner, "name": repo}, pr_number=pr_number
),
reason="ci_failure",
)
if result == "dispatched":
# Only burn an attempt / mark the SHA handled on a real dispatch. A batched
# event is just a nudge to the in-flight run; if that run ends before
# consuming it, leaving the SHA un-handled lets a later webhook or the sweep
# re-dispatch instead of silently dropping the failure.
await _record_attempt(
thread_id, attempts=attempts, handled=handled, dedupe_key=dedupe_key, head_sha=head_sha
)
await post_autofix_status_check(
owner=owner,
repo=repo,
head_sha=head_sha,
token=token,
title=f"Auto-fixing {len(actionable)} failing check(s)",
summary=(
"Open SWE is investigating the failing checks and will push a fix if "
"the cause is clear. Track progress in the linked run."
),
details_url=dashboard_thread_url(thread_id),
)
return result
async def handle_review_feedback(
*,
repo_config: dict[str, str],
pr_number: int,
pr_url: str,
reviewer: str,
body: str,
token: str | None = None,
source: str = "github_review",
) -> str:
"""Auto-respond to a human review comment on an agent-authored PR."""
owner = repo_config.get("owner", "")
repo = repo_config.get("name", "")
if not owner or not repo or not pr_url:
return "missing_repo"
if not await is_review_repo_enabled(owner, repo):
return "repo_not_enabled"
if await is_pr_autofix_disabled(owner, repo, pr_number):
return "pr_disabled"
found = await find_agent_thread_for_pr(pr_url)
if found is None:
return "no_agent_thread"
thread_id, metadata = found
_, _, github_login, user_email = await _thread_autofix_state(metadata)
if not await _user_autofix_enabled(github_login, user_email):
return "autofix_disabled_user"
if token is None:
token = await get_github_app_installation_token()
if not token:
return "no_token"
if not await has_repo_write_permission(owner=owner, repo=repo, username=reviewer, token=token):
logger.info(
"Skipping auto-fix review feedback on %s/%s#%s: %s lacks write access",
owner,
repo,
pr_number,
reviewer or "<unknown>",
)
return "reviewer_no_write_permission"
prompt = _build_review_feedback_prompt(
owner=owner,
repo=repo,
pr_number=pr_number,
pr_url=pr_url,
reviewer=reviewer,
body=body,
)
return await _dispatch_or_batch(
thread_id,
prompt,
configurable=_run_configurable(
metadata, repo_config={"owner": owner, "name": repo}, pr_number=pr_number
),
reason="review_feedback",
detail=f"Reviewer {reviewer or 'unknown'} commented: {body.strip()}"
if body.strip()
else "",
)
async def sweep_open_prs() -> dict[str, int]:
"""Poll open agent-authored PRs and auto-fix failing CI / flag conflicts.
The polling fallback for deployments without reliable CI webhooks, and the
only path that can react to base-branch merge conflicts (GitHub emits no
webhook for those).
"""
counts = {"scanned": 0, "dispatched": 0, "batched": 0, "conflicts": 0}
token = await get_github_app_installation_token()
if not token:
logger.warning("CI monitor sweep: no GitHub App token")
return counts
client = get_client()
try:
threads = await client.threads.search(
metadata={"agent_kind": "agent", "pr_state": "open"}, limit=100
)
except Exception: # noqa: BLE001
logger.warning("CI monitor sweep: thread search failed", exc_info=True)
return counts
for thread in threads or []:
metadata = thread.get("metadata") if isinstance(thread, dict) else None
if not isinstance(metadata, dict):
continue
repo = metadata.get("repo")
pr_number = metadata.get("pr_number")
branch = metadata.get("branch_name")
if not isinstance(repo, dict) or not isinstance(pr_number, int):
continue
owner = repo.get("owner", "")
name = repo.get("name", "")
if not owner or not name:
continue
counts["scanned"] += 1
pr = await fetch_pr(owner=owner, repo=name, pr_number=pr_number, token=token)
if not pr:
continue
head_sha = (pr.get("head") or {}).get("sha", "")
branch = (pr.get("head") or {}).get("ref", "") or (
branch if isinstance(branch, str) else ""
)
if pr.get("mergeable_state") == "dirty":
counts["conflicts"] += 1
await _flag_merge_conflict(
owner=owner,
repo=name,
pr_number=pr_number,
pr_url=pr.get("html_url") or "",
head_sha=head_sha,
token=token,
)
continue
result = await handle_ci_failure(
repo_config={"owner": owner, "name": name},
branch=branch,
head_sha=head_sha,
token=token,
source="ci_monitor",
pr=pr,
)
if result == "dispatched":
counts["dispatched"] += 1
elif result == "batched":
counts["batched"] += 1
logger.info("CI monitor sweep complete: %s", counts)
return counts
async def _flag_merge_conflict(
*, owner: str, repo: str, pr_number: int, pr_url: str, head_sha: str, token: str
) -> None:
"""Ask the agent to rebase a PR that has merge conflicts with its base."""
if await is_pr_autofix_disabled(owner, repo, pr_number):
return
found = await find_agent_thread_for_pr(pr_url)
if found is None:
return
thread_id, metadata = found
_, _, github_login, user_email = await _thread_autofix_state(metadata)
if not await _user_autofix_enabled(github_login, user_email):
return
if metadata.get("autofix_conflict_head") == head_sha:
return
prompt = (
f"The pull request you opened (#{pr_number}, {pr_url}) now has merge "
"conflicts with its base branch. Rebase or merge the base branch into "
"the PR branch, resolve the conflicts carefully, and push. If a "
"conflict resolution is ambiguous, comment on the PR and ask before "
"guessing. Never force-push over commits already on the remote."
)
await _dispatch_or_batch(
thread_id,
prompt,
configurable=_run_configurable(
metadata, repo_config={"owner": owner, "name": repo}, pr_number=pr_number
),
reason="merge_conflict",
)
try:
await get_client().threads.update(
thread_id=thread_id, metadata={"autofix_conflict_head": head_sha}
)
except Exception: # noqa: BLE001
logger.debug("Failed to record conflict head for thread %s", thread_id, exc_info=True)

35
agent/ci_monitor.py Normal file
View file

@ -0,0 +1,35 @@
"""LangGraph entrypoint that polls open agent PRs for CI failures / conflicts.
A fallback for deployments where CI webhooks (``check_run`` / ``workflow_run``)
aren't reliably delivered, and the only path that can react to base-branch
merge conflicts (GitHub emits no webhook for those). Register it on a cron to
sweep periodically; each tick calls :func:`agent.ci_autofix.sweep_open_prs`.
"""
from __future__ import annotations
import logging
from typing import Any, TypedDict
from langgraph.graph import END, START, StateGraph
from langgraph.graph.state import RunnableConfig
from .ci_autofix import sweep_open_prs
logger = logging.getLogger(__name__)
class CIMonitorState(TypedDict, total=False):
result: dict[str, Any]
async def _sweep(_state: CIMonitorState, _config: RunnableConfig) -> dict[str, Any]:
return {"result": await sweep_open_prs()}
def get_ci_monitor(config: RunnableConfig | None = None):
builder = StateGraph(CIMonitorState)
builder.add_node("sweep", _sweep)
builder.add_edge(START, "sweep")
builder.add_edge("sweep", END)
return builder.compile().with_config(config or {})

46
agent/webapp.py
View file

@ -69,6 +69,7 @@ from .utils.github_app import (
get_github_app_installation_token_with_expiry,
)
from .utils.github_checks import complete_review_check_run, create_review_check_run # noqa: F401
from .utils.github_ci import is_failing_ci_payload
from .utils.github_comments import (
OPEN_SWE_TAGS,
build_pr_prompt, # noqa: F401
@ -1368,8 +1369,14 @@ _SUPPORTED_GH_EVENTS = frozenset(
"pull_request_review_comment",
"pull_request_review",
"push",
"check_run",
"check_suite",
"workflow_run",
"status",
]
)
# CI events the auto-fix flow listens to (subset of _SUPPORTED_GH_EVENTS).
_GH_CI_EVENTS = frozenset(["check_run", "check_suite", "workflow_run", "status"])
_SUPPORTED_GH_ISSUE_ACTIONS = frozenset(["edited", "opened", "reopened"])
_SUPPORTED_GH_PULL_REQUEST_ACTIONS = frozenset(
[
@ -1892,6 +1899,15 @@ async def github_webhook(request: Request, background_tasks: BackgroundTasks) ->
background_tasks.add_task(process_github_push_event, payload)
return {"status": "accepted", "message": "Processing GitHub push for reviewer watch"}
if event_type in _GH_CI_EVENTS:
if not is_failing_ci_payload(payload, event_type):
return {"status": "ignored", "reason": "CI event is not a completed failure"}
if not await _is_repo_enabled_for_review(webhook_repo_config):
return {"status": "ignored", "reason": "Repository not enabled for review"}
logger.info("Accepted GitHub %s webhook, scheduling CI auto-fix evaluation", event_type)
background_tasks.add_task(process_github_ci_event, payload, event_type)
return {"status": "accepted", "message": f"Processing GitHub {event_type} for auto-fix"}
if not _is_repo_allowed(webhook_repo_config):
logger.debug(
"Rejecting GitHub webhook: repo '%s/%s' not in allowlist",
@ -1936,6 +1952,22 @@ async def github_webhook(request: Request, background_tasks: BackgroundTasks) ->
comment = payload.get("comment") or payload.get("review", {})
comment_body = (comment.get("body") or "") if comment else ""
is_pr_related_comment = is_pull_request_comment or event_type in {
"pull_request_review_comment",
"pull_request_review",
}
autofix_command = _parse_autofix_command(comment_body)
if autofix_command is not None and is_pr_related_comment:
if not await _is_repo_enabled_for_review(webhook_repo_config):
return {"status": "ignored", "reason": "Repository not enabled for review"}
gate_rejection = await _enforce_public_repo_org_gate(payload, event_type)
if gate_rejection is not None:
return gate_rejection
background_tasks.add_task(
process_github_autofix_command, payload, event_type, disabled=autofix_command
)
return {"status": "accepted", "message": "Processing auto-fix toggle"}
if (
event_type == "pull_request_review_comment"
and _review_comment_reply_parent_id(payload) is not None
@ -1949,6 +1981,14 @@ async def github_webhook(request: Request, background_tasks: BackgroundTasks) ->
return {"status": "accepted", "message": "Processing review finding reply"}
if not any(tag in comment_body.lower() for tag in OPEN_SWE_TAGS):
if _is_actionable_review_payload(payload, event_type) and await _is_repo_enabled_for_review(
webhook_repo_config
):
gate_rejection = await _enforce_public_repo_org_gate(payload, event_type)
if gate_rejection is not None:
return gate_rejection
background_tasks.add_task(process_github_autofix_review, payload, event_type)
return {"status": "accepted", "message": "Processing auto-fix review feedback"}
logger.debug(
"Ignoring GitHub %s%s that does not mention @openswe or @open-swe",
event_type,
@ -1981,10 +2021,16 @@ async def github_webhook(request: Request, background_tasks: BackgroundTasks) ->
# webapp.process_github_issue, webapp.build_github_issue_prompt, etc.) keep working.
from .webhooks.github import ( # noqa: E402,F401
_dispatch_first_review_from_pr_payload,
_is_actionable_review_payload,
_parse_autofix_command,
_pr_ref_from_comment_payload,
build_github_issue_followup_prompt,
build_github_issue_prompt,
build_github_issue_update_prompt,
build_github_pr_review_prompt,
process_github_autofix_command,
process_github_autofix_review,
process_github_ci_event,
process_github_issue,
process_github_pr_close,
process_github_pr_comment,

178
agent/webhooks/github.py
View file

@ -4,12 +4,20 @@ Helpers and constants stay in webapp.py; they are accessed through the module
object (``webapp.X``) so tests that monkeypatch them keep working.
"""
import re
import uuid
from typing import Any
from agent import webapp
from ..ci_autofix import handle_ci_failure, handle_review_feedback
from ..dashboard.autofix_state import set_pr_autofix_disabled
from ..reviewer_findings import FindingInteraction, ReviewerPRMeta, ReviewerSlackThread
from ..utils.github_ci import (
branch_from_check_payload,
head_sha_from_check_payload,
is_failing_ci_payload,
)
from ..utils.github_comments import GitHubAuthError
from ..utils.slack import GitHubPrRef
@ -1045,3 +1053,173 @@ async def process_github_issue(payload: dict[str, Any], event_type: str) -> None
client=langgraph_client,
)
webapp.logger.info("LangGraph run dispatched for thread %s from GitHub issue", thread_id)
# ---- CI auto-fix handlers (re-wired onto the modular dispatch model) ----
async def process_github_ci_event(payload: dict[str, Any], event_type: str) -> None:
"""Auto-fix failing CI on an agent-authored PR from a CI webhook."""
if not is_failing_ci_payload(payload, event_type):
return
repo = payload.get("repository", {})
repo_config = {
"owner": repo.get("owner", {}).get("login", "") or repo.get("owner", {}).get("name", ""),
"name": repo.get("name", ""),
}
if not repo_config["owner"] or not repo_config["name"]:
return
branch = branch_from_check_payload(payload, event_type)
head_sha = head_sha_from_check_payload(payload, event_type)
if not head_sha:
return
result = await handle_ci_failure(
repo_config=repo_config,
branch=branch,
head_sha=head_sha,
source="github_ci",
)
webapp.logger.info(
"CI auto-fix for %s/%s@%s (%s): %s",
repo_config["owner"],
repo_config["name"],
head_sha,
event_type,
result,
)
_AUTOFIX_COMMAND_RE = re.compile(r"autofix\s+(on|off)\b", re.IGNORECASE)
def _parse_autofix_command(comment_body: str) -> bool | None:
"""Return True (disable) / False (enable) for an ``@open-swe autofix on|off`` command.
Returns ``None`` when the comment isn't an auto-fix command. Requires an
Open SWE mention so a passing reference to "autofix off" doesn't toggle it.
"""
if not any(tag in comment_body.lower() for tag in webapp.OPEN_SWE_TAGS):
return None
match = _AUTOFIX_COMMAND_RE.search(comment_body)
if not match:
return None
return match.group(1).lower() == "off"
def _pr_ref_from_comment_payload(payload: dict[str, Any], event_type: str) -> dict[str, Any] | None:
"""Extract ``{owner, name, number, url}`` for the PR a comment belongs to."""
repo = payload.get("repository", {})
owner = repo.get("owner", {}).get("login", "")
name = repo.get("name", "")
if event_type == "issue_comment":
issue = payload.get("issue", {})
number = issue.get("number")
pr = issue.get("pull_request") or {}
url = pr.get("html_url") or issue.get("html_url") or ""
else:
pr = payload.get("pull_request", {})
number = pr.get("number")
url = pr.get("html_url") or ""
if not owner or not name or not isinstance(number, int):
return None
return {"owner": owner, "name": name, "number": number, "url": url}
async def process_github_autofix_command(
payload: dict[str, Any], event_type: str, *, disabled: bool
) -> None:
"""Persist an ``@open-swe autofix on|off`` per-PR toggle and acknowledge it."""
ref = _pr_ref_from_comment_payload(payload, event_type)
if ref is None:
return
await set_pr_autofix_disabled(ref["owner"], ref["name"], ref["number"], disabled)
webapp.logger.info(
"Auto-fix %s for %s/%s#%s via comment",
"disabled" if disabled else "enabled",
ref["owner"],
ref["name"],
ref["number"],
)
comment = payload.get("comment") or {}
comment_id = comment.get("id")
if not isinstance(comment_id, int):
return
token = await webapp.get_github_app_installation_token()
if not token:
return
try:
await webapp.react_to_github_comment(
{"owner": ref["owner"], "name": ref["name"]},
comment_id,
event_type=event_type,
token=token,
pull_number=ref["number"],
node_id=comment.get("node_id"),
)
except Exception: # noqa: BLE001
webapp.logger.debug("Failed to react to auto-fix command comment", exc_info=True)
# GitHub author_association values that imply at least repo-member trust. Used
# as a cheap first gate before the no-mention auto-fix-on-review path; a real
# write-permission check follows in process_github_autofix_review.
_TRUSTED_REVIEW_ASSOCIATIONS = frozenset(["OWNER", "MEMBER", "COLLABORATOR"])
def _is_actionable_review_payload(payload: dict[str, Any], event_type: str) -> bool:
"""Return whether a review event is trusted human feedback worth auto-responding to.
Approvals, the agent's own bot comments, and feedback from non-trusted
authors (read/triage/outside users) are not actionable — auto-fix-on-review
dispatches a write-capable run, so only repo collaborators/members/owners
may trigger it without an explicit ``@open-swe`` mention.
"""
action = payload.get("action", "")
if event_type == "pull_request_review_comment":
if action != "created":
return False
node = payload.get("comment") or {}
elif event_type == "pull_request_review":
if action != "submitted":
return False
node = payload.get("review") or {}
if node.get("state") not in {"changes_requested", "commented"}:
return False
else:
return False
if not isinstance(node, dict):
return False
reviewer = (node.get("user") or {}).get("login", "")
if reviewer in webapp.INTERNAL_BOT_LOGINS:
return False
if node.get("author_association") not in _TRUSTED_REVIEW_ASSOCIATIONS:
return False
body = node.get("body") or ""
return bool(body.strip())
async def process_github_autofix_review(payload: dict[str, Any], event_type: str) -> None:
"""Auto-respond to a human review/review-comment on an agent-authored PR."""
ref = _pr_ref_from_comment_payload(payload, event_type)
if ref is None:
return
comment = payload.get("comment") or payload.get("review", {})
reviewer = (comment.get("user") or {}).get("login", "") if isinstance(comment, dict) else ""
body = (comment.get("body") or "") if isinstance(comment, dict) else ""
if not body.strip() or reviewer in webapp.INTERNAL_BOT_LOGINS:
return
result = await handle_review_feedback(
repo_config={"owner": ref["owner"], "name": ref["name"]},
pr_number=ref["number"],
pr_url=ref["url"],
reviewer=reviewer,
body=body,
source="github_review",
)
webapp.logger.info(
"Auto-fix review feedback for %s/%s#%s: %s",
ref["owner"],
ref["name"],
ref["number"],
result,
)

3
langgraph.json
View file

@ -7,7 +7,8 @@
"reviewer": "agent.reviewer:traced_reviewer_agent",
"analyzer": "agent.analyzer:traced_analyzer",
"chat": "agent.chat:traced_chat_agent",
"scheduler": "agent.scheduler:get_scheduler"
"scheduler": "agent.scheduler:get_scheduler",
"ci_monitor": "agent.ci_monitor:get_ci_monitor"
},
"dependencies": [
"."

189
tests/test_autofix_webhook.py Normal file
View file

@ -0,0 +1,189 @@
"""Unit tests for the auto-fix webhook helpers in agent.webapp."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
import pytest
from agent import webapp
from agent.webhooks import github as webhooks_github
def test_parse_autofix_command() -> None:
assert webapp._parse_autofix_command("@open-swe autofix off") is True
assert webapp._parse_autofix_command("@open-swe autofix on") is False
assert webapp._parse_autofix_command("@openswe please autofix off now") is True
# Missing the mention -> not a command.
assert webapp._parse_autofix_command("autofix off") is None
# Mention but no command keyword.
assert webapp._parse_autofix_command("@open-swe fix this") is None
def test_pr_ref_from_issue_comment() -> None:
payload = {
"repository": {"owner": {"login": "o"}, "name": "r"},
"issue": {
"number": 7,
"pull_request": {"html_url": "https://github.com/o/r/pull/7"},
},
}
ref = webapp._pr_ref_from_comment_payload(payload, "issue_comment")
assert ref == {"owner": "o", "name": "r", "number": 7, "url": "https://github.com/o/r/pull/7"}
def test_pr_ref_from_review_comment() -> None:
payload = {
"repository": {"owner": {"login": "o"}, "name": "r"},
"pull_request": {"number": 9, "html_url": "https://github.com/o/r/pull/9"},
}
ref = webapp._pr_ref_from_comment_payload(payload, "pull_request_review_comment")
assert ref["number"] == 9
def test_pr_ref_none_when_not_a_pr() -> None:
payload = {"repository": {"owner": {"login": "o"}, "name": "r"}, "issue": {"number": 3}}
# issue without pull_request still yields a ref (number present); url empty.
ref = webapp._pr_ref_from_comment_payload(payload, "issue_comment")
assert ref["url"] == ""
def test_is_actionable_review_payload() -> None:
assert webapp._is_actionable_review_payload(
{
"action": "submitted",
"review": {
"state": "changes_requested",
"body": "fix this",
"user": {"login": "a"},
"author_association": "MEMBER",
},
},
"pull_request_review",
)
# Approval is not actionable.
assert not webapp._is_actionable_review_payload(
{
"action": "submitted",
"review": {
"state": "approved",
"body": "lgtm",
"user": {"login": "a"},
"author_association": "MEMBER",
},
},
"pull_request_review",
)
# Bot author is not actionable.
assert not webapp._is_actionable_review_payload(
{
"action": "created",
"comment": {
"body": "x",
"user": {"login": "open-swe[bot]"},
"author_association": "MEMBER",
},
},
"pull_request_review_comment",
)
# Untrusted author (read/triage/outside) is not actionable.
assert not webapp._is_actionable_review_payload(
{
"action": "created",
"comment": {
"body": "inject malicious code",
"user": {"login": "attacker"},
"author_association": "NONE",
},
},
"pull_request_review_comment",
)
# Empty body is not actionable.
assert not webapp._is_actionable_review_payload(
{
"action": "created",
"comment": {"body": " ", "user": {"login": "a"}, "author_association": "OWNER"},
},
"pull_request_review_comment",
)
@pytest.mark.asyncio
async def test_process_github_ci_event_dispatches() -> None:
payload = {
"repository": {"owner": {"login": "o"}, "name": "r"},
"check_run": {
"status": "completed",
"conclusion": "failure",
"head_sha": "sha1",
"check_suite": {"head_branch": "feat"},
},
}
handle = AsyncMock(return_value="dispatched")
with patch.object(webhooks_github, "handle_ci_failure", handle):
await webapp.process_github_ci_event(payload, "check_run")
handle.assert_awaited_once()
kwargs = handle.await_args.kwargs
assert kwargs["repo_config"] == {"owner": "o", "name": "r"}
assert kwargs["head_sha"] == "sha1"
assert kwargs["branch"] == "feat"
@pytest.mark.asyncio
async def test_process_github_ci_event_ignores_success() -> None:
payload = {
"repository": {"owner": {"login": "o"}, "name": "r"},
"check_run": {"status": "completed", "conclusion": "success", "head_sha": "s"},
}
handle = AsyncMock()
with patch.object(webhooks_github, "handle_ci_failure", handle):
await webapp.process_github_ci_event(payload, "check_run")
handle.assert_not_called()
@pytest.mark.asyncio
async def test_process_autofix_command_sets_flag() -> None:
payload = {
"repository": {"owner": {"login": "o"}, "name": "r"},
"issue": {"number": 7, "pull_request": {"html_url": "u"}},
"comment": {"id": 1, "node_id": "n"},
}
setter = AsyncMock()
with (
patch.object(webhooks_github, "set_pr_autofix_disabled", setter),
patch.object(webapp, "get_github_app_installation_token", AsyncMock(return_value="")),
):
await webapp.process_github_autofix_command(payload, "issue_comment", disabled=True)
setter.assert_awaited_once_with("o", "r", 7, True)
@pytest.mark.asyncio
async def test_autofix_review_dispatches_for_writer() -> None:
payload = {
"repository": {"owner": {"login": "o"}, "name": "r"},
"pull_request": {"number": 9, "html_url": "https://github.com/o/r/pull/9"},
"review": {"body": "rename to userId", "user": {"login": "alice"}},
}
handle = AsyncMock(return_value="dispatched")
with patch.object(webhooks_github, "handle_review_feedback", handle):
await webapp.process_github_autofix_review(payload, "pull_request_review")
handle.assert_awaited_once()
@pytest.mark.asyncio
async def test_autofix_review_delegates_permission_check_to_core() -> None:
payload = {
"repository": {"owner": {"login": "o"}, "name": "r"},
"pull_request": {"number": 9, "html_url": "https://github.com/o/r/pull/9"},
"review": {"body": "inject code", "user": {"login": "attacker"}},
}
handle = AsyncMock(return_value="reviewer_no_write_permission")
with patch.object(webhooks_github, "handle_review_feedback", handle):
await webapp.process_github_autofix_review(payload, "pull_request_review")
handle.assert_awaited_once()
def test_ci_events_supported() -> None:
for event in ("check_run", "check_suite", "workflow_run", "status"):
assert event in webapp._SUPPORTED_GH_EVENTS
assert event in webapp._GH_CI_EVENTS

253
tests/test_ci_autofix.py Normal file
View file

@ -0,0 +1,253 @@
"""Unit tests for the CI auto-fix orchestration core."""
from __future__ import annotations
from typing import Any
from unittest.mock import AsyncMock, MagicMock
import pytest
from agent import ci_autofix
_PR = {
"number": 5,
"html_url": "https://github.com/o/r/pull/5",
"base": {"sha": "base"},
"head": {"ref": "feat", "sha": "head1"},
}
@pytest.fixture
def happy(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]:
"""Patch every dependency of handle_ci_failure to a happy-path default."""
runs_create = AsyncMock()
store_put = AsyncMock()
lg_client = MagicMock()
lg_client.runs.create = runs_create
lg_client.store.get_item = AsyncMock(return_value=None)
lg_client.store.put_item = store_put
threads_update = AsyncMock()
store_client = MagicMock()
store_client.threads.update = threads_update
# Auto-fix runs now dispatch through the durable dispatch_agent_run contract
# rather than a raw runs.create; assert against that.
dispatch_run = AsyncMock(return_value={"run_id": "r1"})
mocks: dict[str, Any] = {
"runs_create": dispatch_run,
"threads_update": threads_update,
"status_check": AsyncMock(return_value=True),
"store_put": store_put,
}
monkeypatch.setattr(ci_autofix, "_user_autofix_enabled", AsyncMock(return_value=True))
monkeypatch.setattr(ci_autofix, "is_review_repo_enabled", AsyncMock(return_value=True))
monkeypatch.setattr(
ci_autofix, "get_github_app_installation_token", AsyncMock(return_value="tok")
)
monkeypatch.setattr(ci_autofix, "is_pr_autofix_disabled", AsyncMock(return_value=False))
monkeypatch.setattr(
ci_autofix,
"find_agent_thread_for_pr",
AsyncMock(return_value=("t1", {"github_login": "alice", "autofix_attempts": 0})),
)
monkeypatch.setattr(
ci_autofix,
"list_failing_check_runs",
AsyncMock(return_value=[{"name": "lint", "conclusion": "failure", "details_url": ""}]),
)
monkeypatch.setattr(ci_autofix, "list_failing_statuses", AsyncMock(return_value=[]))
monkeypatch.setattr(ci_autofix, "names_failing_on_base", AsyncMock(return_value=set()))
monkeypatch.setattr(
ci_autofix, "head_commit_author_login", AsyncMock(return_value="open-swe[bot]")
)
monkeypatch.setattr(ci_autofix, "get_thread_active_status", AsyncMock(return_value=False))
monkeypatch.setattr(ci_autofix, "post_autofix_status_check", mocks["status_check"])
monkeypatch.setattr(ci_autofix, "langgraph_client", lambda: lg_client)
monkeypatch.setattr(ci_autofix, "dispatch_agent_run", mocks["runs_create"])
monkeypatch.setattr(ci_autofix, "get_client", lambda: store_client)
return mocks
async def _run(**overrides: Any) -> str:
kwargs: dict[str, Any] = {
"repo_config": {"owner": "o", "name": "r"},
"branch": "feat",
"head_sha": "head1",
"pr": _PR,
}
kwargs.update(overrides)
return await ci_autofix.handle_ci_failure(**kwargs)
@pytest.mark.asyncio
async def test_dispatch_happy_path(happy: dict[str, Any]) -> None:
result = await _run()
assert result == "dispatched"
happy["runs_create"].assert_awaited_once()
happy["threads_update"].assert_awaited()
happy["status_check"].assert_awaited()
@pytest.mark.asyncio
async def test_batches_when_thread_busy(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "get_thread_active_status", AsyncMock(return_value=True))
result = await _run()
assert result == "batched"
happy["store_put"].assert_awaited()
happy["runs_create"].assert_not_called()
# A batched event must not burn an attempt or mark the SHA handled, so a later
# webhook/sweep can still dispatch if the in-flight run never consumes it.
happy["threads_update"].assert_not_awaited()
@pytest.mark.asyncio
async def test_skip_user_disabled(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "_user_autofix_enabled", AsyncMock(return_value=False))
assert await _run() == "autofix_disabled_user"
@pytest.mark.asyncio
async def test_skip_repo_not_enabled(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "is_review_repo_enabled", AsyncMock(return_value=False))
assert await _run() == "repo_not_enabled"
@pytest.mark.asyncio
async def test_skip_pr_disabled(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "is_pr_autofix_disabled", AsyncMock(return_value=True))
assert await _run() == "pr_disabled"
@pytest.mark.asyncio
async def test_skip_no_agent_thread(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "find_agent_thread_for_pr", AsyncMock(return_value=None))
assert await _run() == "no_agent_thread"
@pytest.mark.asyncio
async def test_skip_max_attempts(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(
ci_autofix,
"find_agent_thread_for_pr",
AsyncMock(
return_value=(
"t1",
{"github_login": "alice", "autofix_attempts": ci_autofix.MAX_AUTOFIX_ATTEMPTS},
)
),
)
assert await _run() == "max_attempts"
happy["status_check"].assert_awaited()
happy["runs_create"].assert_not_called()
@pytest.mark.asyncio
async def test_skip_all_failing_on_base(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "names_failing_on_base", AsyncMock(return_value={"lint"}))
assert await _run() == "all_failing_on_base"
@pytest.mark.asyncio
async def test_skip_already_handled(happy: dict[str, Any], monkeypatch) -> None:
key = ci_autofix._dedupe_key("head1")
monkeypatch.setattr(
ci_autofix,
"find_agent_thread_for_pr",
AsyncMock(return_value=("t1", {"github_login": "alice", "autofix_handled": [key]})),
)
assert await _run() == "already_handled"
@pytest.mark.asyncio
async def test_skip_human_commit(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "head_commit_author_login", AsyncMock(return_value="mallory"))
assert await _run() == "human_commit"
happy["runs_create"].assert_not_called()
@pytest.mark.asyncio
async def test_no_failing_checks(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "list_failing_check_runs", AsyncMock(return_value=[]))
assert await _run(failing_checks=None) == "no_failing_checks"
@pytest.mark.asyncio
async def test_ci_read_failed(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "list_failing_check_runs", AsyncMock(return_value=None))
monkeypatch.setattr(ci_autofix, "list_failing_statuses", AsyncMock(return_value=None))
assert await _run(failing_checks=None) == "ci_read_failed"
@pytest.mark.asyncio
async def test_review_feedback_skips_user_disabled(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "_user_autofix_enabled", AsyncMock(return_value=False))
assert (
await ci_autofix.handle_review_feedback(
repo_config={"owner": "o", "name": "r"},
pr_number=5,
pr_url="https://github.com/o/r/pull/5",
reviewer="alice",
body="fix this",
)
== "autofix_disabled_user"
)
happy["runs_create"].assert_not_called()
@pytest.mark.asyncio
async def test_review_feedback_batches_when_thread_busy(happy: dict[str, Any], monkeypatch) -> None:
monkeypatch.setattr(ci_autofix, "has_repo_write_permission", AsyncMock(return_value=True))
monkeypatch.setattr(ci_autofix, "get_thread_active_status", AsyncMock(return_value=True))
result = await ci_autofix.handle_review_feedback(
repo_config={"owner": "o", "name": "r"},
pr_number=5,
pr_url="https://github.com/o/r/pull/5",
reviewer="alice",
body="fix this",
)
assert result == "batched"
happy["runs_create"].assert_not_called()
@pytest.mark.asyncio
async def test_review_feedback_checks_write_permission_after_user_gate(
happy: dict[str, Any], monkeypatch
) -> None:
permission = AsyncMock(return_value=False)
monkeypatch.setattr(ci_autofix, "has_repo_write_permission", permission)
result = await ci_autofix.handle_review_feedback(
repo_config={"owner": "o", "name": "r"},
pr_number=5,
pr_url="https://github.com/o/r/pull/5",
reviewer="alice",
body="fix this",
)
assert result == "reviewer_no_write_permission"
permission.assert_awaited_once()
happy["runs_create"].assert_not_called()
@pytest.mark.asyncio
async def test_find_agent_thread_picks_agent_skips_reviewer(monkeypatch) -> None:
client = MagicMock()
client.threads.search = AsyncMock(
return_value=[
{"thread_id": "rev", "metadata": {"kind": "reviewer", "agent_kind": "agent"}},
{"thread_id": "ag", "metadata": {"agent_kind": "agent"}},
]
)
monkeypatch.setattr(ci_autofix, "get_client", lambda: client)
found = await ci_autofix.find_agent_thread_for_pr("https://github.com/o/r/pull/5")
assert found is not None
assert found[0] == "ag"
@pytest.mark.asyncio
async def test_find_agent_thread_none_when_only_reviewer(monkeypatch) -> None:
client = MagicMock()
client.threads.search = AsyncMock(
return_value=[{"thread_id": "rev", "metadata": {"kind": "reviewer"}}]
)
monkeypatch.setattr(ci_autofix, "get_client", lambda: client)
assert await ci_autofix.find_agent_thread_for_pr("u") is None