diff --git a/agent/ci_autofix.py b/agent/ci_autofix.py new file mode 100644 index 00000000..721aa99b --- /dev/null +++ b/agent/ci_autofix.py @@ -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 " + " --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 "", + ) + 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) diff --git a/agent/ci_monitor.py b/agent/ci_monitor.py new file mode 100644 index 00000000..4fa70692 --- /dev/null +++ b/agent/ci_monitor.py @@ -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 {}) diff --git a/agent/webapp.py b/agent/webapp.py index faab6475..fdb11950 100644 --- a/agent/webapp.py +++ b/agent/webapp.py @@ -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, diff --git a/agent/webhooks/github.py b/agent/webhooks/github.py index a206c205..73ce5e7f 100644 --- a/agent/webhooks/github.py +++ b/agent/webhooks/github.py @@ -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, + ) diff --git a/langgraph.json b/langgraph.json index 450dacf3..07811c48 100644 --- a/langgraph.json +++ b/langgraph.json @@ -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": [ "." diff --git a/tests/test_autofix_webhook.py b/tests/test_autofix_webhook.py new file mode 100644 index 00000000..ee26e93f --- /dev/null +++ b/tests/test_autofix_webhook.py @@ -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 diff --git a/tests/test_ci_autofix.py b/tests/test_ci_autofix.py new file mode 100644 index 00000000..fd204161 --- /dev/null +++ b/tests/test_ci_autofix.py @@ -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