open-swe/agent/tools/publish_review.py

"""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 —
        ``"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)
        conclusion, check_title, check_summary = review_check_conclusion(0)
        await settle_review_check_run(
            thread_id=thread_id,
            owner=owner,
            repo=repo,
            token=token,
            conclusion=conclusion,
            title=check_title,
            summary=check_summary,
        )
        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,
        }
        if verdict_ignored_reason:
            skip_result["verdict_submitted"] = False
            skip_result["verdict_ignored"] = True
            skip_result["verdict_ignored_reason"] = verdict_ignored_reason
        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"
    if verdict_attempted and isinstance(review_id, int) and recorded_state:
        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_submitted=verdict_submitted,
                verdict_ignored_reason=verdict_ignored_reason,
            ),
            token=token,
        )
    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)
    conclusion, check_title, check_summary = review_check_conclusion(len(inline_comments))
    await settle_review_check_run(
        thread_id=thread_id,
        owner=owner,
        repo=repo,
        token=token,
        conclusion=conclusion,
        title=check_title,
        summary=check_summary,
    )

    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,
    }
    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 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."
        )
    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 not None and 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_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_requester: str,
    verdict_ignored_reason: str | None,
) -> str:
    """Append verdict attribution / downgrade context to the review body."""
    if verdict is not None:
        requester = f"@{verdict_requester}" if verdict_requester else "the requester"
        return f"{body}\n\nVerdict (`{verdict}`) requested by {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_submitted: bool,
    verdict_ignored_reason: str | None,
) -> str:
    if verdict_submitted and verdict is not None:
        requester = f"@{verdict_requester}" if verdict_requester else "the requester"
        return (
            f"{body}\n\nReview outcome: **{recorded_state}**. "
            f"Verdict (`{verdict}`) recorded for {requester}."
        )
    reason = (verdict_ignored_reason or "github_state_mismatch").replace("_", " ")
    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 ``<!-- open-swe-review-comment {...} -->``
    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