"""Tool: ``publish_review``. Post the findings list to GitHub as a PR Review.""" from __future__ import annotations import logging from typing import Any from langgraph.config import get_config from ..dashboard.team_settings import get_team_review_trace_links_enabled from ..review.diff import compute_diff_line_set, fetch_pr_diff, is_range_in_diff from ..review.findings import ( REVIEW_FINDING_CAP, REVIEWER_EVAL_PUBLICATION_KEY, SEVERITY_ORDER, Finding, ReviewerThreadMissingError, Severity, _coerce_findings_list, _coerce_surface, _finding_mutation_lock, _get_thread_metadata_strict, filter_findings_for_publish, get_thread_id_from_runtime, get_thread_last_reviewed_sha, get_thread_metadata, get_thread_pr_meta, get_thread_slack_ref, replace_findings, resolve_review_head_sha, set_reviewer_thread_metadata, thread_missing_tool_result, ) from ..review.findings import ( list_findings as list_findings_async, ) from ..review.publish import ( clear_review_started_comment, dismiss_pull_request_review, fetch_pr_review_threads, fetch_pull_request_head_sha, fetch_review_comments, fetch_review_thread_id_for_comment, open_swe_review_exists, parse_review_comment_marker, post_pull_request_review, render_inline_comment_payload, render_resolution_comment, render_review_body, reply_to_review_comment, resolve_review_thread, settle_review_check_run, update_pull_request_review_body, ) from ..review.reconcile import reconcile_findings_with_review_threads from ..utils.dashboard_links import dashboard_review_url from ..utils.github_checks import review_check_conclusion from ..utils.github_org_membership import INTERNAL_BOT_LOGINS from ..utils.github_token import ( GitHubAuthError, get_github_token, invalidate_cached_github_token, ) from ..utils.langsmith import get_langsmith_trace_url from ..utils.slack import post_slack_thread_reply from ..utils.tracing import REVIEW_TRACING_PROJECT logger = logging.getLogger(__name__) _VERDICT_EVENTS = {"approve": "APPROVE", "request_changes": "REQUEST_CHANGES"} # Map the GitHub review event we POST to the review "state" GitHub reports back # on the created review object, so we can confirm the verdict actually landed. _EVENT_TO_STATE = {"APPROVE": "APPROVED", "REQUEST_CHANGES": "CHANGES_REQUESTED"} async def publish_review( severity_threshold: str = "medium", verdict: str | None = None, ) -> dict[str, Any]: """Post all current findings to the PR as a GitHub Review. Call this once at the end of a review run, after you have finished adding findings (and, on a re-review, after marking resolved findings via ``update_finding``). The tool posts one GitHub PR Review for eligible inline findings, records the GitHub comment/thread IDs for future re-reviews, resolves GitHub threads for findings now marked resolved, and advances the reviewer thread's ``last_reviewed_sha``. On a re-review with no new findings to surface, it skips posting a new GitHub Review but still resolves fixed threads and updates reviewer state. Args: severity_threshold: Lowest severity to surface as inline GitHub comments (default ``medium``). Lower-severity findings stay in state and are mentioned in the review summary with a link to the web app, but are not posted as inline PR comments. verdict: Optional review verdict — ``"approve"`` or ``"request_changes"``. Explicitly requested verdicts are honored subject to the safety checks below. Automatically authorized verdicts must agree with authoritative finding state: approve requires no open findings and request_changes requires at least one. A downgraded verdict is posted as a comment review and carries ``verdict_ignored: true``. Returns: Dictionary with ``success``, ``review_id``, ``surfaced_count``, ``hidden_count``, ``resolved_thread_count``, and sometimes ``unresolvable_findings``, plus the flags below. ``success: true`` alone does NOT mean a GitHub Review was posted — check the flags: - ``skipped_empty_re_review: true`` (with ``review_id: null``): an empty re-review was deliberately skipped. No GitHub Review was created; the call was a valid no-op. Do not describe the review as published/posted/submitted. - ``dry_run: true`` (with ``review_id: null``): eval/benchmark mode — the publish was simulated and nothing was posted to GitHub. Do not claim publication. Only a numeric ``review_id`` (with neither flag set) confirms a real GitHub Review was created. When ``verdict`` was passed, the result also carries ``verdict_submitted`` (GitHub confirmed the requested APPROVE/ REQUEST_CHANGES state) or ``verdict_ignored`` + ``verdict_ignored_reason`` (``"verdict_not_requested"``, ``"approve_with_open_findings"``, ``"request_changes_without_open_findings"``, ``"self_review"``, ``"head_moved"`` — the reviewed commit is no longer the PR head — ``"head_check_failed"``, ``"author_unknown"``, or ``"github_state_mismatch"``). """ if severity_threshold not in {"low", "medium", "high", "critical"}: return {"success": False, "error": f"Invalid severity_threshold: {severity_threshold}"} if verdict is not None and verdict not in _VERDICT_EVENTS: return { "success": False, "error": ( f"Invalid verdict: {verdict!r}. Use 'approve', 'request_changes', " "or omit the parameter." ), } config = get_config() raw_configurable = config.get("configurable", {}) if isinstance(config, dict) else {} configurable = raw_configurable if isinstance(raw_configurable, dict) else {} repo_config = configurable.get("repo") pr_number = configurable.get("pr_number") head_sha = configurable.get("head_sha") is_re_review = bool(configurable.get("re_review")) if ( not isinstance(repo_config, dict) or not repo_config.get("owner") or not repo_config.get("name") ): return {"success": False, "error": "Missing repo info in run config"} if not isinstance(pr_number, int): return {"success": False, "error": "Missing pr_number in run config"} if not isinstance(head_sha, str) or not head_sha: return {"success": False, "error": "Missing head_sha in run config"} if _is_reviewer_eval_mode(configurable): eval_threshold = configurable.get("reviewer_eval_severity_threshold") if isinstance(eval_threshold, str) and eval_threshold in { "low", "medium", "high", "critical", }: severity_threshold = eval_threshold eval_cap = configurable.get("reviewer_eval_cap") if not isinstance(eval_cap, int) or isinstance(eval_cap, bool) or eval_cap < 0: eval_cap = REVIEW_FINDING_CAP try: return await _publish_review_eval_dry_run_async( head_sha=head_sha, severity_threshold=_cast_severity(severity_threshold), cap=eval_cap, ) except ReviewerThreadMissingError as exc: return thread_missing_tool_result(exc) token = get_github_token() if not token: return {"success": False, "error": "No GitHub token available"} if configurable.get("verdict_requested") is True: verdict_authorization = "requested" elif configurable.get("verdict_authorized") is True: verdict_authorization = "consistent" else: verdict_authorization = "none" try: result = await _publish_review_async( owner=str(repo_config["owner"]), repo=str(repo_config["name"]), pr_number=pr_number, head_sha=head_sha, token=token, severity_threshold=_cast_severity(severity_threshold), cap=REVIEW_FINDING_CAP, is_re_review=is_re_review, langgraph_run_id=_current_run_id(config), trace_link_config_override=configurable.get("review_trace_link_enabled"), verdict=verdict, verdict_requester=str(configurable.get("github_login") or ""), verdict_authorization=verdict_authorization, ) return result except ReviewerThreadMissingError as exc: return thread_missing_tool_result(exc) except GitHubAuthError as exc: thread_id = get_thread_id_from_runtime() if thread_id: await invalidate_cached_github_token(thread_id) return { "success": False, "error": ( "GitHub returned 401 — the cached OAuth token is invalid or revoked. " "Please re-authenticate and trigger the review again." ), "auth_error": str(exc), } def _cast_severity(value: str) -> Severity: return value # type: ignore[return-value] async def _resolve_review_trace_url(thread_id: str, config_override: object) -> str | None: if config_override is False: return None if not await get_team_review_trace_links_enabled(): return None if not thread_id: return None return get_langsmith_trace_url(thread_id, project_name=REVIEW_TRACING_PROJECT) def _is_reviewer_eval_mode(configurable: dict[str, Any]) -> bool: return configurable.get("reviewer_eval") is True or configurable.get("eval") is True async def _publish_review_eval_dry_run_async( *, head_sha: str, severity_threshold: Severity, cap: int, ) -> dict[str, Any]: """Simulate publish_review for benchmark runs without posting to GitHub.""" thread_id = get_thread_id_from_runtime() findings = await list_findings_async(thread_id) unpublished_findings = [f for f in findings if not _has_publication_identity(f)] open_unpublished = [f for f in unpublished_findings if f.get("status", "open") == "open"] # Out-of-diff findings are disabled: only in-diff findings are surfaced. in_diff_unpublished = [f for f in unpublished_findings if f.get("in_diff", True)] eligible = filter_findings_for_publish( in_diff_unpublished, severity_threshold=severity_threshold, cap=cap, ) eligible_with_payload = [ (finding, payload) for finding in eligible if (payload := render_inline_comment_payload(finding)) is not None ] finding_ids = [ finding["id"] for finding, _payload in eligible_with_payload if isinstance(finding.get("id"), str) ] publication = { "finding_ids": finding_ids, "severity_threshold": severity_threshold, "cap": cap, } await set_reviewer_thread_metadata( thread_id, last_reviewed_sha=head_sha, extra={REVIEWER_EVAL_PUBLICATION_KEY: publication}, ) return { "success": True, "dry_run": True, "review_id": None, "surfaced_count": len(eligible_with_payload), "hidden_count": max(len(open_unpublished) - len(eligible_with_payload), 0), "resolved_thread_count": 0, } async def _publish_review_async( *, owner: str, repo: str, pr_number: int, head_sha: str, token: str, severity_threshold: Severity, cap: int, is_re_review: bool, langgraph_run_id: str | None = None, trace_link_config_override: object = None, verdict: str | None = None, verdict_requester: str = "", verdict_authorization: str = "requested", ) -> dict[str, Any]: thread_id = get_thread_id_from_runtime() verdict_ignored_reason: str | None = None verdict_attempted = verdict is not None reviewed_head_sha = head_sha # The run config's head_sha is frozen at run creation; a push that arrived # mid-run updated the live head in thread metadata. Prefer that so the # review anchors to (and last_reviewed_sha advances to) the commit actually # reviewed, not the stale one this run was created for. head_sha = await resolve_review_head_sha(thread_id, {"head_sha": head_sha}) findings = await _backfill_findings_from_pr_threads( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, token=token, ) review_trace_url = await _resolve_review_trace_url(thread_id, trace_link_config_override) review_ui_url = dashboard_review_url(owner, repo, pr_number) # Re-reviews only post NEW findings. Anything with a github_review_comment_id # already lives on GitHub from a prior publish — reposting would create # duplicate inline comments and break the resolve-on-fix flow (only # whichever duplicate id we'd cache last would resolve later). unpublished_findings = [ f for f in findings if not isinstance(f.get("github_review_comment_id"), int) ] if is_re_review: unpublished_findings = [ f for f in unpublished_findings if f.get("first_seen_sha") == head_sha ] open_unpublished = [f for f in unpublished_findings if f.get("status", "open") == "open"] # In-diff findings become inline comments. Out-of-diff findings are disabled: # they are never surfaced on the PR (any legacy in-state ones are treated as # hidden). in_diff_unpublished = [f for f in unpublished_findings if f.get("in_diff", True)] eligible = filter_findings_for_publish( in_diff_unpublished, severity_threshold=severity_threshold, cap=cap ) severity_rank = SEVERITY_ORDER[severity_threshold] eligible_ids = {f.get("id") for f in eligible} additional_findings_count = sum( 1 for f in in_diff_unpublished if f.get("id") not in eligible_ids and f.get("status", "open") == "open" and SEVERITY_ORDER.get(f.get("severity", "low"), 0) < severity_rank ) inline_comments: list[dict[str, Any]] = [] eligible_with_payload: list[tuple[dict[str, Any], dict[str, Any]]] = [] for finding in eligible: payload = render_inline_comment_payload(finding) if payload is None: continue inline_comments.append(payload) eligible_with_payload.append((dict(finding), payload)) # With nothing new to surface, skip the "no issues found" summary if Open # SWE has already reviewed this PR — the user already saw the previous # result, and posting another summary on every push is noise. We can't rely # on the static re_review flag alone: a push that lands mid-run is delivered # as a queued message into the still-running first-review run, whose # configurable still says re_review=False, so that path would post a # duplicate "No issues found". Key off the actual PR state (an existing Open # SWE review summary) instead. Still resolve threads for findings that just # moved to resolved, and advance last_reviewed_sha so subsequent pushes # don't redo the same diff. # A pending verdict must always reach GitHub — "approve if clean" with zero # findings still posts an APPROVE review — so the skip only applies to # plain comment publishes. if ( not inline_comments and not verdict_attempted and await _open_swe_already_reviewed( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, token=token, is_re_review=is_re_review, ) ): resolved_thread_count = await _resolve_threads_for_resolved_findings( owner=owner, repo=repo, pr_number=pr_number, token=token, findings=findings, ) await set_reviewer_thread_metadata(thread_id, last_reviewed_sha=head_sha) await clear_review_started_comment(thread_id=thread_id, owner=owner, repo=repo, token=token) skip_result: dict[str, Any] = { "success": True, "review_id": None, "surfaced_count": 0, "hidden_count": max(len(open_unpublished), 0), "resolved_thread_count": resolved_thread_count, "skipped_empty_re_review": True, "blocking_finding_count": sum( 1 for finding in findings if _finding_blocks_verdict(finding) ), "verdict_authorization": verdict_authorization, } conclusion, check_title, check_summary = review_check_conclusion(skip_result) await settle_review_check_run( thread_id=thread_id, owner=owner, repo=repo, token=token, conclusion=conclusion, title=check_title, summary=check_summary, ) return skip_result review_body = render_review_body( pr_number=pr_number, surfaced_count=len(inline_comments), trace_url=review_trace_url, ui_url=review_ui_url, additional_findings_count=additional_findings_count, ) review_response, event, verdict_ignored_reason, findings = await _post_review_guarded( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, head_sha=head_sha, body=review_body, inline_comments=inline_comments, token=token, verdict=verdict, verdict_authorization=verdict_authorization, verdict_requester=verdict_requester, reviewed_head_sha=reviewed_head_sha, ) # If GitHub rejected the batch because one or more inline comments anchor # to a file/line that's not in the PR diff, drop just those findings and # retry once. Returning the bare 422 to the agent only invites it to # retry publish_review with byte-identical args until findings drain. unresolvable_findings: list[str] = [] if ( isinstance(review_response, dict) and review_response.get("_error_kind") == "unresolved_anchor" ): valid_with_payload, dropped_ids = await _filter_against_pr_diff( eligible_with_payload, owner=owner, repo=repo, pr_number=pr_number, token=token, ) if dropped_ids and valid_with_payload: retry_inline = [p for _, p in valid_with_payload] retry_body = render_review_body( pr_number=pr_number, surfaced_count=len(retry_inline), trace_url=review_trace_url, ui_url=review_ui_url, additional_findings_count=additional_findings_count, ) retry_response, event, verdict_ignored_reason, findings = await _post_review_guarded( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, head_sha=head_sha, body=retry_body, inline_comments=retry_inline, token=token, verdict=verdict, verdict_authorization=verdict_authorization, verdict_requester=verdict_requester, reviewed_head_sha=reviewed_head_sha, ) if isinstance(retry_response, dict) and "_error" not in retry_response: review_response = retry_response review_body = retry_body inline_comments = retry_inline eligible_with_payload = valid_with_payload unresolvable_findings = dropped_ids else: retry_error = ( retry_response.get("_error", "unknown error") if isinstance(retry_response, dict) else "no response" ) return { "success": False, "error": f"Failed to POST PR review: {retry_error}", "unresolvable_findings": dropped_ids, "hint": ( "Call update_finding(status='resolved') on these ids " "or fix their file/line before retrying." ), } elif event != "COMMENT": # A verdict is pending but every inline comment anchors outside the # diff. GitHub accepts a bodied review with zero inline comments, so # post the authorized verdict rather than dropping it — the verdict # must land even when the findings can't be anchored. verdict_only_body = render_review_body( pr_number=pr_number, surfaced_count=0, trace_url=review_trace_url, ui_url=review_ui_url, additional_findings_count=additional_findings_count, ) ( verdict_only_response, event, verdict_ignored_reason, findings, ) = await _post_review_guarded( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, head_sha=head_sha, body=verdict_only_body, inline_comments=[], token=token, verdict=verdict, verdict_authorization=verdict_authorization, verdict_requester=verdict_requester, reviewed_head_sha=reviewed_head_sha, ) if isinstance(verdict_only_response, dict) and "_error" not in verdict_only_response: review_response = verdict_only_response review_body = verdict_only_body inline_comments = [] eligible_with_payload = [] unresolvable_findings = dropped_ids else: verdict_error = ( verdict_only_response.get("_error", "unknown error") if isinstance(verdict_only_response, dict) else "no response" ) return { "success": False, "error": f"Failed to POST PR review: {verdict_error}", "unresolvable_findings": dropped_ids, } else: # Either nothing to drop (no diff_line_set available, so we can't # tell which findings are bad) or everything would be dropped. # Either way, do not retry — surface the structural signal so the # agent stops retrying with the same args. return { "success": False, "error": f"Failed to POST PR review: {review_response['_error']}", "unresolvable_findings": dropped_ids, "hint": ( "Call update_finding(status='resolved') on these ids " "or fix their file/line before retrying." ), } if isinstance(review_response, dict) and "_error" in review_response: return { "success": False, "error": f"Failed to POST PR review: {review_response['_error']}", } if review_response is None: # Defensive guard: with the upstream change this should never happen, # but keep a clear signal if it does so the agent doesn't retry blindly. return { "success": False, "error": "Failed to POST PR review: no response from GitHub", } review_id = review_response.get("id") if isinstance(review_response, dict) else None # Trust GitHub's recorded review state, not the event we asked for. returned_state = review_response.get("state") if isinstance(review_response, dict) else None recorded_state = returned_state.upper() if isinstance(returned_state, str) else "" expected_state = _EVENT_TO_STATE.get(event) verdict_submitted = bool( verdict_attempted and expected_state and recorded_state == expected_state and review_id is not None ) if verdict_attempted and not verdict_submitted and verdict_ignored_reason is None: verdict_ignored_reason = "github_state_mismatch" body_update_failed = False if verdict_submitted and isinstance(review_id, int) and recorded_state: body_updated = await update_pull_request_review_body( owner=owner, repo=repo, pr_number=pr_number, review_id=review_id, body=_decorate_recorded_review_body( review_body, recorded_state=recorded_state, verdict=verdict, verdict_requester=verdict_requester, verdict_authorization=verdict_authorization, verdict_submitted=verdict_submitted, verdict_ignored_reason=verdict_ignored_reason, ), token=token, ) body_update_failed = body_updated is not True await _reconcile_last_verdict( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, token=token, event=event, review_id=review_id if isinstance(review_id, int) else None, head_sha=head_sha, surfaced_count=len(inline_comments), verdict_submitted=verdict_submitted, ) if review_id is not None and inline_comments: # Record the GitHub review id AND inline comment ids in a single # findings write. Previously these were three separate read-replace # cycles (out-of-diff review id, inline review id, comment ids); each # extra write widened the window where a crash could leave findings # half-stamped — surfaced on GitHub but with no recorded comment id, so # a later resolve-on-fix couldn't find the thread. comment_records: list[dict[str, Any]] = [] if inline_comments: comment_records = await fetch_review_comments( owner=owner, repo=repo, pr_number=pr_number, review_id=review_id, token=token, ) if langgraph_run_id is None: metadata = await get_thread_metadata(thread_id) current_run_id = metadata.get("current_reviewer_run_id") if isinstance(current_run_id, str) and current_run_id: langgraph_run_id = current_run_id await _record_review_publication( thread_id=thread_id, review_id=review_id, inline_with_payload=eligible_with_payload, comment_records=comment_records, langgraph_run_id=langgraph_run_id, ) if review_id is not None and inline_comments: current_findings = await list_findings_async(thread_id) if _missing_comment_ids_for_published_findings(current_findings, eligible_with_payload): await _backfill_findings_from_pr_threads( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, token=token, ) await _store_thread_ids_on_findings( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, token=token, ) resolved_thread_count = await _resolve_threads_for_resolved_findings( owner=owner, repo=repo, pr_number=pr_number, token=token, findings=await list_findings_async(thread_id), ) if not is_re_review: await _maybe_post_slack_completion_reply( thread_id=thread_id, owner=owner, repo=repo, pr_number=pr_number, review_id=review_id, surfaced_count=len(inline_comments), ) await set_reviewer_thread_metadata(thread_id, last_reviewed_sha=head_sha) await clear_review_started_comment(thread_id=thread_id, owner=owner, repo=repo, token=token) result: dict[str, Any] = { "success": True, "review_id": review_id, "surfaced_count": len(inline_comments), "hidden_count": max(len(open_unpublished) - len(inline_comments), 0), "resolved_thread_count": resolved_thread_count, "blocking_finding_count": sum( 1 for finding in findings if _finding_blocks_verdict(finding) ), "verdict_authorization": verdict_authorization, } if recorded_state: result["review_state"] = recorded_state if verdict_submitted: result["verdict_submitted"] = True result["verdict_event"] = event elif verdict_ignored_reason: result["verdict_submitted"] = False result["verdict_ignored"] = True result["verdict_ignored_reason"] = verdict_ignored_reason if body_update_failed: result["body_update_failed"] = True result["body_update_message"] = ( "GitHub recorded the review state, but updating the review body failed. " "The original body remains neutral and does not claim a verdict." ) if unresolvable_findings: result["unresolvable_findings"] = unresolvable_findings result["hint"] = ( "Some findings had anchors not in the PR diff; " "call update_finding to fix or resolve them." ) conclusion, check_title, check_summary = review_check_conclusion(result) await settle_review_check_run( thread_id=thread_id, owner=owner, repo=repo, token=token, conclusion=conclusion, title=check_title, summary=check_summary, ) return result def _finding_blocks_verdict(finding: Finding) -> bool: status = finding.get("status", "open") if status in {"open", "needs_reassessment"}: return True interactions = finding.get("interactions") if not isinstance(interactions, list) or not interactions: return False latest = interactions[-1] return isinstance(latest, dict) and latest.get("needs_reassessment") is True async def _post_review_guarded( *, thread_id: str, owner: str, repo: str, pr_number: int, head_sha: str, body: str, inline_comments: list[dict[str, Any]], token: str, verdict: str | None, verdict_authorization: str, verdict_requester: str, reviewed_head_sha: str, ) -> tuple[dict[str, Any] | None, str, str | None, list[Finding]]: if verdict is None: response = await post_pull_request_review( owner=owner, repo=repo, pr_number=pr_number, head_sha=head_sha, body=body, inline_comments=inline_comments, token=token, event="COMMENT", ) return response, "COMMENT", None, await list_findings_async(thread_id) async with _finding_mutation_lock(thread_id): metadata = await _get_thread_metadata_strict(thread_id) findings = _coerce_findings_list(metadata.get("findings")) submitted_verdict = verdict ignored_reason: str | None = None post_head_sha = head_sha if verdict_authorization == "none": submitted_verdict = None ignored_reason = "verdict_not_requested" elif verdict_authorization == "consistent": open_count = sum(1 for finding in findings if _finding_blocks_verdict(finding)) if verdict == "approve" and open_count: submitted_verdict = None ignored_reason = "approve_with_open_findings" elif verdict == "request_changes" and not open_count: submitted_verdict = None ignored_reason = "request_changes_without_open_findings" if submitted_verdict is not None: pr_author = _pr_author_from_thread(metadata) bot_logins = {login.casefold() for login in INTERNAL_BOT_LOGINS} if not pr_author: submitted_verdict = None ignored_reason = "author_unknown" elif pr_author.casefold() in bot_logins: submitted_verdict = None ignored_reason = "self_review" if submitted_verdict is not None and head_sha != reviewed_head_sha: submitted_verdict = None ignored_reason = "head_moved" if submitted_verdict is not None: live_head_sha = await fetch_pull_request_head_sha( owner=owner, repo=repo, pr_number=pr_number, token=token, ) if live_head_sha is None: submitted_verdict = None ignored_reason = "head_check_failed" elif live_head_sha != reviewed_head_sha: submitted_verdict = None ignored_reason = "head_moved" post_head_sha = live_head_sha event = _VERDICT_EVENTS.get(submitted_verdict or "", "COMMENT") decorated_body = _decorate_review_body( body, verdict=submitted_verdict, verdict_authorization=verdict_authorization, verdict_requester=verdict_requester, verdict_ignored_reason=ignored_reason, ) response = await post_pull_request_review( owner=owner, repo=repo, pr_number=pr_number, head_sha=post_head_sha, body=decorated_body, inline_comments=inline_comments, token=token, event=event, ) return response, event, ignored_reason, findings async def _open_swe_already_reviewed( *, thread_id: str, owner: str, repo: str, pr_number: int, token: str, is_re_review: bool, ) -> bool: """Decide whether to suppress a duplicate empty "no issues found" summary. Suppress only when we are *certain* a prior Open SWE review exists, so a transient GitHub failure never causes a double-post: - ``is_re_review`` is a durable signal (the dispatching webhook set it from the persisted ``last_reviewed_sha``), so trust it outright. - Otherwise consult durable reviewer state (``last_reviewed_sha`` on thread metadata): a non-empty value means this thread already published once. - Only as a last resort hit the GitHub reviews API. That call is tri-state: ``True``/``False`` are authoritative, but ``None`` means "unknown" (pagination or the request failed). On ``None`` we do NOT suppress — a possible duplicate summary is better than silently swallowing the only review the user will ever see, and re-posting is the safe failure mode. """ if is_re_review: return True metadata = await get_thread_metadata(thread_id) if get_thread_last_reviewed_sha(metadata): return True exists = await open_swe_review_exists(owner=owner, repo=repo, pr_number=pr_number, token=token) return exists is True def _pr_author_from_thread(metadata: dict[str, Any]) -> str: pr_meta = get_thread_pr_meta(metadata) if pr_meta is None: return "" author = pr_meta.get("author") return author if isinstance(author, str) else "" def _decorate_review_body( body: str, *, verdict: str | None, verdict_authorization: str, verdict_requester: str, verdict_ignored_reason: str | None, ) -> str: """Append verdict attribution / downgrade context to the review body.""" if verdict is not None: if verdict_authorization == "consistent": return ( f"{body}\n\nAutomatic verdict evaluation (`{verdict}`) pending " "based on authoritative finding state." ) requester = f"@{verdict_requester}" if verdict_requester else "an explicit human request" return f"{body}\n\nVerdict (`{verdict}`) pending for {requester}." if verdict_ignored_reason: reason = verdict_ignored_reason.replace("_", " ") return f"{body}\n\n> Verdict withheld ({reason}). Published as a comment review." return body def _decorate_recorded_review_body( body: str, *, recorded_state: str, verdict: str | None, verdict_requester: str, verdict_authorization: str, verdict_submitted: bool, verdict_ignored_reason: str | None, ) -> str: if verdict_submitted and verdict is not None: if verdict_authorization == "consistent": return ( f"{body}\n\nReview outcome: **{recorded_state}**. " f"Automatic verdict (`{verdict}`) recorded based on authoritative finding state." ) requester = f"@{verdict_requester}" if verdict_requester else "an explicit human request" return ( f"{body}\n\nReview outcome: **{recorded_state}**. " f"Verdict (`{verdict}`) recorded for {requester}." ) reason = (verdict_ignored_reason or "github_state_mismatch").replace("_", " ") if verdict_authorization == "consistent": return ( f"{body}\n\nReview outcome: **{recorded_state}**. " f"The automatic verdict evaluation was withheld ({reason})." ) return ( f"{body}\n\nReview outcome: **{recorded_state}**. " f"The requested verdict was withheld ({reason})." ) async def _reconcile_last_verdict( *, thread_id: str, owner: str, repo: str, pr_number: int, token: str, event: str, review_id: int | None, head_sha: str, surfaced_count: int, verdict_submitted: bool, ) -> None: """Record a submitted verdict and dismiss a stale prior APPROVE. A previously recorded APPROVE goes stale the moment a later publish surfaces new findings without re-approving — leave it standing and the PR keeps an approved state that no longer reflects the reviewer's opinion. Entirely best-effort: failures are logged and never block the publish. """ if not verdict_submitted and surfaced_count == 0: return try: metadata = await get_thread_metadata(thread_id) last_verdict = metadata.get("last_verdict") stale_approval_id: int | None = None if ( isinstance(last_verdict, dict) and last_verdict.get("event") == "APPROVE" and isinstance(last_verdict.get("review_id"), int) and event != "APPROVE" and surfaced_count > 0 ): stale_approval_id = last_verdict["review_id"] await dismiss_pull_request_review( owner=owner, repo=repo, pr_number=pr_number, review_id=stale_approval_id, message=( "Dismissing stale approval: a later Open SWE review surfaced new " "findings on this pull request." ), token=token, ) if verdict_submitted and review_id is not None: await set_reviewer_thread_metadata( thread_id, extra={"last_verdict": {"event": event, "review_id": review_id, "sha": head_sha}}, ) elif stale_approval_id is not None: await set_reviewer_thread_metadata(thread_id, extra={"last_verdict": None}) except Exception: # noqa: BLE001 — advisory bookkeeping must not fail the publish logger.exception("Failed to reconcile last_verdict for %s/%s#%s", owner, repo, pr_number) def _has_publication_identity(finding: Finding) -> bool: return isinstance(finding.get("github_review_comment_id"), int) or isinstance( finding.get("github_review_id"), int ) def _int_list(value: Any) -> list[int]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, int)] def _str_list(value: Any) -> list[str]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, str) and item] def _comment_ids_for_finding(finding: dict[str, Any]) -> list[int]: comment_ids = _int_list(finding.get("github_review_comment_ids")) comment_id = finding.get("github_review_comment_id") if isinstance(comment_id, int) and comment_id not in comment_ids: comment_ids.insert(0, comment_id) return comment_ids def _thread_ids_for_finding(finding: dict[str, Any]) -> list[str]: thread_ids = _str_list(finding.get("github_review_thread_ids")) thread_id = finding.get("github_review_thread_id") if isinstance(thread_id, str) and thread_id and thread_id not in thread_ids: thread_ids.insert(0, thread_id) return thread_ids async def _backfill_findings_from_pr_threads( *, thread_id: str, owner: str, repo: str, pr_number: int, token: str, ) -> list[Finding]: findings = await list_findings_async(thread_id) if not findings: return findings review_threads = await fetch_pr_review_threads( owner=owner, repo=repo, pr_number=pr_number, token=token, ) if not review_threads: return findings return await reconcile_findings_with_review_threads(thread_id, review_threads) def _missing_comment_ids_for_published_findings( findings: list[Finding], eligible_with_payload: list[tuple[dict[str, Any], dict[str, Any]]], ) -> bool: finding_ids = { finding.get("id") for finding, _payload in eligible_with_payload if isinstance(finding.get("id"), str) } for finding in findings: if finding.get("id") in finding_ids and not isinstance( finding.get("github_review_comment_id"), int ): return True return False def _apply_review_id( findings: list[Finding], *, finding_ids: set[str], review_id: int, ) -> bool: updated = False for finding in findings: if finding.get("id") in finding_ids and finding.get("github_review_id") != review_id: finding["github_review_id"] = review_id if isinstance(finding.get("id"), str): surface = _coerce_surface(finding, str(finding["id"])) surface["github_review_id"] = review_id finding["surface"] = surface updated = True return updated def _apply_comment_ids( findings: list[Finding], *, comment_id_by_finding_id: dict[str, int], langgraph_run_id: str | None, ) -> bool: updated = False for finding in findings: finding_id = finding.get("id") if not isinstance(finding_id, str): continue comment_id = comment_id_by_finding_id.get(finding_id) if comment_id is None: continue finding["github_review_comment_id"] = comment_id comment_ids = _int_list(finding.get("github_review_comment_ids")) if comment_id not in comment_ids: comment_ids.append(comment_id) finding["github_review_comment_ids"] = comment_ids surface = _coerce_surface(finding, finding_id) surface["state"] = "surfaced" surface["github_review_comment_id"] = comment_id surface["severity_threshold_at_publish"] = finding.get("severity") surface["surfaced_at_sha"] = finding.get("last_confirmed_sha") or finding.get( "first_seen_sha" ) finding["surface"] = surface if langgraph_run_id: finding["github_review_run_id"] = langgraph_run_id updated = True return updated def _comment_id_by_finding_id( eligible_with_payload: list[tuple[dict[str, Any], dict[str, Any]]], comment_records: list[dict[str, Any]], ) -> dict[str, int]: """Map each surfaced finding id to its GitHub comment id via the marker. The embedded Open SWE marker is the *only* source of truth. Every comment this reviewer posts carries a ```` marker keyed by finding id (see ``render_inline_comment_body``), so the match is exact. The old ``(path, line, body)`` fallback collided whenever two findings shared a path/line/body — it cached the same comment id on both, which corrupts resolve-on-fix (resolving one would target the wrong thread). Findings whose comment lacks a parseable marker are left out here; ``_backfill_findings_from_pr_threads`` recovers them via the same marker against the PR's review threads. """ by_marker_id: dict[str, int] = {} for record in comment_records: body = record.get("body", "") comment_id = record.get("id") if isinstance(body, str) and isinstance(comment_id, int): marker = parse_review_comment_marker(body) if marker is not None: by_marker_id[marker["id"]] = comment_id out: dict[str, int] = {} for finding_snapshot, _payload in eligible_with_payload: finding_id = finding_snapshot.get("id") if isinstance(finding_id, str) and finding_id in by_marker_id: out[finding_id] = by_marker_id[finding_id] return out async def _record_review_publication( *, thread_id: str, review_id: int, inline_with_payload: list[tuple[dict[str, Any], dict[str, Any]]], comment_records: list[dict[str, Any]], langgraph_run_id: str | None, ) -> None: """Stamp the review id and inline comment ids onto findings in one write. Collapsing the review-id and comment-id updates into a single read-modify-write keeps publication identity atomic: a finding is never persisted carrying a review id without also carrying whatever comment id GitHub returned for it in the same record. """ review_finding_ids = { finding.get("id") for finding, _payload in inline_with_payload if isinstance(finding.get("id"), str) } comment_id_by_finding_id = _comment_id_by_finding_id(inline_with_payload, comment_records) latest = await list_findings_async(thread_id) changed = _apply_review_id( latest, finding_ids={fid for fid in review_finding_ids if isinstance(fid, str)}, review_id=review_id, ) changed = ( _apply_comment_ids( latest, comment_id_by_finding_id=comment_id_by_finding_id, langgraph_run_id=langgraph_run_id, ) or changed ) if changed: await replace_findings(thread_id, latest) async def _resolve_diff_line_set( *, owner: str, repo: str, pr_number: int, token: str, ) -> dict[str, set[int]] | None: """Return the new-side line set for the PR diff, fetching it if needed. Reviewer runs clear ``configurable['diff_line_set']`` before the agent starts (so ``add_finding`` trusts the agent's anchors), which means the publish-time retry path can't rely on it being populated. Fetch the PR's unified diff from the GitHub REST API and recompute the line set on the fly. Returns ``None`` if the fetch fails — caller treats that as "we can't tell which finding is bad, don't retry blindly". """ config = get_config() configurable = config.get("configurable", {}) if isinstance(config, dict) else {} cached = configurable.get("diff_line_set") if isinstance(configurable, dict) else None if isinstance(cached, dict): return cached diff_text = await fetch_pr_diff(owner=owner, repo=repo, pr_number=pr_number, token=token) if diff_text is None: return None return compute_diff_line_set(diff_text) async def _filter_against_pr_diff( eligible_with_payload: list[tuple[dict[str, Any], dict[str, Any]]], *, owner: str, repo: str, pr_number: int, token: str, ) -> tuple[list[tuple[dict[str, Any], dict[str, Any]]], list[str]]: """Drop findings whose path/line range is not in the current PR diff. Returns ``(valid_with_payload, dropped_finding_ids)``. When the diff cannot be resolved (fetch failed and no cached set), we return everything unchanged and an empty drop list — the caller will then surface the original error rather than retry blindly. """ diff_line_set = await _resolve_diff_line_set( owner=owner, repo=repo, pr_number=pr_number, token=token ) if diff_line_set is None: return list(eligible_with_payload), [] valid: list[tuple[dict[str, Any], dict[str, Any]]] = [] dropped: list[str] = [] for finding, payload in eligible_with_payload: path = payload.get("path") # Prefer the finding's recorded range; fall back to the payload line. start_line = finding.get("start_line") end_line = finding.get("end_line") if end_line is None: payload_line = payload.get("line") if isinstance(payload_line, int): end_line = payload_line if start_line is None: start_line = payload_line side = finding.get("side") if finding.get("side") in {"LEFT", "RIGHT"} else "RIGHT" if isinstance(path, str) and is_range_in_diff( diff_line_set, path, start_line, end_line, side=side ): valid.append((finding, payload)) else: finding_id = finding.get("id") if isinstance(finding_id, str): dropped.append(finding_id) return valid, dropped async def _maybe_post_slack_completion_reply( *, thread_id: str, owner: str, repo: str, pr_number: int, review_id: int | None, surfaced_count: int, ) -> None: """Post a one-line completion summary to the Slack thread that started this review. Only fires for first reviews (gated by the caller). No-op if the reviewer thread has no ``slack_thread`` metadata — i.e. the review wasn't started from Slack. """ metadata = await get_thread_metadata(thread_id) slack_ref = get_thread_slack_ref(metadata) if slack_ref is None: return if surfaced_count == 0: headline = "*Open SWE Review*: No issues found." else: issue_word = "issue" if surfaced_count == 1 else "issues" headline = f"*Open SWE Review* found {surfaced_count} potential {issue_word}." review_url = f"https://github.com/{owner}/{repo}/pull/{pr_number}" if isinstance(review_id, int): review_url = f"{review_url}#pullrequestreview-{review_id}" text = f"{headline} <{review_url}|View review>" await post_slack_thread_reply(slack_ref["channel_id"], slack_ref["thread_ts"], text) async def _store_thread_ids_on_findings( *, thread_id: str, owner: str, repo: str, pr_number: int, token: str, ) -> None: findings = await list_findings_async(thread_id) comment_ids_by_finding_id: dict[str, list[int]] = {} for finding in findings: finding_id = finding.get("id") comment_ids = _comment_ids_for_finding(finding) if isinstance(finding_id, str) and comment_ids and not _thread_ids_for_finding(finding): comment_ids_by_finding_id[finding_id] = comment_ids if not comment_ids_by_finding_id: return threads = await fetch_pr_review_threads( owner=owner, repo=repo, pr_number=pr_number, token=token, ) thread_id_by_comment_id: dict[int, str] = {} for thread in threads: github_thread_id = thread.get("id") if not isinstance(github_thread_id, str) or not github_thread_id: continue for comment in thread.get("comments") or []: if not isinstance(comment, dict): continue comment_id = comment.get("id") if isinstance(comment_id, int): thread_id_by_comment_id[comment_id] = github_thread_id updated = False for finding in findings: finding_id = finding.get("id") if not isinstance(finding_id, str): continue thread_ids = _thread_ids_for_finding(finding) for comment_id in comment_ids_by_finding_id.get(finding_id, []): github_thread_id = thread_id_by_comment_id.get(comment_id) if not github_thread_id: continue if not isinstance(finding.get("github_review_thread_id"), str): finding["github_review_thread_id"] = github_thread_id updated = True if github_thread_id not in thread_ids: thread_ids.append(github_thread_id) finding["github_review_thread_ids"] = thread_ids updated = True surface = _coerce_surface(finding, finding_id) surface["state"] = "surfaced" surface["github_review_thread_id"] = github_thread_id finding["surface"] = surface updated = True if updated: await replace_findings(thread_id, findings) async def _resolve_threads_for_resolved_findings( *, owner: str, repo: str, pr_number: int, token: str, findings: list[dict[str, Any]], ) -> int: """Resolve GitHub review threads for findings that just transitioned to resolved. Posts a resolution comment to the thread, then resolves it. Multiple threads can exist when an earlier run duplicated a comment before publication identity was backfilled. """ resolved_count = 0 mutated = False for finding in findings: status = finding.get("status") if status not in {"resolved", "dismissed"}: continue thread_node_ids = _thread_ids_for_finding(finding) comment_ids = _comment_ids_for_finding(finding) for comment_id in comment_ids: thread_node_id = await fetch_review_thread_id_for_comment( owner=owner, repo=repo, pr_number=pr_number, review_comment_id=comment_id, token=token, ) if thread_node_id and thread_node_id not in thread_node_ids: thread_node_ids.append(thread_node_id) if not thread_node_ids: continue resolved_thread_ids = _str_list(finding.get("github_resolved_thread_ids")) posted_resolution_comment_ids = _int_list( finding.get("github_posted_resolution_comment_ids") ) for idx, thread_node_id in enumerate(thread_node_ids): if thread_node_id in resolved_thread_ids: continue primary_comment_id = comment_ids[idx] if idx < len(comment_ids) else None resolution_body = render_resolution_comment(finding, status) if ( primary_comment_id and primary_comment_id not in posted_resolution_comment_ids and resolution_body is not None ): reply_response = await reply_to_review_comment( owner=owner, repo=repo, pr_number=pr_number, review_comment_id=primary_comment_id, body=resolution_body, token=token, ) if reply_response and isinstance(reply_response.get("id"), int): posted_resolution_comment_ids.append(primary_comment_id) mutated = True ok = await resolve_review_thread(thread_node_id=thread_node_id, token=token) if ok: resolved_thread_ids.append(thread_node_id) resolved_count += 1 mutated = True if resolved_thread_ids: finding["github_resolved_thread_ids"] = resolved_thread_ids if posted_resolution_comment_ids: finding["github_posted_resolution_comment_ids"] = posted_resolution_comment_ids if thread_node_ids: finding["github_review_thread_ids"] = thread_node_ids if not isinstance(finding.get("github_review_thread_id"), str): finding["github_review_thread_id"] = thread_node_ids[0] if thread_node_ids and all( thread_id in resolved_thread_ids for thread_id in thread_node_ids ): finding["github_thread_resolved"] = True if isinstance(finding.get("id"), str): surface = _coerce_surface(finding, str(finding["id"])) surface["state"] = "resolved" if thread_node_ids: surface["github_review_thread_id"] = thread_node_ids[0] finding["surface"] = surface if mutated: thread_id = get_thread_id_from_runtime() await replace_findings(thread_id, findings) return resolved_count def _current_run_id(config: dict[str, Any]) -> str | None: candidates = [config.get("run_id")] configurable = config.get("configurable") if isinstance(configurable, dict): candidates.append(configurable.get("run_id")) for candidate in candidates: if isinstance(candidate, str) and candidate: return candidate return None