fix: make review publish idempotent (partial-failure recovery) (#1477)

Publish had partial-failure windows that double-posted summaries or
corrupted findings state. This hardens the recovery paths:

- open_swe_review_exists is now tri-state (True/False/None). On a
  pagination/API failure it returns None ("unknown") instead of False,
  and the empty-summary dedup keys off the durable last_reviewed_sha
  before consulting GitHub, so a transient failure never double-posts a
  "no issues found" summary.
- Comment-id backfill matches strictly on the embedded open-swe marker;
  the colliding (path, line, body) fallback is gone, so similar findings
  no longer share a comment id and break resolve-on-fix.
- Review-id and comment-id stamping collapse into one guarded
  read-modify-write (re-reads latest before writing), removing the
  half-stamped intermediate states the prior multi-write flow left open.
- New mutate_findings primitive centralizes read-modify-write so finding
  updates operate on the freshest persisted list and skip no-op writes.

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

agent/reviewer_findings.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 import logging
 import uuid
+from collections.abc import Callable
 from typing import Any, Literal, TypedDict, cast
 
 from langgraph.config import get_config
@@ -331,16 +332,41 @@ async def get_finding(thread_id: str, finding_id: str) -> Finding | None:
 
 
 async def replace_findings(thread_id: str, findings: list[Finding]) -> None:
-    """Overwrite the findings list on a thread's metadata."""
+    """Overwrite the findings list on a thread's metadata.
+
+    Prefer :func:`mutate_findings` for read-modify-write updates: it re-reads the
+    freshest persisted list immediately before mutating, which shrinks the
+    lost-update window a blind overwrite here leaves open.
+    """
     client = get_client()
     await client.threads.update(thread_id=thread_id, metadata={"findings": findings})
 
 
+async def mutate_findings(
+    thread_id: str,
+    mutator: Callable[[list[Finding]], bool],
+) -> list[Finding]:
+    """Read the latest findings, apply ``mutator`` in place, persist iff changed.
+
+    Centralizes the read-modify-write so every mutation operates on the freshest
+    persisted list rather than a stale in-memory snapshot. ``mutator`` edits the
+    list in place and returns ``True`` when it changed something; we only write
+    on change, so a no-op mutation never clobbers a concurrent update.
+    """
+    findings = await list_findings(thread_id)
+    if mutator(findings):
+        await replace_findings(thread_id, findings)
+    return findings
+
+
 async def append_finding(thread_id: str, finding: Finding) -> Finding:
     """Append a finding and persist the new list."""
-    findings = await list_findings(thread_id)
+
-    findings.append(finding)
+    def _append(findings: list[Finding]) -> bool:
-    await replace_findings(thread_id, findings)
+        findings.append(finding)
+        return True
+
+    await mutate_findings(thread_id, _append)
     return finding
 
 
@@ -350,17 +376,18 @@ async def update_finding_fields(
     updates: dict[str, Any],
 ) -> Finding | None:
     """Apply field updates to one finding by id and persist."""
-    findings = await list_findings(thread_id)
+    captured: dict[str, Finding] = {}
-    updated: Finding | None = None
+
-    for finding in findings:
+    def _apply(findings: list[Finding]) -> bool:
-        if finding.get("id") == finding_id:
+        for finding in findings:
-            finding.update(updates)
+            if finding.get("id") == finding_id:
-            updated = finding
+                finding.update(updates)
-            break
+                captured["finding"] = finding
-    if updated is None:
+                return True
-        return None
+        return False
-    await replace_findings(thread_id, findings)
+
-    return updated
+    await mutate_findings(thread_id, _apply)
+    return captured.get("finding")
 
 
 async def update_finding_surface(
@@ -369,21 +396,22 @@ async def update_finding_surface(
     updates: dict[str, Any],
 ) -> Finding | None:
     """Apply updates to the nested surface record and legacy GitHub fields."""
-    findings = await list_findings(thread_id)
+    captured: dict[str, Finding] = {}
-    updated: Finding | None = None
+
-    for finding in findings:
+    def _apply(findings: list[Finding]) -> bool:
-        if finding.get("id") != finding_id:
+        for finding in findings:
-            continue
+            if finding.get("id") != finding_id:
-        surface = _coerce_surface(finding, finding_id)
+                continue
-        surface.update(updates)
+            surface = _coerce_surface(finding, finding_id)
-        finding["surface"] = surface
+            surface.update(updates)
-        _sync_legacy_surface_fields(finding, surface)
+            finding["surface"] = surface
-        updated = finding
+            _sync_legacy_surface_fields(finding, surface)
-        break
+            captured["finding"] = finding
-    if updated is None:
+            return True
-        return None
+        return False
-    await replace_findings(thread_id, findings)
+
-    return updated
+    await mutate_findings(thread_id, _apply)
+    return captured.get("finding")
 
 
 async def append_finding_interaction(
@@ -392,29 +420,29 @@ async def append_finding_interaction(
     interaction: FindingInteraction,
 ) -> Finding | None:
     """Persist a GitHub review-thread interaction on one finding."""
-    findings = await list_findings(thread_id)
+    captured: dict[str, Finding] = {}
-    updated: Finding | None = None
+
-    for finding in findings:
+    def _apply(findings: list[Finding]) -> bool:
-        if finding.get("id") != finding_id:
+        for finding in findings:
-            continue
+            if finding.get("id") != finding_id:
-        interactions = finding.get("interactions")
+                continue
-        if not isinstance(interactions, list):
+            captured["finding"] = finding
-            interactions = []
+            interactions = finding.get("interactions")
-        github_comment_id = interaction.get("github_comment_id")
+            if not isinstance(interactions, list):
-        if isinstance(github_comment_id, int) and any(
+                interactions = []
-            isinstance(item, dict) and item.get("github_comment_id") == github_comment_id
+            github_comment_id = interaction.get("github_comment_id")
-            for item in interactions
+            if isinstance(github_comment_id, int) and any(
-        ):
+                isinstance(item, dict) and item.get("github_comment_id") == github_comment_id
-            updated = finding
+                for item in interactions
-            break
+            ):
-        interactions.append(interaction)
+                return False
-        finding["interactions"] = interactions
+            interactions.append(interaction)
-        updated = finding
+            finding["interactions"] = interactions
-        break
+            return True
-    if updated is None:
+        return False
-        return None
+
-    await replace_findings(thread_id, findings)
+    await mutate_findings(thread_id, _apply)
-    return updated
+    return captured.get("finding")
 
 
 def _coerce_surface(finding: Finding, finding_id: str) -> FindingSurface:

agent/reviewer_publish.py

@@ -431,8 +431,8 @@ async def open_swe_review_exists(
     repo: str,
     pr_number: int,
     token: str,
-) -> bool:
+) -> bool | None:
-    """Return True if Open SWE has already posted a review summary on this PR.
+    """Return whether Open SWE has already posted a review summary on this PR.
 
     Detected via the ``review_summary_marker`` that ``render_review_body``
     embeds in every Open SWE review body. The reviewer uses this to avoid
@@ -441,8 +441,14 @@ async def open_swe_review_exists(
     into the still-running first-review run, whose configurable still says
     ``re_review=False``, so the empty-review guard can't trust that flag alone.
 
-    On any API failure this returns False (fail open): the only consequence is
+    Tri-state on purpose:
-    a possible duplicate summary, never a suppressed first review.
+    - ``True``  — an Open SWE review summary was found.
+    - ``False`` — the full review list was paginated successfully and carried
+      no Open SWE summary.
+    - ``None``  — the answer is unknown because an API call (or a page partway
+      through pagination) failed. Callers must not treat ``None`` as "no review
+      exists": the old fail-open-as-False behaviour double-posted "no issues"
+      summaries whenever pagination failed mid-walk.
     """
     marker = review_summary_marker(pr_number)
     url = f"{_GITHUB_API_BASE}/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
@@ -460,9 +466,11 @@ async def open_swe_review_exists(
                     repo,
                     pr_number,
                 )
-                return False
+                return None
             data = response.json()
-            if not isinstance(data, list) or not data:
+            if not isinstance(data, list):
+                return None
+            if not data:
                 return False
             for review in data:
                 if isinstance(review, dict) and marker in (review.get("body") or ""):

agent/tools/publish_review.py

@@ -15,6 +15,7 @@ from ..reviewer_findings import (
     _coerce_surface,
     filter_findings_for_publish,
     get_thread_id_from_runtime,
+    get_thread_last_reviewed_sha,
     get_thread_metadata,
     get_thread_slack_ref,
     replace_findings,
@@ -281,11 +282,13 @@ async def _publish_review_async(
     if (
         not inline_comments
         and not eligible_out_of_diff
-        and (
+        and await _open_swe_already_reviewed(
-            is_re_review
+            thread_id=thread_id,
-            or await open_swe_review_exists(
+            owner=owner,
-                owner=owner, repo=repo, pr_number=pr_number, token=token
+            repo=repo,
-            )
+            pr_number=pr_number,
+            token=token,
+            is_re_review=is_re_review,
         )
     ):
         resolved_thread_count = await _resolve_threads_for_resolved_findings(
@@ -405,41 +408,37 @@ async def _publish_review_async(
         }
     review_id = review_response.get("id") if isinstance(review_response, dict) else None
 
-    if review_id is not None and eligible_out_of_diff:
+    if review_id is not None and (eligible_out_of_diff or inline_comments):
-        # Mark surfaced out-of-diff findings so re-review doesn't repost them.
+        # Record the GitHub review id AND inline comment ids in a single
-        await _store_review_id_on_findings(
+        # 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,
-            findings=findings,
-            eligible_with_payload=[(dict(f), {}) for f in eligible_out_of_diff],
             review_id=review_id,
-        )
+            out_of_diff_findings=eligible_out_of_diff,
-
+            inline_with_payload=eligible_with_payload,
-    if review_id is not None and inline_comments:
-        await _store_review_id_on_findings(
-            thread_id=thread_id,
-            findings=findings,
-            eligible_with_payload=eligible_with_payload,
-            review_id=review_id,
-        )
-        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 _store_comment_ids_on_findings(
-            thread_id=thread_id,
-            findings=findings,
-            eligible_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(
@@ -497,6 +496,39 @@ async def _publish_review_async(
     return result
 
 
+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 _has_publication_identity(finding: Finding) -> bool:
     return isinstance(finding.get("github_review_comment_id"), int) or isinstance(
         finding.get("github_review_id"), int
@@ -570,18 +602,12 @@ def _missing_comment_ids_for_published_findings(
     return False
 
 
-async def _store_review_id_on_findings(
+def _apply_review_id(
-    *,
-    thread_id: str,
     findings: list[Finding],
-    eligible_with_payload: list[tuple[dict[str, Any], dict[str, Any]]],
+    *,
+    finding_ids: set[str],
     review_id: int,
-) -> None:
+) -> bool:
-    finding_ids = {
-        finding.get("id")
-        for finding, _payload in eligible_with_payload
-        if isinstance(finding.get("id"), str)
-    }
     updated = False
     for finding in findings:
         if finding.get("id") in finding_ids and finding.get("github_review_id") != review_id:
@@ -591,8 +617,120 @@ async def _store_review_id_on_findings(
                 surface["github_review_id"] = review_id
                 finding["surface"] = surface
             updated = True
-    if updated:
+    return updated
-        await replace_findings(thread_id, findings)
+
+
+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,
+    out_of_diff_findings: list[Finding],
+    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 in out_of_diff_findings if isinstance(finding.get("id"), str)
+    }
+    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)
+
+    # Re-read the freshest persisted list right before mutating so this single
+    # write merges onto any update that landed since the snapshot the caller
+    # passed in, instead of blindly overwriting it.
+    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(
@@ -703,77 +841,6 @@ async def _maybe_post_slack_completion_reply(
     await post_slack_thread_reply(slack_ref["channel_id"], slack_ref["thread_ts"], text)
 
 
-async def _store_comment_ids_on_findings(
-    *,
-    thread_id: str,
-    findings: list[Finding],
-    eligible_with_payload: list[tuple[dict[str, Any], dict[str, Any]]],
-    comment_records: list[dict[str, Any]],
-    langgraph_run_id: str | None,
-) -> None:
-    """Match returned GitHub comment ids back to the findings that produced them.
-
-    Prefer the Open SWE marker because it survives body formatting changes;
-    fall back to ``(path, line, body)`` for older comments.
-    """
-    by_marker_id: dict[str, int] = {}
-    by_key: dict[tuple[str, int, str], int] = {}
-    for record in comment_records:
-        path = record.get("path")
-        line = record.get("line") or record.get("original_line")
-        body = record.get("body", "")
-        comment_id = record.get("id")
-        if (
-            isinstance(path, str)
-            and isinstance(line, int)
-            and isinstance(body, str)
-            and isinstance(comment_id, int)
-        ):
-            by_key[(path, line, body)] = comment_id
-            marker = parse_review_comment_marker(body)
-            if marker is not None:
-                by_marker_id[marker["id"]] = comment_id
-
-    updated = False
-    findings_by_id = {f.get("id"): f for f in findings}
-    for finding_snapshot, payload in eligible_with_payload:
-        finding_id = finding_snapshot.get("id")
-        comment_id = by_marker_id.get(finding_id) if isinstance(finding_id, str) else None
-        line_value = payload.get("line")
-        if comment_id is None and isinstance(line_value, int):
-            key = (
-                str(payload.get("path", "")),
-                line_value,
-                str(payload.get("body", "")),
-            )
-            comment_id = by_key.get(key)
-        if comment_id is None:
-            continue
-        finding = findings_by_id.get(finding_id)
-        if finding 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
-        if isinstance(finding_id, str):
-            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
-
-    if updated:
-        await replace_findings(thread_id, list(findings_by_id.values()))
-
-
 async def _store_thread_ids_on_findings(
     *,
     thread_id: str,

tests/test_reviewer_findings.py

@@ -13,6 +13,7 @@ from agent.reviewer_findings import (
     append_finding,
     filter_findings_for_publish,
     list_findings,
+    mutate_findings,
     new_finding,
     new_finding_id,
     replace_findings,
@@ -142,6 +143,42 @@ async def test_append_finding_appends_to_existing_list() -> None:
     assert [f["id"] for f in persisted] == ["f_a", "f_b"]
 
 
+@pytest.mark.asyncio
+async def test_mutate_findings_reads_latest_before_mutating() -> None:
+    """mutate_findings must operate on the freshest persisted list, not a stale
+    snapshot — the mutator receives whatever ``list_findings`` returns now."""
+    latest = [_f(id="f_fresh")]
+    fake_client = AsyncMock()
+    fake_client.threads.get.return_value = {"metadata": {"findings": latest}}
+
+    seen: list[str] = []
+
+    def _mutator(findings: list[Finding]) -> bool:
+        seen.extend(f["id"] for f in findings)
+        findings[0]["status"] = "resolved"
+        return True
+
+    with patch("agent.reviewer_findings.get_client", return_value=fake_client):
+        result = await mutate_findings("tid", _mutator)
+
+    assert seen == ["f_fresh"]
+    assert result[0]["status"] == "resolved"
+    fake_client.threads.update.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_mutate_findings_skips_write_when_unchanged() -> None:
+    """A no-op mutation must NOT persist, so it can never clobber a concurrent
+    update that landed between the read and a would-be write."""
+    fake_client = AsyncMock()
+    fake_client.threads.get.return_value = {"metadata": {"findings": [_f(id="f_a")]}}
+
+    with patch("agent.reviewer_findings.get_client", return_value=fake_client):
+        await mutate_findings("tid", lambda _findings: False)
+
+    fake_client.threads.update.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_update_finding_fields_mutates_only_target() -> None:
     a = _f(id="f_a", description="orig-a")

tests/test_reviewer_publish.py

@@ -852,6 +852,94 @@ async def test_publish_review_skips_review_existence_check_on_re_review() -> Non
     assert result["skipped_empty_re_review"] is True
 
 
+@pytest.mark.asyncio
+async def test_publish_review_dedup_keys_off_durable_last_reviewed_sha() -> None:
+    """A non-empty ``last_reviewed_sha`` on thread metadata means this thread
+    already published once. The empty-summary guard must trust that durable
+    signal and suppress without ever hitting the reviews API."""
+    from agent.tools.publish_review import _publish_review_async
+
+    review_exists = AsyncMock(return_value=False)
+    post_review = AsyncMock()
+
+    with (
+        patch("agent.tools.publish_review.get_thread_id_from_runtime", return_value="tid"),
+        patch("agent.tools.publish_review.list_findings_async", AsyncMock(return_value=[])),
+        patch(
+            "agent.tools.publish_review.get_thread_metadata",
+            AsyncMock(return_value={"last_reviewed_sha": "oldsha"}),
+        ),
+        patch("agent.tools.publish_review.open_swe_review_exists", review_exists),
+        patch("agent.tools.publish_review.post_pull_request_review", post_review),
+        patch(
+            "agent.tools.publish_review._resolve_threads_for_resolved_findings",
+            new_callable=AsyncMock,
+            return_value=0,
+        ),
+        patch("agent.tools.publish_review.set_reviewer_thread_metadata", new_callable=AsyncMock),
+    ):
+        result = await _publish_review_async(
+            owner="o",
+            repo="r",
+            pr_number=7,
+            head_sha="newsha",
+            token="t",
+            severity_threshold="medium",
+            cap=15,
+            is_re_review=False,
+        )
+
+    review_exists.assert_not_called()
+    post_review.assert_not_called()
+    assert result["skipped_empty_re_review"] is True
+
+
+@pytest.mark.asyncio
+async def test_publish_review_posts_summary_when_review_existence_unknown() -> None:
+    """When the reviews API can't answer (``open_swe_review_exists`` returns
+    ``None``) and there is no durable prior-review signal, the guard must NOT
+    suppress — re-posting the summary is the safe failure mode, never silently
+    swallowing the only review the user sees."""
+    from agent.tools.publish_review import _publish_review_async
+
+    review_exists = AsyncMock(return_value=None)
+    post_review = AsyncMock(return_value={"id": 321})
+
+    with (
+        patch("agent.tools.publish_review.get_thread_id_from_runtime", return_value="tid"),
+        patch("agent.tools.publish_review.list_findings_async", AsyncMock(return_value=[])),
+        patch(
+            "agent.tools.publish_review.get_thread_metadata",
+            AsyncMock(return_value={}),
+        ),
+        patch("agent.tools.publish_review.open_swe_review_exists", review_exists),
+        patch("agent.tools.publish_review.post_pull_request_review", post_review),
+        patch("agent.tools.publish_review.fetch_review_comments", AsyncMock(return_value=[])),
+        patch(
+            "agent.tools.publish_review._resolve_threads_for_resolved_findings",
+            new_callable=AsyncMock,
+            return_value=0,
+        ),
+        patch("agent.tools.publish_review.set_reviewer_thread_metadata", new_callable=AsyncMock),
+        patch("agent.tools.publish_review._maybe_post_slack_completion_reply", AsyncMock()),
+    ):
+        result = await _publish_review_async(
+            owner="o",
+            repo="r",
+            pr_number=7,
+            head_sha="newsha",
+            token="t",
+            severity_threshold="medium",
+            cap=15,
+            is_re_review=False,
+        )
+
+    review_exists.assert_awaited_once()
+    post_review.assert_awaited_once()
+    assert "skipped_empty_re_review" not in result
+    assert result["review_id"] == 321
+
+
 @pytest.mark.asyncio
 async def test_open_swe_review_exists_detects_summary_marker() -> None:
     response = MagicMock()
@@ -886,7 +974,10 @@ async def test_open_swe_review_exists_false_without_marker() -> None:
 
 
 @pytest.mark.asyncio
-async def test_open_swe_review_exists_fails_open_on_http_error() -> None:
+async def test_open_swe_review_exists_returns_none_on_http_error() -> None:
+    """A failed reviews API call is reported as ``None`` (unknown), never
+    ``False`` — the empty-summary dedup must not treat a transient failure as
+    "no prior review exists" and double-post."""
     import httpx
 
     client_cm = AsyncMock()
@@ -895,7 +986,7 @@ async def test_open_swe_review_exists_fails_open_on_http_error() -> None:
 
     with patch("agent.reviewer_publish.httpx.AsyncClient", return_value=client_cm):
         exists = await open_swe_review_exists(owner="o", repo="r", pr_number=7, token="t")
-    assert exists is False
+    assert exists is None
 
 
 @pytest.mark.asyncio
@@ -1152,6 +1243,118 @@ async def test_re_review_only_posts_current_head_unpublished_findings() -> None:
     assert new["github_review_comment_id"] == 303
 
 
+@pytest.mark.asyncio
+async def test_publish_review_matches_comment_ids_by_marker_not_path_line_body() -> None:
+    """Two findings on the same path/line with identical rendered bodies must
+    each get their OWN comment id, matched via the embedded marker. The old
+    ``(path, line, body)`` fallback collided here and cached one comment id on
+    both findings, breaking resolve-on-fix."""
+    from agent.tools.publish_review import _publish_review_async
+
+    f1 = _f(id="f_one", file="dup.py", start_line=5, end_line=5, description="same text")
+    f2 = _f(id="f_two", file="dup.py", start_line=5, end_line=5, description="same text")
+    findings = [f1, f2]
+    post_review = AsyncMock(return_value={"id": 700})
+    # GitHub returns one comment per finding; the only thing that distinguishes
+    # them is the marker embedded in each body.
+    fetch_comments = AsyncMock(
+        return_value=[
+            {"id": 901, "path": "dup.py", "line": 5, "body": render_inline_comment_body(f1)},
+            {"id": 902, "path": "dup.py", "line": 5, "body": render_inline_comment_body(f2)},
+        ]
+    )
+
+    with (
+        patch("agent.tools.publish_review.get_thread_id_from_runtime", return_value="tid"),
+        patch("agent.tools.publish_review.list_findings_async", AsyncMock(return_value=findings)),
+        patch("agent.tools.publish_review.post_pull_request_review", post_review),
+        patch("agent.tools.publish_review.fetch_review_comments", fetch_comments),
+        patch(
+            "agent.tools.publish_review._resolve_threads_for_resolved_findings",
+            new_callable=AsyncMock,
+            return_value=0,
+        ),
+        patch("agent.tools.publish_review._store_thread_ids_on_findings", new_callable=AsyncMock),
+        patch("agent.tools.publish_review.set_reviewer_thread_metadata", new_callable=AsyncMock),
+        patch("agent.tools.publish_review._maybe_post_slack_completion_reply", AsyncMock()),
+    ):
+        result = await _publish_review_async(
+            owner="o",
+            repo="r",
+            pr_number=7,
+            head_sha="sha",
+            token="t",
+            severity_threshold="medium",
+            cap=15,
+            is_re_review=False,
+        )
+
+    assert result["success"] is True
+    by_id = {f["id"]: f for f in findings}
+    assert by_id["f_one"]["github_review_comment_id"] == 901
+    assert by_id["f_two"]["github_review_comment_id"] == 902
+
+
+@pytest.mark.asyncio
+async def test_publish_review_records_review_id_and_comment_id_in_single_write() -> None:
+    """The post-publish bookkeeping stamps review id + comment id onto findings
+    in one ``replace_findings`` call, so a finding is never persisted with a
+    review id but no comment id."""
+    from agent.tools.publish_review import _publish_review_async
+
+    finding = _f(id="f_new", file="x.py", start_line=3, end_line=3)
+    findings = [finding]
+    post_review = AsyncMock(return_value={"id": 555})
+    fetch_comments = AsyncMock(
+        return_value=[
+            {"id": 808, "path": "x.py", "line": 3, "body": render_inline_comment_body(finding)},
+        ]
+    )
+    replace = AsyncMock()
+
+    with (
+        patch("agent.tools.publish_review.get_thread_id_from_runtime", return_value="tid"),
+        patch("agent.tools.publish_review.list_findings_async", AsyncMock(return_value=findings)),
+        patch("agent.tools.publish_review.replace_findings", replace),
+        patch("agent.tools.publish_review.post_pull_request_review", post_review),
+        patch("agent.tools.publish_review.fetch_review_comments", fetch_comments),
+        patch("agent.tools.publish_review._store_thread_ids_on_findings", new_callable=AsyncMock),
+        patch(
+            "agent.tools.publish_review._resolve_threads_for_resolved_findings",
+            new_callable=AsyncMock,
+            return_value=0,
+        ),
+        patch("agent.tools.publish_review.set_reviewer_thread_metadata", new_callable=AsyncMock),
+        patch("agent.tools.publish_review._maybe_post_slack_completion_reply", AsyncMock()),
+    ):
+        result = await _publish_review_async(
+            owner="o",
+            repo="r",
+            pr_number=7,
+            head_sha="sha",
+            token="t",
+            severity_threshold="medium",
+            cap=15,
+            is_re_review=False,
+        )
+
+    assert result["success"] is True
+    # Exactly one persisted snapshot carries both ids together — never a
+    # half-stamped intermediate state.
+    persisted_snapshots = [call.args[1] for call in replace.await_args_list]
+    assert any(
+        snap[0].get("github_review_id") == 555 and snap[0].get("github_review_comment_id") == 808
+        for snap in persisted_snapshots
+    )
+    assert all(
+        not (
+            snap[0].get("github_review_id") == 555
+            and snap[0].get("github_review_comment_id") is None
+        )
+        for snap in persisted_snapshots
+    )
+
+
 @pytest.mark.asyncio
 async def test_publish_review_posts_summary_when_no_findings() -> None:
     """An empty findings list must still post a review so the user sees feedback."""